[[405925]]

從本質(zhì)上說(shuō)，Raft 算法是通過(guò)一切以領(lǐng)導(dǎo)者為準(zhǔn)的方式，實(shí)現(xiàn)一系列值的共識(shí)和各節(jié)點(diǎn)日志的一致

• Leader 選舉，Leader 故障后集群能快速選出新 Leader;
• 日志復(fù)制， 集群只有 Leader 能寫入日志， Leader 負(fù)責(zé)復(fù)制日志到 Follower 節(jié)點(diǎn)，并強(qiáng)制 Follower 節(jié)點(diǎn)與自己保持相同;
• 安全性，成員變更，一個(gè)任期內(nèi)集群只能產(chǎn)生一個(gè) Leader、已提交的日志條目在發(fā)生 Leader 選舉時(shí)，一定會(huì)存在更高任期的新 Leader 日志中、各個(gè)節(jié)點(diǎn)的狀態(tài)機(jī)應(yīng)用的任意位置的日志條目?jī)?nèi)容應(yīng)一樣等。

Leader 選舉

raft 算法本質(zhì)上是一個(gè)大的狀態(tài)機(jī)，任何的操作例如選舉、提交數(shù)據(jù)等，最后都被封裝成一個(gè)消息結(jié)構(gòu)體，輸入到 raft 算法庫(kù)的狀態(tài)機(jī)中。raft 算法其實(shí)由好幾個(gè)協(xié)議組成，etcd-raft 將其統(tǒng)一定義在了 Message 結(jié)構(gòu)體之中，以下總結(jié)了該結(jié)構(gòu)體的成員用途：

type Message struct { 
Type             MessageType `protobuf:"varint,1,opt,name=type,enum=raftpb.MessageType" json:"type"` // 消息類型 
To               uint64      `protobuf:"varint,2,opt,name=to" json:"to"` // 消息接收者的節(jié)點(diǎn)ID 
From             uint64      `protobuf:"varint,3,opt,name=from" json:"from"` // 消息發(fā)送者的節(jié)點(diǎn) ID 
Term             uint64      `protobuf:"varint,4,opt,name=term" json:"term"` // 發(fā)送消息的節(jié)點(diǎn)的Term值。如果Term值為0，則為本地消息，在etcd-raft模塊的實(shí)現(xiàn)中，對(duì)本地消息進(jìn)行特殊處理。 
LogTerm          uint64      `protobuf:"varint,5,opt,name=logTerm" json:"logTerm"` // 該消息攜帶的第一條Entry記錄的Term值，日志所處的任期ID 
Index            uint64      `protobuf:"varint,6,opt,name=index" json:"index"` // 日志索引ID，用于節(jié)點(diǎn)向 Leader 匯報(bào)自己已經(jīng)commit的日志數(shù)據(jù)ID 
Entries          []Entry     `protobuf:"bytes,7,rep,name=entries" json:"entries"` // 如果是MsgApp類型的消息，則該字段中保存了Leader節(jié)點(diǎn)復(fù)制到Follower節(jié)點(diǎn)的Entry記錄 
Commit           uint64      `protobuf:"varint,8,opt,name=commit" json:"commit"` // 消息發(fā)送節(jié)點(diǎn)提交日志索引 
Snapshot         Snapshot    `protobuf:"bytes,9,opt,name=snapshot" json:"snapshot"` // 在傳輸快照時(shí)，該字段保存了快照數(shù)據(jù) 
Reject           bool        `protobuf:"varint,10,opt,name=reject" json:"reject"` // 主要用于響應(yīng)類型的消息，表示是否拒絕收到的消息 
RejectHint       uint64      `protobuf:"varint,11,opt,name=rejectHint" json:"rejectHint"` //在Follower節(jié)點(diǎn)拒絕Leader節(jié)點(diǎn)的消息之后，會(huì)在該字段記錄一個(gè)Entry索引值供Leader節(jié)點(diǎn) 
Context          []byte      `protobuf:"bytes,12,opt,name=context" json:"context,omitempty"` // 消息攜帶的一些上下文信息。例如，該消息是否與Leader節(jié)點(diǎn)轉(zhuǎn)移相關(guān) 
XXX_unrecognized []byte      `json:"-"` 
} 

