Shulou(Shulou.com)06/02 Report--

This article mainly explains "what is the principle and function of RaftPeerSet in nacos". The content of the explanation is simple and clear, and it is easy to learn and understand. Please follow the editor's train of thought to study and learn "what is the principle and function of RaftPeerSet in nacos".

Order

This paper mainly studies the RaftPeerSet of nacos.

RaftPeerSet

Nacos-1.1.3/naming/src/main/java/com/alibaba/nacos/naming/consistency/persistent/raft/RaftPeerSet.java

@ Component@DependsOn ("serverListManager") public class RaftPeerSet implements ServerChangeListener, ApplicationContextAware {@ Autowired private ServerListManager serverListManager; private ApplicationContext applicationContext; private AtomicLong localTerm = new AtomicLong (0L); private RaftPeer leader = null; private Map peers = new HashMap (); private Set sites = new HashSet (); private boolean ready = false; public RaftPeerSet () {} @ Override public void setApplicationContext (ApplicationContext applicationContext) throws BeansException {this.applicationContext = applicationContext } @ PostConstruct public void init () {serverListManager.listen (this);} public RaftPeer getLeader () {if (STANDALONE_MODE) {return local ();} return leader;} public Set allSites () {return sites;} public boolean isReady () {return ready } public void remove (List servers) {for (String server: servers) {peers.remove (server);}} public RaftPeer update (RaftPeer peer) {peers.put (peer.ip, peer); return peer;} public boolean isLeader (String ip) {if (STANDALONE_MODE) {return true } if (leader = = null) {Loggers.RAFT.warn ("[IS LEADER] no leader is available now!"); return false;} return StringUtils.equals (leader.ip, ip);} public Set allServersIncludeMyself () {return peers.keySet ();} public Set allServersWithoutMySelf () {Set servers = new HashSet (peers.keySet ()) / / exclude myself servers.remove (local () .ip); return servers;} public Collection allPeers () {return peers.values ();} public int size () {return peers.size ();} public RaftPeer decideLeader (RaftPeer candidate) {peers.put (candidate.ip, candidate); SortedBag ips = new TreeBag (); int maxApproveCount = 0 String maxApprovePeer = null; for (RaftPeer peer: peers.values ()) {if (StringUtils.isEmpty (peer.voteFor)) {continue;} ips.add (peer.voteFor); if (ips.getCount (peer.voteFor) > maxApproveCount) {maxApproveCount = ips.getCount (peer.voteFor); maxApprovePeer = peer.voteFor }} if (maxApproveCount > = majorityCount ()) {RaftPeer peer = peers.get (maxApprovePeer); peer.state = RaftPeer.State.LEADER; if (! Objects.equals (leader, peer)) {leader = peer; applicationContext.publishEvent (new LeaderElectFinishedEvent (this, leader)) Loggers.RAFT.info ("{} has become the LEADER", leader.ip);} return leader;} public RaftPeer makeLeader (RaftPeer candidate) {if (! Objects.equals (leader, candidate)) {leader = candidate; applicationContext.publishEvent (new MakeLeaderEvent (this, leader)) Loggers.RAFT.info ("{} has become the LEADER, local: {}, leader: {}", leader.ip, JSON.toJSONString (local ()), JSON.toJSONString (leader));} for (final RaftPeer peer: peers.values ()) {Map params = new HashMap (1) If (! Objects.equals (peer, candidate) & & peer.state = = RaftPeer.State.LEADER) {try {String url = RaftCore.buildURL (peer.ip, RaftCore.API_GET_PEER) HttpClient.asyncHttpGet (url, null, params, new AsyncCompletionHandler () {@ Override public Integer onCompleted (Response response) throws Exception {if (response.getStatusCode ()! = HttpURLConnection.HTTP_OK) {Loggers.RAFT.error ("[NACOS-RAFT] get peer failed: {}) Peer: {} ", response.getResponseBody (), peer.ip) Peer.state = RaftPeer.State.FOLLOWER; return 1;} update (JSON.parseObject (response.getResponseBody (), RaftPeer.class)); return 0 }});} catch (Exception e) {peer.state = RaftPeer.State.FOLLOWER; Loggers.RAFT.error ("[NACOS-RAFT] error while getting peer from peer: {}", peer.ip);} return update (candidate) } public RaftPeer local () {RaftPeer peer = peers.get (NetUtils.localServer ()); if (peer = = null & & SystemUtils.STANDALONE_MODE) {RaftPeer localPeer = new RaftPeer (); localPeer.ip = NetUtils.localServer (); localPeer.term.set (localTerm.get ()); peers.put (localPeer.ip, localPeer); return localPeer } if (peer = = null) {throw new IllegalStateException ("unable to find local peer:" + NetUtils.localServer () + ", all peers:" + Arrays.toString (peers.keySet (). ToArray ());} return peer;} public RaftPeer get (String server) {return peers.get (server)) } public int majorityCount () {return peers.size () / 2 + 1;} public void reset () {leader = null; for (RaftPeer peer: peers.values ()) {peer.voteFor = null;}} public void setTerm (long term) {localTerm.set (term);} public long getTerm () {return localTerm.get () } public boolean contains (RaftPeer remote) {return peers.containsKey (remote.ip);} /.}

RaftPeerSet provides remove, update, isLeader, allServersIncludeMyself, allServersWithoutMySelf, allPeers, decideLeader, makeLeader, majorityCount, reset and other methods.

The decideLeader method traverses the peers, and then uses TreeBag to count the peer.voteFor. When maxApproveCount is greater than or equal to majorityCount (), the state of the corresponding peer is marked as RaftPeer.State.LEADER, and then determines whether the leader has changed, and the LeaderElectFinishedEvent event is issued when the change is made.

The makeLeader method determines whether the candidate is consistent with the current leader. If it is inconsistent, update the leader to candidate, issue the MakeLeaderEvent event, then traverse the peers to send the API_GET_PEER request to the node that is not candidate and the state is in the LEADER state, then update the local information of the peer, and update its state to RaftPeer.State.FOLLOWER if the request fails

RaftCore.MasterElection

Nacos-1.1.3/naming/src/main/java/com/alibaba/nacos/naming/consistency/persistent/raft/RaftCore.java

Public class MasterElection implements Runnable {@ Override public void run () {try {if (! peers.isReady ()) {return;} RaftPeer local = peers.local (); local.leaderDueMs-= GlobalExecutor.TICK_PERIOD_MS If (local.leaderDueMs > 0) {return;} / / reset timeout local.resetLeaderDue (); local.resetHeartbeatDue (); sendVote () } catch (Exception e) {Loggers.RAFT.warn ("[RAFT] error while master election {}", e);}} public void sendVote () {RaftPeer local = peers.get (NetUtils.localServer ()) Loggers.RAFT.info ("leader timeout, start voting,leader: {}, term: {}", JSON.toJSONString (getLeader ()), local.term); peers.reset (); local.term.incrementAndGet (); local.voteFor = local.ip; local.state = RaftPeer.State.CANDIDATE; Map params = new HashMap (1) Params.put ("vote", JSON.toJSONString (local)); for (final String server: peers.allServersWithoutMySelf ()) {final String url = buildURL (server, API_VOTE) Try {HttpClient.asyncHttpPost (url, null, params) New AsyncCompletionHandler () {@ Override public Integer onCompleted (Response response) throws Exception {if (response.getStatusCode ()! = HttpURLConnection.HTTP_OK) {Loggers.RAFT.error ("NACOS-RAFT vote failed: {}, url: {}", response.getResponseBody (), url) Return 1;} RaftPeer peer = JSON.parseObject (response.getResponseBody (), RaftPeer.class); Loggers.RAFT.info ("received approve from peer: {}", JSON.toJSONString (peer)); peers.decideLeader (peer) Return 0;}});} catch (Exception e) {Loggers.RAFT.warn ("error while sending vote to server: {}", server);}}

When the request is successful, the sendVote method of RaftCore.MasterElection executes the peers.decideLeader (peer) method to elect leader

RaftCore.receivedBeat

Nacos-1.1.3/naming/src/main/java/com/alibaba/nacos/naming/consistency/persistent/raft/RaftCore.java

@ Componentpublic class RaftCore {/ / Public RaftPeer receivedBeat (JSONObject beat) throws Exception {final RaftPeer local = peers.local (); final RaftPeer remote = new RaftPeer (); remote.ip = beat.getJSONObject ("peer"). GetString ("ip"); remote.state = RaftPeer.State.valueOf (beat.getJSONObject ("peer"). GetString ("state"); remote.term.set (beat.getJSONObject ("peer"). GetLongValue ("term")) Remote.heartbeatDueMs = beat.getJSONObject ("peer"). GetLongValue ("heartbeatDueMs"); remote.leaderDueMs = beat.getJSONObject ("peer"). GetLongValue ("leaderDueMs"); remote.voteFor = beat.getJSONObject ("peer"). GetString ("voteFor") If (remote.state! = RaftPeer.State.LEADER) {Loggers.RAFT.info ("[RAFT] invalid state from master, state: {}, remote peer: {}", remote.state, JSON.toJSONString (remote)); throw new IllegalArgumentException ("invalid state from master, state:" + remote.state) } if (local.term.get () > remote.term.get ()) {Loggers.RAFT.info ("[RAFT] out of date beat, beat-from-term: {}, beat-to-term: {}, remote peer: {}, and leaderDueMs: {}", remote.term.get (), local.term.get (), JSON.toJSONString (remote), local.leaderDueMs) Throw new IllegalArgumentException ("out of date beat, beat-from-term:" + remote.term.get () + ", beat-to-term:" + local.term.get ());} if (local.state! = RaftPeer.State.FOLLOWER) {Loggers.RAFT.info ("[RAFT] make remote as leader, remote peer: {}", JSON.toJSONString (remote) / / mk follower local.state = RaftPeer.State.FOLLOWER; local.voteFor = remote.ip;} final JSONArray beatDatums = beat.getJSONArray ("datums"); local.resetLeaderDue (); local.resetHeartbeatDue (); peers.makeLeader (remote); Map receivedKeysMap = new HashMap (datums.size ()) For (Map.Entry entry: datums.entrySet ()) {receivedKeysMap.put (entry.getKey (), 0);} / / now check datums List batch = new ArrayList (); if (! switchDomain.isSendBeatOnly ()) {int processedCount = 0 If (Loggers.RAFT.isDebugEnabled ()) {Loggers.RAFT.debug ("[RAFT] received beat with {} keys, RaftCore.datums' size is {}, remote server: {}, term: {}, local term: {}", beatDatums.size (), datums.size (), remote.ip, remote.term, local.term) } for (Object object: beatDatums) {processedCount = processedCount + 1; JSONObject entry = (JSONObject) object; String key = entry.getString ("key"); final String datumKey; if (KeyBuilder.matchServiceMetaKey (key)) {datumKey = KeyBuilder.detailServiceMetaKey (key) } else if (KeyBuilder.matchInstanceListKey (key)) {datumKey = KeyBuilder.detailInstanceListkey (key);} else {/ / ignore corrupted key: continue;} long timestamp = entry.getLong ("timestamp"); receivedKeysMap.put (datumKey, 1) Try {if (datums.containsKey (datumKey) & & datums.get (datumKey). Timestamp.get () > = timestamp & & processedCount

= timestamp) {batch.add (datumKey);} if (batch.size () < 50 & & processedCount < beatDatums.size ()) {continue;} String keys = StringUtils.join (batch, ",") If (batch.size ()

Welcome to subscribe "Shulou Technology Information " to get latest news, interesting things and hot topics in the IT industry, and controls the hottest and latest Internet news, technology news and IT industry trends.