[[405925]]

從本質上說，Raft 算法是通過一切以領導者為準的方式，實現一系列值的共識和各節點日志的一致

• Leader 選舉，Leader 故障后集群能快速選出新 Leader;
• 日志復制， 集群只有 Leader 能寫入日志， Leader 負責復制日志到 Follower 節點，并強制 Follower 節點與自己保持相同;
• 安全性，成員變更，一個任期內集群只能產生一個 Leader、已提交的日志條目在發生 Leader 選舉時，一定會存在更高任期的新 Leader 日志中、各個節點的狀態機應用的任意位置的日志條目內容應一樣等。

Leader 選舉

raft 算法本質上是一個大的狀態機，任何的操作例如選舉、提交數據等，最后都被封裝成一個消息結構體，輸入到 raft 算法庫的狀態機中。raft 算法其實由好幾個協議組成，etcd-raft 將其統一定義在了 Message 結構體之中，以下總結了該結構體的成員用途：

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"` // 消息接收者的節點ID 
From             uint64      `protobuf:"varint,3,opt,name=from" json:"from"` // 消息發送者的節點 ID 
Term             uint64      `protobuf:"varint,4,opt,name=term" json:"term"` // 發送消息的節點的Term值。如果Term值為0，則為本地消息，在etcd-raft模塊的實現中，對本地消息進行特殊處理。 
LogTerm          uint64      `protobuf:"varint,5,opt,name=logTerm" json:"logTerm"` // 該消息攜帶的第一條Entry記錄的Term值，日志所處的任期ID 
Index            uint64      `protobuf:"varint,6,opt,name=index" json:"index"` // 日志索引ID，用于節點向 Leader 匯報自己已經commit的日志數據ID 
Entries          []Entry     `protobuf:"bytes,7,rep,name=entries" json:"entries"` // 如果是MsgApp類型的消息，則該字段中保存了Leader節點復制到Follower節點的Entry記錄 
Commit           uint64      `protobuf:"varint,8,opt,name=commit" json:"commit"` // 消息發送節點提交日志索引 
Snapshot         Snapshot    `protobuf:"bytes,9,opt,name=snapshot" json:"snapshot"` // 在傳輸快照時，該字段保存了快照數據 
Reject           bool        `protobuf:"varint,10,opt,name=reject" json:"reject"` // 主要用于響應類型的消息，表示是否拒絕收到的消息 
RejectHint       uint64      `protobuf:"varint,11,opt,name=rejectHint" json:"rejectHint"` //在Follower節點拒絕Leader節點的消息之后，會在該字段記錄一個Entry索引值供Leader節點 
Context          []byte      `protobuf:"bytes,12,opt,name=context" json:"context,omitempty"` // 消息攜帶的一些上下文信息。例如，該消息是否與Leader節點轉移相關 
XXX_unrecognized []byte      `json:"-"` 
} 

Message結構體相關的數據類型為 MessageType，MessageType 有 19 種。當然，并不是所有的消息類型都會用到上面定義的Message結構體中的所有字段，因此其中有些字段是Optinal的。

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

然后是 raft 算法的實現，node 結構體實現了 Node 接口，對etcd-raft模塊具體實現的一層封裝，方便上層模塊使用etcd-raft模塊。其定義如下：

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

下面我們來看看 raft StateMachine 的狀態機轉換，實際上就是 raft 算法中各種角色的轉換。每個 raft 節點，可能具有以下三種狀態中的一種。

• Candidate：候選人狀態，該狀態意味著將進行一次新的選舉。
• Follower：跟隨者狀態，該狀態意味著選舉結束。
• Leader：領導者狀態，選舉出來的節點，所有數據提交都必須先提交到 Leader 上。

每一個狀態都有其對應的狀態機，每次收到一條提交的數據時，都會根據其不同的狀態將消息輸入到不同狀態的狀態機中。同時，在進行 tick 操作時，每種狀態對應的處理函數也是不一樣的。因此 raft 結構體中將不同的狀態及其不同的處理函數，獨立出來幾個成員變量：

• state，保存當前節點狀態;
• tick 函數，每個狀態對應的 tick 函數不同;
• step，狀態機函數，同樣每個狀態對應的狀態機也不相同

我們接著看 etcd-raft 狀態轉換。etcd-raft StateMachine 封裝在 raft機構體中，etcd為了不讓entry落后的太多的直接進行選舉，多了一個其PreCandidate狀態，轉換如下圖：

raft 狀態轉換的接口都在 raft.go 中，其定義如下：

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

tick 函數，每個狀態對應的 tick 函數不同，下面分析兩個tick：

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

跟隨者、預選候選人、候選人、領導者 4 種節點狀態都有分別對應的功能函數，當需要查看各節點狀態相關的功能實現時(比如，跟隨者如何接收和處理日志)，都可以將對應的函數作為入口函數，來閱讀代碼和研究功能實現。

日志復制

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

(2)向當前節點的raftLog中追加Entry記錄。

(3)更新當前節點對應的Progress實例。

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

raft.appendEntry()方法的具體實現如下：

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指定為當前leader節點的任期號 
  es[i].Index = li + 1 + uint64(i) //為日志記錄指定的Index 
 } 
 li = r.raftLog.append(es...)//向raft中追加記錄 
    //更新當前節點對應的Progress,主要是更新Next和Match 
 r.prs.Progress[r.id].MaybeUpdate(li) 
    //嘗試提交Entry記錄 
 r.maybeCommit() 
 return true 
} 