Message結(jié)構(gòu)體相關(guān)的數(shù)據(jù)類型為 MessageType，MessageType 有 19 種。當(dāng)然，并不是所有的消息類型都會(huì)用到上面定義的Message結(jié)構(gòu)體中的所有字段，因此其中有些字段是Optinal的。

   MsgHup            MessageType = 0  //當(dāng)Follower節(jié)點(diǎn)的選舉計(jì)時(shí)器超時(shí)，會(huì)發(fā)送MsgHup消息 
MsgBeat           MessageType = 1  //Leader發(fā)送心跳，主要作用是探活，F(xiàn)ollower接收到MsgBeat會(huì)重置選舉計(jì)時(shí)器，防止Follower發(fā)起新一輪選舉 
MsgProp           MessageType = 2  //客戶端發(fā)往到集群的寫請(qǐng)求是通過(guò)MsgProp消息表示的 
MsgApp            MessageType = 3  //當(dāng)一個(gè)節(jié)點(diǎn)通過(guò)選舉成為L(zhǎng)eader時(shí)，會(huì)發(fā)送MsgApp消息 
MsgAppResp        MessageType = 4  //MsgApp的響應(yīng)消息 
MsgVote           MessageType = 5  //當(dāng)PreCandidate狀態(tài)節(jié)點(diǎn)收到半數(shù)以上的投票之后，會(huì)發(fā)起新一輪的選舉，即向集群中的其他節(jié)點(diǎn)發(fā)送MsgVote消息 
MsgVoteResp       MessageType = 6  //MsgVote選舉消息響應(yīng)的消息 
MsgSnap           MessageType = 7  //Leader向Follower發(fā)送快照信息 
MsgHeartbeat      MessageType = 8  //Leader發(fā)送的心跳消息 
MsgHeartbeatResp  MessageType = 9  //Follower處理心跳回復(fù)返回的消息類型 
MsgUnreachable    MessageType = 10 //Follower消息不可達(dá) 
MsgSnapStatus     MessageType = 11 //如果Leader發(fā)送MsgSnap消息時(shí)出現(xiàn)異常，則會(huì)調(diào)用Raft接口發(fā)送MsgUnreachable和MsgSnapStatus消息 
MsgCheckQuorum    MessageType = 12 //Leader檢測(cè)是否保持半數(shù)以上的連接 
MsgTransferLeader MessageType = 13 //Leader節(jié)點(diǎn)轉(zhuǎn)移時(shí)使用，本地消息 
MsgTimeoutNow     MessageType = 14 //Leader節(jié)點(diǎn)轉(zhuǎn)移超時(shí)，會(huì)發(fā)該類型的消息，使Follower的選舉計(jì)時(shí)器立即過(guò)期，并發(fā)起新一輪的選舉 
MsgReadIndex      MessageType = 15 //客戶端發(fā)往集群的只讀消息使用MsgReadIndex消息（只讀的兩種模式：ReadOnlySafe和ReadOnlyLeaseBased） 
MsgReadIndexResp  MessageType = 16 //MsgReadIndex消息的響應(yīng)消息 
MsgPreVote        MessageType = 17 //PreCandidate狀態(tài)下的節(jié)點(diǎn)發(fā)送的消息 
MsgPreVoteResp    MessageType = 18 //預(yù)選節(jié)點(diǎn)收到的響應(yīng)消息   

然后是 raft 算法的實(shí)現(xiàn)，node 結(jié)構(gòu)體實(shí)現(xiàn)了 Node 接口，對(duì)etcd-raft模塊具體實(shí)現(xiàn)的一層封裝，方便上層模塊使用etcd-raft模塊。其定義如下：

