作者：京東物流郭益如

導讀

在分布式系統中，什么是拜占庭將軍問題？產生的場景和解決方案是什么？什么是 Raft 共識演算法？Raft 演算法是如何解決拜占庭將軍問題的？其核心原理和演算法邏輯是什么？除了 Raft，還有哪些共識演算法？共識問題作為分布式系統的一大難點和痛點，本文主要介紹了其產生的背景、原因，以及通用的 Raft 演算法解決方案，

01 拜占庭將軍問題

【分布式對等網路中的通信容錯問題，

在分布式計算中，不同的計算機通過通訊交換資訊達成共識按照一套協作策略行動，有時候，系統中的成員計算機可能出錯而發送錯誤的資訊，用于傳遞資訊的通訊網路也可能導致資訊損壞，使得網路中不同的成員關于全體協作的策略得出不同結論，從而破壞系統一致性，這就是拜占庭將軍問題，

拜占庭將軍問題被認為是容錯性問題中最難的問題型別之一，】

9 位將軍兵分 9 路去打仗，他們各自有權力觀測敵情并做出行動判斷 —— 進攻或撤退，他們必須行動一致，即所有軍隊一起進攻或者一起撤退，否則部分進攻部分撤退會造成災難性后果，

前提：

將軍之間只能通過信使互相聯系，每位將軍將自己的判斷發送給其他將軍，并接收其他將軍發送的判斷；
收到資訊的將軍綜合所有的判斷，當超過半數都選擇進攻時，就決定進攻，當超過半數都選擇撤退時就決定撤退；

問題是，將軍中間可能出現叛徒，他可能會選擇相反的結果進行通信（投票），也可能選擇性的發送資訊，叛徒要達成的目標是：

選擇性的發送資訊，欺騙某些將軍采取進攻的行動；
促成一個錯誤的決定，比如將軍們不希望進攻時進攻；
迷惑某些將軍，使得他們無法做出決定；

如果叛徒達成了其中之一，任何的攻擊結果都是注定要失敗的，只有完全達成一致的努力才能獲得勝利，

比如，可能 9 位將軍中有 8 位忠誠的將軍和一名叛徒，8 位將軍中 4 位選擇進攻，4 位選擇撤退，叛徒分別給選擇進攻的將軍發送進攻的資訊，給選擇撤退的將軍發送撤退資訊，這樣一來，在4 位選擇進攻的將軍看，共 5 位將軍選擇進攻，從而發起進攻；而在 4 位選擇撤退的將軍看，共 5 位將軍選擇撤退，從而發起撤退，這樣各個將軍的一致性就遭到了破壞，

并且，叛徒將軍可能會偽造其他將軍的身份發送信件；

拜占庭將軍問題描述的是，在存在資訊丟失的不可靠信道上試圖通過訊息傳遞的方式達到一致性是不可能的，在系統中除了存在的訊息延遲或不可送達故障外，還可能包括訊息篡改、節點處理例外等潛在性例外，

1.1 拜占庭容錯

在早期的解決方案中，一種是 “拜占庭容錯” ，它遵循“少數服從多數”的共識機制，即使出現了錯誤或偽造的資訊，只要有問題的將軍數量不到 1/3，仍可以達到“拜占庭容錯”，使整個系統便可以正常運作，

為什么是 1/3呢？

其原理是這樣的，假設將軍總數是 N，其中正直的將軍數量是 S，反叛的將軍數量是 T，那么 N=S+T；

為了保證即使反叛的將軍都不去投票也能產生最終的結果，那么 S 必須要超過半數，這種情況下，S 都做出相同的選擇，依然可以達成共識，即 S>T；

如果叛徒給一半支持進攻的將軍發送進攻資訊，給一半支持撤退的將軍發送撤退資訊，這種情況要保證也能產生最終的投票結果，則 X > S/2 + E；

綜合以上關系，可以得到：

N = S + T

X < S

X > S/2 + T

求解以上不等式，可以得到：

(N-T)/2 > T，即 N > 3T

所以要保證正直的將軍至少占所有將軍總數的 2/3，才有可能達成共識，