在Progress.mayUpdate()方法中，會嘗試修改Match字段和Next字段，用來標識對應節點Entry記錄復制的情況。Leader節點除了在向自身raftLog中追加記錄時(即appendEntry()方法)會調用該方法，當Leader節點收到Follower節點的MsgAppResp消息(即MsgApp消息的響應消息)時，也會調用該方法嘗試修改Follower節點對應的Progress實例。Progress.MayUpdate()方法的具體實現如下：

func (pr *Progress) MaybeUpdate(n uint64) bool { 
 var updated bool 
 if pr.Match < n { 
  pr.Match = n //n之前所有的Entry記錄都已經寫入對應節點的raftLog中 
  updated = true 
        //下面將Progress.paused設置為false，表示leader節點可以繼續向對應Follower 
        //節點發送MsgApp消息 
  pr.ProbeAcked() 
 } 
 pr.Next = max(pr.Next, n+1)//將Next值加一，下一次復制Entry記錄開始的位置 
 return updated 
} 

如果該Entry記錄已經復制到了半數以上的節點中，則在raft.maybeCommit()方法中會嘗試將其提交。除了 appendEntry()方法，在 Leader 節點每次收到 MsgAppResp 消息時也會調用maybeCommit()方法，maybeCommit()方法的具體實現如下：

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分兩個組，JointConfig是大多數的組，有興趣的看一看quorum包的實現 
func (c JointConfig) CommittedIndex(l AckedIndexer) Index {//比較大多數的node的前倆個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 相關的所有處理都抽象為了 Message，通過 Step 接口處理各類消息的入口，首先根據Term"值"對消息進行分類處理，再根據消息的"類型"進行分類處理：

func (r *raft) Step(m pb.Message) error { 
 switch {//首先根據消息的Term值進行分類處理 
 case m.Term == 0://本地消息不做處理。MsgHup，MsgProp和MsgReadIndex是本地消息 
 case m.Term > r.Term: 
 case m.Term < r.Term://細節部分，可以自己研究源碼 
 } 
 switch m.Type {//根據Message的Type進行分類處理 
 case pb.MsgHup://這里針對MsgHup類型的消息進行處理。 
  if r.preVote {//檢查是不是開啟了preVote，如果是開啟了先調用raft.hup方法，發起preVote。 
  } else { 
   r.hup(campaignElection)//下面講述 
  } 
 case pb.MsgVote, pb.MsgPreVote: //對MsgVote，MsgPreVote類型的消息進行處理。 
  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://對于其他類型的消息處理，對應的node的step函數處理 
  err := r.step(r, m) 
  if err != nil { 
   return err 
  } 
 } 
 return nil 
} 

這里主要使用hup函數對Message來做處理，在raft.campaign()方法中，除了完成狀態切換，還會向集群中的其他節點發送相應類型的消息，例如，如果當前 Follower 節點要切換成 PreCandidate 狀態，則會發送 MsgPreVote 消息：

func (r *raft) hup(t CampaignType) { 
 if r.state == StateLeader {//忽略leader 
  return 
 } 
    //方法會檢查prs字段中是否還存在當前節點對應的Progress實例，這是為了監測當前節點是否被從集群中移除了 
    if !r.promotable() { 
  return 
 } 
    //獲取raftLog中已提交但未應用的Entry記錄，異常處理 
 ents, err := r.raftLog.slice(r.raftLog.applied+1, r.raftLog.committed+1, noLimit) 
 r.campaign(t) 
} 
func (r *raft) campaign(t CampaignType) { 
    //該方法的會發送一條包含Term值和類型 
 var term uint64 
 var voteMsg pb.MessageType 
 if t == campaignPreElection {//切換的目標狀態是Precandidate 
  r.becomePreCandidate() 
  voteMsg = pb.MsgPreVote 
        //確定要發送的Term值，這里只是增加了消息的Term值，并未增加raft.term字段的值 
  term = r.Term + 1 
 } else {//切換的目標狀態是Candidate 
  r.becomeCandidate() 
  voteMsg = pb.MsgVote 
        //給raft.Term字段的值，并將當前節點的選票投給自身 
  term = r.Term 
 } 
 if _, _, res := r.poll(r.id, voteRespMsgType(voteMsg), true); res == quorum.VoteWon { 
        //當得到足夠的選票時，則將PreCandidate狀態的節點切換成Candidate狀態 
        //Candidate狀態的節點則切換成Leader狀態 
  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 {//狀態切換完成之后，當前節點會向集群中所有節點發送指定類型的消息 
  if id == r.id { //跳過當前節點自身 
   continue 
  } 
        var ctx []byte 
        //在進行Leader節點轉移時，MsgPreVote或MsgVote消息會在Context字段中設置該特殊值 
  if t == campaignTransfer { 
   ctx = []byte(t) 
  } 
        //發送指定類型的消息，其中Index和LogTerm分別是當前節點的raftLog 
        //最后一條消息的Index值和Term值 
  r.send(pb.Message{Term: term, To: id, Type: voteMsg, Index: r.raftLog.lastIndex(), LogTerm: r.raftLog.lastTerm(), Context: ctx}) 
 } 
} 

Follower 節點在選舉計時器超時的行為：首先它會通過 tickElection()創建MsgHup消息并將其交給raft.Step()方法進行處理;raft.Step()方法會將當前Follower節點切換成PreCandidate狀態，然后創建MsgPreVote類型的消息，最后將該消息追加到raft.msgs字段中，等待上層模塊將其發送出去。

本文轉載自微信公眾號「運維開發故事」，可以通過以下二維碼關注。轉載本文請聯系運維開發故事公眾號。