type node struct { 
 
propc      chan msgWithResult      //該通道用于接收MsgProp類型的消息 
 
recvc      chan pb.Message         //除MsgProp外的其他類型的消息都是由該通道接收的 
 
confc      chan pb.ConfChangeV2    //當(dāng)節(jié)點(diǎn)收到EntryConfChange類型的Entry記錄時(shí)，會(huì)轉(zhuǎn)換成ConfChange，并寫入該通道中等待處理。在ConfChange中封裝了其唯一 ID、待處理的節(jié)點(diǎn) ID （NodeID 字段）及處理類型（Type 字段，例如，ConfChangeAddNode類型表示添加節(jié)點(diǎn)）等信息 
confstatec chan pb.ConfState       //在ConfState中封裝了當(dāng)前集群中所有節(jié)點(diǎn)的ID，該通道用于向上層模塊返回ConfState實(shí)例 
 
readyc     chan Ready              //Ready結(jié)構(gòu)體的功能在上一小節(jié)已經(jīng)介紹過(guò)了，該通道用于向上層模塊返回Ready實(shí)例，即node.Ready（）方法的返回值 
 
advancec   chan struct{}           //當(dāng)上層模塊處理完通過(guò)上述readyc通道獲取到的Ready實(shí)例之后，會(huì)通過(guò)node.Advance（）方法向該通道寫入信號(hào)，從而通知底層raft實(shí)例 
 
tickc      chan struct{}                //用來(lái)接收邏輯時(shí)鐘發(fā)出的信號(hào)，之后會(huì)根據(jù)當(dāng)前節(jié)點(diǎn)的角色推進(jìn)選舉計(jì)時(shí)器和心跳計(jì)時(shí)器 
 
done       chan struct{}           //當(dāng)檢測(cè)到done通道關(guān)閉后，在其上阻塞的goroutine會(huì)繼續(xù)執(zhí)行，并進(jìn)行相應(yīng)的關(guān)閉操作 
 
stop       chan struct{}           //當(dāng)node.Stop（）方法被調(diào)用時(shí)，會(huì)向該通道發(fā)送信號(hào)，在后續(xù)介紹中會(huì)提到，有另一個(gè)goroutine會(huì)嘗試讀取該通道中的內(nèi)容，當(dāng)讀取到信息之后，會(huì)關(guān)閉done通道。 
 
status     chan chan Status        //注意該通道的類型，其中傳遞的元素也是Channel類型，即node.Status（）方法的返回值 
 
 rn        *RawNode 
 
} 

下面我們來(lái)看看 raft StateMachine 的狀態(tài)機(jī)轉(zhuǎn)換，實(shí)際上就是 raft 算法中各種角色的轉(zhuǎn)換。每個(gè) raft 節(jié)點(diǎn)，可能具有以下三種狀態(tài)中的一種。

• Candidate：候選人狀態(tài)，該狀態(tài)意味著將進(jìn)行一次新的選舉。
• Follower：跟隨者狀態(tài)，該狀態(tài)意味著選舉結(jié)束。
• Leader：領(lǐng)導(dǎo)者狀態(tài)，選舉出來(lái)的節(jié)點(diǎn)，所有數(shù)據(jù)提交都必須先提交到 Leader 上。

每一個(gè)狀態(tài)都有其對(duì)應(yīng)的狀態(tài)機(jī)，每次收到一條提交的數(shù)據(jù)時(shí)，都會(huì)根據(jù)其不同的狀態(tài)將消息輸入到不同狀態(tài)的狀態(tài)機(jī)中。同時(shí)，在進(jìn)行 tick 操作時(shí)，每種狀態(tài)對(duì)應(yīng)的處理函數(shù)也是不一樣的。因此 raft 結(jié)構(gòu)體中將不同的狀態(tài)及其不同的處理函數(shù)，獨(dú)立出來(lái)幾個(gè)成員變量：