1.2 拜占庭演算法

拜占庭演算法是一種共識演算法，確定共識的原則，各個節點通過這個共識原則既可以保證一致性，又能保證基本的磁區容錯性，

共識是可容錯系統中的一個基本問題：即使面對故障，服務器如何在共享狀態上達成一致？

02 Raft演算法

理解，首先 MCube 會依據模板快取狀態判斷是否需要網路獲取最新模板，當獲取到模板后進行模板加載，加載階段會將產物轉換為視圖樹的結構，轉換完成后將通過運算式引擎決議運算式并取得正確的值，通過事件決議引擎決議用戶自定義事件并完成事件的系結，完成決議賦值以及事件系結后進行視圖的渲染，最終將目標頁面展示到螢屏，

【Raft 演算法解決的是簡化版的拜占庭將軍問題，即在不考慮資料丟失、篡改的情況下的拜占庭將軍問題，】

假設現在有 3 位將軍 A、B、C，將軍中沒有叛徒，信使的資訊可靠，但有可能被暗殺，此時將軍們如何達成進攻的一致性決定？

方案： Raft的方案是，在所有的將軍中選出一個大將軍，用來做出所有的決定，大將軍派信使給其他將軍，如果過一段時間沒有回復（可能被暗殺）就再派一個信使，直到收到回復，

如果大將軍的信使，派出去一個被干掉一個，其他將軍們總也收不到大將軍的資訊，他們如何達成一致性決定？

方案： 每位將軍都有一個隨機時間的計時器，時間一到，他就把自己當成大將軍的候選人，派信使將選舉結果給將軍 B、C，如果將軍 B、C 還沒有把選舉大將軍結果投給其他人（包括自己）時，他們就會把選舉票投給 A，A 將軍的信使回傳 A 時，A 將軍就知道自己收到了足夠的票數，成為了新的大將軍，

Raft 演算法是一種簡單易懂的共識演算法，它通過首先選舉一個 Leader 主節點，然后讓Leader 完全負責資料同步，依靠狀態機和主從同步的方式，在各個節點之間實作資料的一致性，

通過這種主節點進行資料同步的方式，Raft 將一致性問題拆分成了三個相對獨立的子問題：

1. 主節點選取 Leader Election： 啟動集群時，或者現有主節點失敗時，會啟動新的投票，獲得大多數選票（N/2+1）的節點會成為新的主節點；

2. 復制日志 Log Replication： 主節點從客戶端接收日志資訊，再把資訊復制到其他從節點上，使得日志資訊都能保持資料一致；

3. 安全性： Raft 定義了一系列規范來保證資料安全性，

2.1 Raft節點

Raft 演算法為節點定義了三種角色： Leader（主節點） 、Follower（從節點） 、Candidate（參與投票競爭的節點） ，節點的角色是可以轉換的，在任意的時間，每個服務器一定處于三種狀態中的一個，

每個節點上都有一個倒計時器（Election Timeout），隨機值在 150ms ~ 300ms 之間，當節點收到選舉請求，或收到 Leader 的 Heartbeat 時，就會重置倒計時，

2.1.1 主節點 Leader

通常情況下，系統中只有一個主節點，用來發起心跳，處理所有的客戶端請求，創建日志和同步日志，

2.1.2 從節點 Follower

除主節點外，其他的節點都是從節點，用于接收主節點的心跳和日志資料，保證其資料狀態與主節點一致，以及在 Leader 選舉時，投票給 Candidate，

如果有客戶端跟Follower 聯系，那么 Follower 會把請求重定向給 Leader，

2.1.3 候選人 Candidate

候選人 Candidate是在 Leader 選舉程序中的臨時角色，由 Follower 轉換而來，用于發起投票參與競選，

Raft 節點狀態圖：

圖1 Raft 節點狀態圖

啟動時，或 Follower 接收不到 Leader 資訊時，它就會變成 Candidate 并發起一次選舉，獲得集群中大多數選票的Candidate 就成為新的 Leader，

2.1.4 任期 Term

Raft 把時間分割成任意長度的任期 Term，用連續的整數標記，

圖2 各任期 Term 下的狀態演變

每一個任期都會開始一次新的選舉，一個或多個 Candidate 會嘗試成為 Leader，如果一個 Candidate 贏得了選舉，它就會在該任期內擔任 Leader，直到任期結束或者服務器宕機，在某些情況下，沒有選出 Leader（如選票瓜分等），則會開啟下一個任期并立刻開始新的選舉，

任期在 Raft 演算法中充當邏輯時鐘的作用，每一個節點都會存盤當前的 Term 號，這一編號在整個集群時期內單調增長，服務器之間通信的時候也會交換當前的 Term 號：

如果一個節點發現其 Term 號比其他服務器小，那么它會更新自己的 Term 號到較大的值；
如果一個 Candidate 或者 Leader 發現其 Term 號過期了，那么它會立即恢復成 Follower 狀態；
如果一個節點接收到的請求中 Term 號過期了，那么它會直接拒絕此次請求，

Raft 演算法中服務器節點之間通信使用遠程程序呼叫（RPCs），并且基本的一致性演算法只需要兩種型別的 RPCs，請求投票（RequestVote） RPCs 由候選人在選舉期間發起，然后附加條目（AppendEntries）RPCs 由領導者發起，用來復制日志和提供一種心跳機制，如果未及時收到回應，則請求者有責任重試 RPC，

2.1.5 事件 Entry

每一個事件是一個 Entry，只有 Leader 可以創建 Entry，結構為 <term, index, cmd>其中 cmd 是可以應用到狀態機的操作，

2.1.6 日志 Log

日志是 Raft 的核心概念，是一個由 Entry 構成的陣列，只有 Leader 才可以改變其他節點的 Log，Leader 先把 Entry 添加到自己的 Log 陣列中，發起共識請求，獲得同意后，才會將 Entry 提交給狀態機，Follower 只能從 Leader 中獲取新日志和當前的 CommitIndex，然后把對應的 Entry 應用到自己的狀態機中，

2.2 選取主節點 Leader Election

2.2.1 選舉機制

raft 通過心跳機制來觸發 Leader 的選舉；

Leader 會向所有的 Follower 周期性發送心跳來保證自己的 Leader 地位，

如果服務器能夠收到來自 Leader 或者 Candidate 的有效資訊，那么它會一直保持為 Follower 狀態，并且重繪自己的 electionElapsed，重新計時，

如果一個 Follower 在一個周期內沒有收到任何資訊，也就是選舉超時，它就會認為此時沒有可用的 Leader，開始進行一次選舉以選出一個新的 Leader，
當服務器啟動時，所有的節點都是 Follower，

2.2.2 選舉程序

Follower 自增的 term 號并且轉換狀態為 Candidate，然后他會向所有節點發起 RequestVoteRPC 請求， Candidate 的狀態會持續到以下情況發生：

獲得大多數選票（N/2 +1），贏得選舉，成為 Leader
其他節點贏得選舉
一輪選舉結束，無人勝出

在 Candidate 等待選票的時候，它可能收到其他節點宣告其是 Leader 的心跳：

該 Leader 的 term 號大于等于自己的 term 號，說明對方已經成為 Leader，則自己回退為 Follower，
該 Leader 的 term 號小于自己的 term 號，那么會拒絕該請求并讓該節點更新 term，

為了避免出現“腦裂”，即同一時刻出現多個 Candidate，導致沒有 Candidate 獲得大多數選票的狀況，Raft 增加了隨機選舉超時時間的方法，每一個Candidate 在發起選舉后，都會隨機化一個超時時間（ 150-300 毫秒），使得各個服務器分散開來，在大多數情況下只有一個服務器會率先超時，贏得選舉，

相關代碼實作：