• state，保存當(dāng)前節(jié)點(diǎn)狀態(tài);
• tick 函數(shù)，每個(gè)狀態(tài)對(duì)應(yīng)的 tick 函數(shù)不同;
• step，狀態(tài)機(jī)函數(shù)，同樣每個(gè)狀態(tài)對(duì)應(yīng)的狀態(tài)機(jī)也不相同

我們接著看 etcd-raft 狀態(tài)轉(zhuǎn)換。etcd-raft StateMachine 封裝在 raft機(jī)構(gòu)體中，etcd為了不讓entry落后的太多的直接進(jìn)行選舉，多了一個(gè)其PreCandidate狀態(tài)，轉(zhuǎn)換如下圖：

raft 狀態(tài)轉(zhuǎn)換的接口都在 raft.go 中，其定義如下：

//在newRaft（）函數(shù)中完成初始化之后，會(huì)調(diào)用 becomeFollower（）方法將節(jié)點(diǎn)切換成 Follower狀態(tài)，其中會(huì)設(shè)置raft實(shí)例的多個(gè)字段 
func (r *raft) becomeFollower(term uint64, lead uint64) { 
 r.step = stepFollower //設(shè)置函數(shù)處理Follower節(jié)點(diǎn)處理消息的行為 
 r.reset(term) //在reset（）方法中會(huì)重置raft實(shí)例的多個(gè)字段 
 r.tick = r.tickElection //將tick字段設(shè)置成tickElection函數(shù) 
 r.lead = lead //設(shè)置當(dāng)前節(jié)點(diǎn)的leader節(jié)點(diǎn) 
    //修改當(dāng)前節(jié)點(diǎn)的角色 
 r.state = StateFollower 
} 
 
//如果當(dāng)前集群開啟了 PreVote 模式，當(dāng) Follower 節(jié)點(diǎn)的選舉計(jì)時(shí)器超時(shí)時(shí)，會(huì)先調(diào)用becomePreCandidate（）方法切換到PreCandidate狀態(tài)，becomePreCandidate（） 
func (r *raft) becomePreCandidate() { 
    //檢查當(dāng)前節(jié)點(diǎn)的狀態(tài)，禁止leader直接切換到PreCandidate狀態(tài) 
 if r.state == StateLeader { 
  panic("invalid transition [leader -> pre-candidate]") 
 } 
    //設(shè)置函數(shù)處理Candidate節(jié)點(diǎn)處理消息的行為 
 r.step = stepCandidate  
 r.prs.ResetVotes() 
 r.tick = r.tickElection 
 r.lead = None 
    //修改當(dāng)前節(jié)點(diǎn)的角色 
 r.state = StatePreCandidate  
} 
//當(dāng)節(jié)點(diǎn)可以連接到集群中半數(shù)以上的節(jié)點(diǎn)時(shí)，會(huì)調(diào)用 becomeCandidate（）方法切換到Candidate狀態(tài)，becomeCandidate（） 
func (r *raft) becomeCandidate() { 
 // TODO(xiangli) remove the panic when the raft implementation is stable 
 if r.state == StateLeader { 
  panic("invalid transition [leader -> candidate]") 
 } 
    //在reset（）方法中會(huì)重置raft實(shí)例的多個(gè)字段 
 r.step = stepCandidate 
 r.reset(r.Term + 1) //在reset（）方法中會(huì)重置raft實(shí)例的多個(gè)字段 
 r.tick = r.tickElection 
 r.Vote = r.id //在此次的選舉中，Candidate節(jié)點(diǎn)會(huì)將選票投給自己 
    //修改當(dāng)前節(jié)點(diǎn)的角色 
 r.state = StateCandidate 
} 
 
//當(dāng) Candidate 節(jié)點(diǎn)得到集群中半數(shù)以上節(jié)點(diǎn)的選票時(shí)，會(huì)調(diào)用 becomeLeader（）方法切換成Leader狀態(tài)，becomeLeader（） 
func (r *raft) becomeLeader() { 
    //檢查當(dāng)前節(jié)點(diǎn)的狀態(tài)，機(jī)制從follower直接切換成leader狀態(tài) 
 if r.state == StateFollower { 
  panic("invalid transition [follower -> leader]") 
 } 
 r.step = stepLeader 
 r.reset(r.Term) //在reset（）方法中會(huì)重置raft實(shí)例的多個(gè)字段 
 r.tick = r.tickHeartbeat 
 r.lead = r.id //將leader字段設(shè)置成當(dāng)前節(jié)點(diǎn)的id 
 r.state = StateLeader //更新當(dāng)前節(jié)點(diǎn)的角色 
    //檢查未提交的記錄中是否存在多條集群配置變更的Entry記錄 
 r.prs.Progress[r.id].BecomeReplicate() 
 r.pendingConfIndex = r.raftLog.lastIndex() 
 emptyEnt := pb.Entry{Data: nil} 
    //向當(dāng)前節(jié)點(diǎn)的raftLog中追加一條空的Entry記錄 
 if !r.appendEntry(emptyEnt) { 
    } 
 r.reduceUncommittedSize([]pb.Entry{emptyEnt}) 
} 

tick 函數(shù)，每個(gè)狀態(tài)對(duì)應(yīng)的 tick 函數(shù)不同，下面分析兩個(gè)tick：

func (r *raft) tickElection() { 
 r.electionElapsed++ //遞增electionElapsed計(jì)時(shí)器 
 
 if r.promotable() && r.pastElectionTimeout() { //檢查是否在集群中與檢查單簽的選舉計(jì)時(shí)器是否超時(shí) 
  r.electionElapsed = 0 
  r.Step(pb.Message{From: r.id, Type: pb.MsgHup}) //發(fā)起step處理pb.MsgHup類型消息。 
 } 
} 
 
func (r *raft) tickHeartbeat() { 
 r.heartbeatElapsed++ //遞增heartbeatElapsed計(jì)時(shí)器 
 r.electionElapsed++ //遞增electionElapsed計(jì)時(shí)器 
 if r.electionElapsed >= r.electionTimeout { 
  r.electionElapsed = 0 //重置選舉計(jì)時(shí)器，leader節(jié)點(diǎn)不會(huì)主動(dòng)發(fā)起選舉 
  if r.checkQuorum { //進(jìn)行多數(shù)檢查 
   r.Step(pb.Message{From: r.id, Type: pb.MsgCheckQuorum}) //發(fā)起大多數(shù)檢查。 
  } 
        //選舉計(jì)時(shí)器處于electionElapsed~randomizedElectionTimeout時(shí)段之間時(shí)，不能進(jìn)行l(wèi)eader轉(zhuǎn)移 
  if r.state == StateLeader && r.leadTransferee != None { 
   r.abortLeaderTransfer() //清空raft.leadTransferee字段，放棄轉(zhuǎn)移 
  } 
 } 
 if r.state != StateLeader { //只有l(wèi)aeder能發(fā)送tickHeartbeat 
  return 
 } 
 if r.heartbeatElapsed >= r.heartbeatTimeout { //心跳計(jì)時(shí)器超時(shí) 
  r.heartbeatElapsed = 0 //重置心跳計(jì)時(shí)器 
  r.Step(pb.Message{From: r.id, Type: pb.MsgBeat}) //發(fā)起step處理MsgBeat類型消息 
 } 
} 