【plain】
func (rf *Raft) RequestVote(request *RequestVoteRequest, response *RequestVoteResponse) {
  rf.mu.Lock()
  defer rf.mu.Unlock()
  defer rf.persist()
  defer DPrintf("{Node %v}'s state is {state %v,term %v,commitIndex %v,lastApplied %v,firstLog %v,lastLog %v} before processing requestVoteRequest %v and reply requestVoteResponse %v", rf.me, rf.state, rf.currentTerm, rf.commitIndex, rf.lastApplied, rf.getFirstLog(), rf.getLastLog(), request, response)




  if request.Term < rf.currentTerm || (request.Term == rf.currentTerm && rf.votedFor != -1 && rf.votedFor != request.CandidateId) {
    response.Term, response.VoteGranted = rf.currentTerm, false
    return
  }
  if request.Term > rf.currentTerm {
    rf.ChangeState(StateFollower)
    rf.currentTerm, rf.votedFor = request.Term, -1
  }
  if !rf.isLogUpToDate(request.LastLogTerm, request.LastLogIndex) {
    response.Term, response.VoteGranted = rf.currentTerm, false
    return
  }
  rf.votedFor = request.CandidateId
  rf.electionTimer.Reset(RandomizedElectionTimeout())
  response.Term, response.VoteGranted = rf.currentTerm, true
}


func (rf *Raft) StartElection() {
  request := rf.genRequestVoteRequest()
  DPrintf("{Node %v} starts election with RequestVoteRequest %v", rf.me, request)
  // use Closure
  grantedVotes := 1
  rf.votedFor = rf.me
  rf.persist()
  for peer := range rf.peers {
    if peer == rf.me {
      continue
    }
    go func(peer int) {
      response := new(RequestVoteResponse)
      if rf.sendRequestVote(peer, request, response) {
        rf.mu.Lock()
        defer rf.mu.Unlock()
        DPrintf("{Node %v} receives RequestVoteResponse %v from {Node %v} after sending RequestVoteRequest %v in term %v", rf.me, response, peer, request, rf.currentTerm)
        if rf.currentTerm == request.Term && rf.state == StateCandidate {
          if response.VoteGranted {
            grantedVotes += 1
            if grantedVotes > len(rf.peers)/2 {
              DPrintf("{Node %v} receives majority votes in term %v", rf.me, rf.currentTerm)
              rf.ChangeState(StateLeader)
              rf.BroadcastHeartbeat(true)
            }
          } else if response.Term > rf.currentTerm {
            DPrintf("{Node %v} finds a new leader {Node %v} with term %v and steps down in term %v", rf.me, peer, response.Term, rf.currentTerm)
            rf.ChangeState(StateFollower)
            rf.currentTerm, rf.votedFor = response.Term, -1
            rf.persist()
          }
        }
      }
    }(peer)
  }
}

2.3 日志同步 Log Replication

Raft 通過Leader 向集群中所有 Follower 進行日志同步來保證整個集群資料的最終一致性，

只有 Leader 有權限接受客戶端的請求并且同步資料給集群中其他節點，每一個客戶端的請求都包含一條需要被復制狀態機 RSM（Replicated State Mechine）執行的命令，Leader 收到客戶端請求后，會生成一個 Entry，包含，再將這個 entry 添加到自己的日志末尾后，向所有的節點廣播該 Entry，要求其他服務器復制這條 Entry，

圖3 主從節點進行 Entry 復制

如圖所示，Logs 日志是一個順序存盤的 Entry 陣列，方框內是任期 Term 號，

2.3.1 日志同步流程

例如，在 Term3 中，Leader 最后一個 Entry 的Index 為 7，x 值為 5，收到請求 set x=4時：

圖4 日志同步流程

Leader 收到客戶端請求 x←4 時，Leader 會生成一條新的 Entry<8, 3, set x=4>，并將該 Entry 添加到自己的 Log 陣列最后
Leader 通過 AppendEntries RPC 廣播該 Entry；
如果 Follower 接受該 Entry，則會將 Entry 添加到其日志后面，同時回傳給 Leader 同意，
如果 Leader 收到了多數的成功回應，Leader 認為這個 Entry 是 committed，應用到自己的狀態機 RSM 中，并且向客戶端回傳執行結果，之后，該 commited 資訊會隨著之后的 AppendEntryRPC 傳達到其他節點，
committed 表示被 Leader 創建的 Entry 已經復制到了大多數的服務器上，Leader 會跟蹤它記錄的最大索引值 Index，并在之后的 AppendEntries RPC（包括心跳）中，包含該索引值，以此確保其他服務器同步這個 Entry 已經提交，Follower 接收到該資訊后，也會按順序同步更新到本地的狀態機中，

Raft 通過這種日志機制來保證不同服務器上日志的一致性和安全性：

在兩個日志里，有兩個 entry 擁有相同的 index 和 term，那么它們一定有相同的 cmd；
在兩個日志里，有兩個 entry 擁有相同的 index 和 term，那么它們前面的 entry 也一定相同，

2.3.2 Leader Crash

一般情況下，Leader 和 Follower 的日志保持一致，AppendEntries 的一致性檢查通常不會失敗，然后，Leader Crash 可能會導致資料丟失：

圖5 Leader Crash時的資料狀況

當最上面的 Leader 掌權后，Follower 日志可能有 a~f 幾種情況：

1. 日志丟失（a，b）；

2. Follower 含有未提交資料（c、d）；

3. 日志丟失 + Follower 含有未提交資料（e、f）；

場景 f 可能出現的情況為：

如果一臺服務器在 Term2 時是 Leader，并且向它的日志中添加了一些資料條目，然后在資料提交前宕機了；接著該 Leader 很快重啟后，又稱為了任期 3 的 Leader，接著又向它的日志中添加了一些資料，然后在 Term2，Term3 資料條目提交前，又宕機了，之后一直處于宕機狀態，直到有新的 Leader 產生，

當遇到這種一致性檢查失敗的情況時，Leader 通過強制 Follower 復制自己的日志來處理日志的不一致，這就意味著，在 Follower 上的沖突日志會被領導者的日志覆寫，

Leader 找到 Follower 與它日志一致的地方（Index=3），然后洗掉 Follower 在該位置之后的日志，接著把這之后的日志發送給 Follower：

Leader 給每一個Follower 維護了一個 nextIndex，它表示 Leader 將要發送給該追隨者的下一條日志條目的索引，當一個 Leader 開始掌權時，它會將 nextIndex 初始化為它的最新的日志條目索引數+1，如果一個 Follower 的日志和 Leader 的不一致，AppendEntries 一致性檢查會在下一次 AppendEntries RPC 時回傳失敗，在失敗之后，Leader 會將 nextIndex 遞減然后重試 AppendEntries RPC，最終 nextIndex 會達到一個 Leader 和 Follower 日志一致的地方，這時，AppendEntries 會回傳成功，Follower 中沖突的日志條目都被移除了，并且添加所缺少的上了 Leader 的日志條目，一旦 AppendEntries 回傳成功，Follower 和 Leader 的日志就一致了，這樣的狀態會保持到該任期結束，

相關實作代碼：

【plain】
func (rf *Raft) replicateOneRound(peer int) {
  rf.mu.RLock()
  if rf.state != StateLeader {
    rf.mu.RUnlock()
    return
  }
  prevLogIndex := rf.nextIndex[peer] - 1
  if prevLogIndex < rf.getFirstLog().Index {
    // only snapshot can catch up
    request := rf.genInstallSnapshotRequest()
    rf.mu.RUnlock()
    response := new(InstallSnapshotResponse)
    if rf.sendInstallSnapshot(peer, request, response) {
      rf.mu.Lock()
      rf.handleInstallSnapshotResponse(peer, request, response)
      rf.mu.Unlock()
    }
  } else {
    // just entries can catch up
    request := rf.genAppendEntriesRequest(prevLogIndex)
    rf.mu.RUnlock()
    response := new(AppendEntriesResponse)
    if rf.sendAppendEntries(peer, request, response) {
      rf.mu.Lock()
      rf.handleAppendEntriesResponse(peer, request, response)
      rf.mu.Unlock()
    }
  }
}