跟隨者、預(yù)選候選人、候選人、領(lǐng)導(dǎo)者 4 種節(jié)點(diǎn)狀態(tài)都有分別對(duì)應(yīng)的功能函數(shù)，當(dāng)需要查看各節(jié)點(diǎn)狀態(tài)相關(guān)的功能實(shí)現(xiàn)時(shí)(比如，跟隨者如何接收和處理日志)，都可以將對(duì)應(yīng)的函數(shù)作為入口函數(shù)，來(lái)閱讀代碼和研究功能實(shí)現(xiàn)。

日志復(fù)制

這里重點(diǎn)看一下raft.appendEntry()方法，它的主要操作步驟如下：(1)設(shè)置待追加的Entry記錄的Term值和Index值。

(2)向當(dāng)前節(jié)點(diǎn)的raftLog中追加Entry記錄。

(3)更新當(dāng)前節(jié)點(diǎn)對(duì)應(yīng)的Progress實(shí)例。

(4)嘗試提交Entry記錄，即修改raftLog.committed字段的值。

raft.appendEntry()方法的具體實(shí)現(xiàn)如下：

func (r *raft) appendEntry(es ...pb.Entry) (accepted bool) { 
 li := r.raftLog.lastIndex()//獲取raftLog中最后一條記錄的索引值 
 for i := range es {//更新待追加記錄的Term值和索引值 
  es[i].Term = r.Term//Entry記錄的Term指定為當(dāng)前l(fā)eader節(jié)點(diǎn)的任期號(hào) 
  es[i].Index = li + 1 + uint64(i) //為日志記錄指定的Index 
 } 
 li = r.raftLog.append(es...)//向raft中追加記錄 
    //更新當(dāng)前節(jié)點(diǎn)對(duì)應(yīng)的Progress,主要是更新Next和Match 
 r.prs.Progress[r.id].MaybeUpdate(li) 
    //嘗試提交Entry記錄 
 r.maybeCommit() 
 return true 
} 

在Progress.mayUpdate()方法中，會(huì)嘗試修改Match字段和Next字段，用來(lái)標(biāo)識(shí)對(duì)應(yīng)節(jié)點(diǎn)Entry記錄復(fù)制的情況。Leader節(jié)點(diǎn)除了在向自身raftLog中追加記錄時(shí)(即appendEntry()方法)會(huì)調(diào)用該方法，當(dāng)Leader節(jié)點(diǎn)收到Follower節(jié)點(diǎn)的MsgAppResp消息(即MsgApp消息的響應(yīng)消息)時(shí)，也會(huì)調(diào)用該方法嘗試修改Follower節(jié)點(diǎn)對(duì)應(yīng)的Progress實(shí)例。Progress.MayUpdate()方法的具體實(shí)現(xiàn)如下：

func (pr *Progress) MaybeUpdate(n uint64) bool { 
 var updated bool 
 if pr.Match < n { 
  pr.Match = n //n之前所有的Entry記錄都已經(jīng)寫入對(duì)應(yīng)節(jié)點(diǎn)的raftLog中 
  updated = true 
        //下面將Progress.paused設(shè)置為false，表示leader節(jié)點(diǎn)可以繼續(xù)向?qū)?yīng)Follower 
        //節(jié)點(diǎn)發(fā)送MsgApp消息 
  pr.ProbeAcked() 
 } 
 pr.Next = max(pr.Next, n+1)//將Next值加一，下一次復(fù)制Entry記錄開始的位置 
 return updated 
} 

如果該Entry記錄已經(jīng)復(fù)制到了半數(shù)以上的節(jié)點(diǎn)中，則在raft.maybeCommit()方法中會(huì)嘗試將其提交。除了 appendEntry()方法，在 Leader 節(jié)點(diǎn)每次收到 MsgAppResp 消息時(shí)也會(huì)調(diào)用maybeCommit()方法，maybeCommit()方法的具體實(shí)現(xiàn)如下：

func (r *raft) maybeCommit() bool { 
 mci := r.prs.Committed() 
 return r.raftLog.maybeCommit(mci, r.Term) 
} 
 