func (rf *Raft) AppendEntries(request *AppendEntriesRequest, response *AppendEntriesResponse) {
  rf.mu.Lock()
  defer rf.mu.Unlock()
  defer rf.persist()
  defer DPrintf("{Node %v}'s state is {state %v,term %v,commitIndex %v,lastApplied %v,firstLog %v,lastLog %v} before processing AppendEntriesRequest %v and reply AppendEntriesResponse %v", rf.me, rf.state, rf.currentTerm, rf.commitIndex, rf.lastApplied, rf.getFirstLog(), rf.getLastLog(), request, response)


  if request.Term < rf.currentTerm {
    response.Term, response.Success = rf.currentTerm, false
    return
  }


  if request.Term > rf.currentTerm {
    rf.currentTerm, rf.votedFor = request.Term, -1
  }


  rf.ChangeState(StateFollower)
  rf.electionTimer.Reset(RandomizedElectionTimeout())


  if request.PrevLogIndex < rf.getFirstLog().Index {
    response.Term, response.Success = 0, false
    DPrintf("{Node %v} receives unexpected AppendEntriesRequest %v from {Node %v} because prevLogIndex %v < firstLogIndex %v", rf.me, request, request.LeaderId, request.PrevLogIndex, rf.getFirstLog().Index)
    return
  }


  if !rf.matchLog(request.PrevLogTerm, request.PrevLogIndex) {
    response.Term, response.Success = rf.currentTerm, false
    lastIndex := rf.getLastLog().Index
    if lastIndex < request.PrevLogIndex {
      response.ConflictTerm, response.ConflictIndex = -1, lastIndex+1
    } else {
      firstIndex := rf.getFirstLog().Index
      response.ConflictTerm = rf.logs[request.PrevLogIndex-firstIndex].Term
      index := request.PrevLogIndex - 1
      for index >= firstIndex && rf.logs[index-firstIndex].Term == response.ConflictTerm {
        index--
      }
      response.ConflictIndex = index
    }
    return
  }


  firstIndex := rf.getFirstLog().Index
  for index, entry := range request.Entries {
    if entry.Index-firstIndex >= len(rf.logs) || rf.logs[entry.Index-firstIndex].Term != entry.Term {
      rf.logs = shrinkEntriesArray(append(rf.logs[:entry.Index-firstIndex], request.Entries[index:]...))
      break
    }
  }


  rf.advanceCommitIndexForFollower(request.LeaderCommit)


  response.Term, response.Success = rf.currentTerm,True
}

2.3.3 安全性

Leader 需要保證其存盤全部已經提交的日志條目，保證日志條目只能從 Leader 流向 Follower，且 Leader 永遠不會覆寫已經存在的日志條目，

每個 Candidate 發送 RequestVoteRPC 時，都會帶上最后一個 Entry 的資訊，所有節點收到投票資訊時，會對該 Entry 進行比較，如果發現自己的更新，則拒絕投票給該 Candidate，

03 其他一致性演算法

3.1 Paxos 演算法

早期的共識演算法，由拜占庭將軍問題的提出者 Leslie Lamport 所發明，谷歌的分布式鎖服務 Chubby 就是以 Paxos 演算法為基礎，

3.2 ZAB 演算法

Zookeeper 所使用的一致性演算法，在流程上和 Raft 演算法比較接近，

3.3 PBFT 演算法

區塊鏈技術所使用的共識演算法之一，適用于私有鏈的共識，

04 總結

Raft 演算法是很廣泛的強一致性、去中心化和高可用的分布式協議，是一種 leader-based 的共識演算法，通過將共識問題拆分成主節點選舉和主從日志同步，以及安全流程，來提高分布式系統的資料一致性、可靠性和容錯性；首先選舉主節點，然后主節點負責接收外部請求、資料復制、提交，保證系統中資料都是一致的，

轉載請註明出處，本文鏈接：https://www.uj5u.com/ruanti/544228.html

標籤：架構設計

上一篇：多執行緒并發：以AQS中acquire()方法為例來分析多執行緒間的同步與協作

下一篇：拜占庭將軍問題和 Raft 共識演算法講解