func (p *ProgressTracker) Committed() uint64 { 
 return uint64(p.Voters.CommittedIndex(matchAckIndexer(p.Progress))) 
} 
//將node分兩個(gè)組，JointConfig是大多數(shù)的組，有興趣的看一看quorum包的實(shí)現(xiàn) 
func (c JointConfig) CommittedIndex(l AckedIndexer) Index {//比較大多數(shù)的node的前倆個(gè)Index，返回Match的值。 
 idx0 := c[0].CommittedIndex(l) 
 idx1 := c[1].CommittedIndex(l) 
 if idx0 < idx1 { 
  return idx0 
 } 
 return idx1 
} 
//更新raftLog.committed字段，完成提交 
func (l *raftLog) maybeCommit(maxIndex, term uint64) bool { 
 if maxIndex > l.committed && l.zeroTermOnErrCompacted(l.term(maxIndex)) == term { 
  l.commitTo(maxIndex) 
  return true 
 } 
 return false 
} 

etcd 將 raft 相關(guān)的所有處理都抽象為了 Message，通過(guò) Step 接口處理各類消息的入口，首先根據(jù)Term"值"對(duì)消息進(jìn)行分類處理，再根據(jù)消息的"類型"進(jìn)行分類處理：

func (r *raft) Step(m pb.Message) error { 
 switch {//首先根據(jù)消息的Term值進(jìn)行分類處理 
 case m.Term == 0://本地消息不做處理。MsgHup，MsgProp和MsgReadIndex是本地消息 
 case m.Term > r.Term: 
 case m.Term < r.Term://細(xì)節(jié)部分，可以自己研究源碼 
 } 
 switch m.Type {//根據(jù)Message的Type進(jìn)行分類處理 
 case pb.MsgHup://這里針對(duì)MsgHup類型的消息進(jìn)行處理。 
  if r.preVote {//檢查是不是開啟了preVote，如果是開啟了先調(diào)用raft.hup方法，發(fā)起preVote。 
  } else { 
   r.hup(campaignElection)//下面講述 
  } 
 case pb.MsgVote, pb.MsgPreVote: //對(duì)MsgVote，MsgPreVote類型的消息進(jìn)行處理。 
  canVote := r.Vote == m.From || 
   (r.Vote == None && r.lead == None) || 
   (m.Type == pb.MsgPreVote && m.Term > r.Term) 
  if canVote && r.raftLog.isUpToDate(m.Index, m.LogTerm) { 
   r.send(pb.Message{To: m.From, Term: m.Term, Type: voteRespMsgType(m.Type)}) 
   if m.Type == pb.MsgVote { 
    r.electionElapsed = 0 
    r.Vote = m.From 
   } 
  } else { 
   r.send(pb.Message{To: m.From, Term: r.Term, Type: voteRespMsgType(m.Type), Reject: true}) 
  } 
 default://對(duì)于其他類型的消息處理，對(duì)應(yīng)的node的step函數(shù)處理 
  err := r.step(r, m) 
  if err != nil { 
   return err 
  } 
 } 
 return nil 
} 

這里主要使用hup函數(shù)對(duì)Message來(lái)做處理，在raft.campaign()方法中，除了完成狀態(tài)切換，還會(huì)向集群中的其他節(jié)點(diǎn)發(fā)送相應(yīng)類型的消息，例如，如果當(dāng)前 Follower 節(jié)點(diǎn)要切換成 PreCandidate 狀態(tài)，則會(huì)發(fā)送 MsgPreVote 消息：

func (r *raft) hup(t CampaignType) { 
 if r.state == StateLeader {//忽略leader 
  return 
 } 
    //方法會(huì)檢查prs字段中是否還存在當(dāng)前節(jié)點(diǎn)對(duì)應(yīng)的Progress實(shí)例，這是為了監(jiān)測(cè)當(dāng)前節(jié)點(diǎn)是否被從集群中移除了 
    if !r.promotable() { 
  return 
 } 
    //獲取raftLog中已提交但未應(yīng)用的Entry記錄，異常處理 
 ents, err := r.raftLog.slice(r.raftLog.applied+1, r.raftLog.committed+1, noLimit) 
 r.campaign(t) 
} 
func (r *raft) campaign(t CampaignType) { 
    //該方法的會(huì)發(fā)送一條包含Term值和類型 
 var term uint64 
 var voteMsg pb.MessageType 
 if t == campaignPreElection {//切換的目標(biāo)狀態(tài)是Precandidate 
  r.becomePreCandidate() 
  voteMsg = pb.MsgPreVote 
        //確定要發(fā)送的Term值，這里只是增加了消息的Term值，并未增加raft.term字段的值 
  term = r.Term + 1 
 } else {//切換的目標(biāo)狀態(tài)是Candidate 
  r.becomeCandidate() 
  voteMsg = pb.MsgVote 
        //給raft.Term字段的值，并將當(dāng)前節(jié)點(diǎn)的選票投給自身 
  term = r.Term 
 } 
 if _, _, res := r.poll(r.id, voteRespMsgType(voteMsg), true); res == quorum.VoteWon { 
        //當(dāng)?shù)玫阶銐虻倪x票時(shí)，則將PreCandidate狀態(tài)的節(jié)點(diǎn)切換成Candidate狀態(tài) 
        //Candidate狀態(tài)的節(jié)點(diǎn)則切換成Leader狀態(tài) 
  if t == campaignPreElection { 
   r.campaign(campaignElection) 
  } else { 
   r.becomeLeader() 
  } 
  return 
 } 
 var ids []uint64 
 { 
  idMap := r.prs.Voters.IDs() 
  ids = make([]uint64, 0, len(idMap)) 
  for id := range idMap { 
   ids = append(ids, id) 
  } 
  sort.Slice(ids, func(i, j int) bool { return ids[i] < ids[j] }) 
 } 
 for _, id := range ids {//狀態(tài)切換完成之后，當(dāng)前節(jié)點(diǎn)會(huì)向集群中所有節(jié)點(diǎn)發(fā)送指定類型的消息 
  if id == r.id { //跳過(guò)當(dāng)前節(jié)點(diǎn)自身 
   continue 
  } 
        var ctx []byte 
        //在進(jìn)行Leader節(jié)點(diǎn)轉(zhuǎn)移時(shí)，MsgPreVote或MsgVote消息會(huì)在Context字段中設(shè)置該特殊值 
  if t == campaignTransfer { 
   ctx = []byte(t) 
  } 
        //發(fā)送指定類型的消息，其中Index和LogTerm分別是當(dāng)前節(jié)點(diǎn)的raftLog 
        //最后一條消息的Index值和Term值 
  r.send(pb.Message{Term: term, To: id, Type: voteMsg, Index: r.raftLog.lastIndex(), LogTerm: r.raftLog.lastTerm(), Context: ctx}) 
 } 
} 

Follower 節(jié)點(diǎn)在選舉計(jì)時(shí)器超時(shí)的行為：首先它會(huì)通過(guò) tickElection()創(chuàng)建MsgHup消息并將其交給raft.Step()方法進(jìn)行處理;raft.Step()方法會(huì)將當(dāng)前Follower節(jié)點(diǎn)切換成PreCandidate狀態(tài)，然后創(chuàng)建MsgPreVote類型的消息，最后將該消息追加到raft.msgs字段中，等待上層模塊將其發(fā)送出去。

本文轉(zhuǎn)載自微信公眾號(hào)「運(yùn)維開發(fā)故事」，可以通過(guò)以下二維碼關(guān)注。轉(zhuǎn)載本文請(qǐng)聯(lián)系運(yùn)維開發(fā)故事公眾號(hào)。