淺談分布式一緻性：Raft 與 SOFAJRaft

作者 | 家純

來源 | 阿裡技術公衆号

一 分布式共識算法 (Consensus Algorithm)

1 如何了解分布式共識?

多個參與者針對某一件事達成完全一緻：一件事，一個結論。

已達成一緻的結論，不可推翻。

2 有哪些分布式共識算法?

• Paxos：被認為是分布式共識算法的根本，其他都是其變種，但是 paxos 論文中隻給出了單個提案的過程，并沒有給出複制狀态機中需要的 multi-paxos 的相關細節的描述，實作 paxos 具有很高的工程複雜度（如多點可寫，允許日志空洞等）。
• Zab：被應用在 zookeeper 中，業界使用廣泛，但沒用抽象成通用 library。
• Raft：以容易了解著稱，業界也湧現出很多 raft 實作，比如 etcd、braft、tikv 等。

二 Raft 介紹

1 特點：Strong Leader

• 系統中必須存在且同一時刻隻能有一個 leader，隻有 leader 可以接受 clients 發過來的請求。
• Leader 負責主動與所有 followers 通信，負責将“提案”發送給所有followers，同時收集多數派的 followers 應答。
• Leader 還需向所有 followers 主動發送心跳維持上司地位（保持存在感）。

另外，身為 leader 必須保持一直 heartbeat 的狀态。

2 複制狀态機

對于一個無限增長的序列a[1, 2, 3…]，如果對于任意整數i， a[i]的值滿足分布式一緻性， 這個系統就滿足一緻性狀态機的要求。

基本上所有的真實系統都會有源源不斷的操作，這時候單獨對某個特定的值達成一緻顯然是不夠的。為了讓真實系統保證所有的副本的一緻性，通常會把操作轉化為 write-ahead-log(WAL)。然後讓系統中所有副本對 WAL 保持一緻，這樣每個副本按照順序執行 WAL 裡的操作，就能保證最終的狀态是一緻的。

• Client 向 leader 發送寫請求。
• Leader 把“操作”轉化為 WAL 寫本地 log 的同時也将 log 複制到所有 followers。
• Leader 收到多數派應答，将 log 對應的“操作”應用到狀态機。
• 回複 client 處理結果。

3 Raft 中的基本概念

Raft-node 的 3 種角色/狀态

• Follower：完全被動，不能發送任何請求, 隻接受并響應來自 leader 和 candidate 的 message, node啟動後的初始狀态必須是 follower。
• Leader：處理所有來自用戶端的請求，以及複制 log 到所有 followers。
• Candidate：用來競選一個新 leader (candidate 由 follower 觸發逾時而來)。

Message 的 3 種類型

• RequestVote RPC：Candidate 發出。
• AppendEntries (Heartbeat) RPC：Leader 發出。
• InstallSnapshot RPC：Leader 發出。

任期邏輯時鐘

• 時間被劃分為一個個任期(term)，term id 按時間軸單調遞增。
• 每一個任期的開始都是 leader 選舉，選舉成功之後，leader在任期内管理整個叢集, 也就是“選舉 + 正常操作”。
• 每個任期最多一個 leader，可以沒有 leader (spilt-vote 導緻)。

4 Raft 功能分解

Leader 選舉

逾時驅動：Heartbeat / Election timeout

随機的逾時時間：降低選舉碰撞導緻選票被瓜分的機率

選舉流程：Follower --> Candidate (選舉逾時觸發)

• 赢得選舉：Candidate --> Leader
• 另一個節點赢得選舉：Candidate --> Follower
• 一段時間内沒有任何節點器赢得選舉：Candidate --> Candidate

選舉動作：

• Current term++
• 發送 RequestVote RPC

New Leader 選取原則 (最大送出原則)

• Candidates include log info in RequestVote RPCs(index & term of last log entry)
• During elections, choose candidate with log most likely to contain all committed entries
• Voting server V denies vote if its log is “more complete”：(lastTermV > lastTermC) ||((lastTermV == lastTermC) && (lastIndexV > lastIndexC))
• Leader will have “most complete” log among electing majority

安全性：一個 term，最多選出一個 leader，可以沒 leader，下一個 term 再選。

影響 raft 選舉成功率的幾個時間參數

• RTT(Round Trip Time)：網絡延時
• Heartbeat timeout：心跳間隔，通常應該比 election timeout 小一個數量級，目的是讓 leader 能夠持續發送心跳來阻止 followers 觸發選舉
• Election timeout：Leader 與 followers 間通信逾時觸發選舉的時間
• MTBF(Meantime Between Failure)：Servers 連續正常故障時間間隔 RTT << Heartbeat timeout < Election timeout(ET) << MTBF

随機選主觸發時間：Random(ET, 2ET)

日志複制

Raft 日志格式

• (TermId, LogIndex, LogValue)
• 其中 (TermId, LogIndex) 能确定唯一一條日志

Log replication關鍵點

• 連續性：日志不允許出現空洞
• 有效性：
  • 不同節點，擁有相同 term 和 logIndex 的日志 value 一定相同
  • Leader 上的日志一定是有效的
  • Follower 上的日志是否有效，通過 leader 日志對比判斷 (How?)

Followers 日志有效性檢查

• AppendEntries RPC 中還會攜帶前一條日志的唯一辨別 (prevTermId, prevLogIndex)
• 遞歸推導

Followers 日志恢複

• Leader 将 nextIndex 遞減并重發 AppendEntries，直到與 leader 日志一緻

Commit Index 推進

CommitIndex (TermId, LogIndex)

• 所謂 commitIndex，就是已達成多數派，可以應用到狀态機的最新的日志位置
• 日志被複制到 followers 後，先持久化，并不能馬上被應用到狀态機
• 隻有 leader 知道日志是否達成多數派，是否可以應用到狀态機
• Followers 記錄 leader 發來的目前 commitIndex，所有小于等于 commitIndex 的日志均可以應用到狀态機

CommitIndex推進

• Leader 在下一個 AppendEntries RPC (也包括 Heartbeat)中攜帶目前的 commitIndex
• Followers 檢查日志有效性通過則接受 AppendEntries 并同時更新本地 commitIndex, 最後把所有小于等于 commitIndex 的日志應用到狀态機

AppendEntries RPC

• 完整資訊：(currentTerm, logEntries[], prevTerm, prevLogIndex, commitTerm, commitLogIndex)
• currentTerm, logEntries[]：日志資訊，為了效率，日志通常為多條
• prevTerm, prevLogIndex：日志有效性檢查
• commitTerm, commitLogIndex：最新的送出日志位點(commitIndex)

階段小結：現在我們能用 raft 做什麼?

• 連續确定多個提案，確定叢集中各個系統節點狀态完全一緻
• 自動選主，保證在隻有少數派當機的情況下持續可用
• 日志強同步，當機後零資料丢失

三 SOFAJRaft

一個純 Java 的 raft 算法實作庫，使用 Java 重寫了所有功能，并有一些改進和優化。

1 SOFAJRaft 整體功能

功能支援

Leader election：選主。

Log replication and recovery：日志複制和日志恢複，log recovery就是要保證已經被 commit 的資料一定不會丢失，log recovery 包含兩個方面

• Current term 日志恢複，主要針對一些 follower 節點重新開機加入叢集或者是新增 follower 節點
• Prev term 日志恢複，主要針對 leader 切換前後的日志一緻性

Snapshot and log compaction：定時生成 snapshot，實作 log compaction加速啟動和恢複，以及InstallSnapshot 給 followers 拷貝資料。

Membership change：叢集線上配置變更，增加節點、删除節點、替換節點等。

Transfer leader：主動變更 leader，用于重新開機維護，leader 負載平衡等。

Symmetric network partition tolerance：對稱網絡分區容忍性。

Pre-Vote：如上圖 S1 為目前 leader，網絡分區造成 S2 不斷增加本地 term，為了避免網絡恢複後S2發起選舉導緻正在良心工作的 leader step-down, 進而導緻整個叢集重新發起選舉，在 request-vote 之前會先進行 pre-vote(currentTerm + 1，lastLogIndex, lastLogTerm)，多數派成功後才會轉換狀态為 candidate 發起真正的 request-vote，是以分區後的節點，pre-vote不會成功，也就不會導緻叢集一段時間内無法正常提供服務。

Asymmetric network partition tolerance：非對稱網絡分區容忍性。

如上圖 S1 為目前 leader，S2 不斷逾時觸發選主，S3 提升 term 打斷目前 lease，進而拒絕 leader 的更新，這個時候可以增加一個 trick 的檢查，每個 follower 維護一個時間戳記錄收到 leader 上資料更新的時間(也包括心跳)，隻有超過 election timeout 之後才允許接受 request-vote 請求。

Fault tolerance: 容錯性，少數派故障，不影響系統整體可用性：

• 機器掉電
• 強殺應用
• 慢節點(GC, OOM等)
• 網絡故障
• 其他各種奇葩原因導緻 raft 節點無法正常工作

Workaround when quorate peers are dead：多數派故障時整個 grop 已不具備可用性, 安全的做法是等待多數節點恢複，隻有這樣才能保證資料安全，但是如果業務更追求可用性，放棄資料一緻性的話可以通過手動 reset_peers 指令迅速重建整個叢集，恢複叢集可用。

Metrics：SOFAJRaft 内置了基于 metrics 類庫的性能名額統計，具有豐富的性能統計名額。

Jepsen：除了單元測試之外，SOFAJRaft 還使用 jepsen 這個分布式驗證和故障注入測試架構模拟了很多種情況，都已驗證通過：

• 随機分區，一大一小兩個網絡分區
• 随機增加和移除節點
• 随機停止和啟動節點
• 随機 kill -9 和啟動節點
• 随機劃分為兩組，互通一個中間節點，模拟分區情況
• 随機劃分為不同的 majority 分組

性能優化

Batch：SOFAJRaft 中整個鍊路都是 batch 的，依靠 disruptor 中的 MPSC 模型批量消費，包括但不限于：

• 批量送出 task
• 批量網絡發送
• 本地 IO batch 寫入，要保證日志不丢，一般每一條 log entry 都要進行 fsync, 比較耗時，SOFAJRaft 中做了合并寫入的優化
• 批量應用到狀态機

Replication pipeline：流水線複制，leader 跟 followers 節點的 log 同步是串行 batch 的方式，每個 batch 發送之後需要等待 batch 同步完成之後才能繼續發送下一批(ping-pong), 這樣會導緻較長的延遲。可以通過 leader 跟 followers 節點之間的 pipeline 複制來改進，有效降低更新的延遲, 提高吞吐。

Append log in parallel：Leader 持久化 log entries 和向 followers 發送 log entries 是并行的。

Fully concurrent replication：Leader 向所有 follwers 發送 log 也是完全并發的。

Asynchronous：Jraft 中整個鍊路幾乎沒有任何阻塞，完全異步的，是一個 callback 程式設計模型。

ReadIndex：優化 raft read 走 raft log 的性能問題，每次 read，僅記錄 commitIndex，然後發送所有 peers heartbeat 來确認 leader 身份，如果 leader 身份确認成功，等到 applied index >= commitIndex，就可以傳回 client read 了，基于 ReadIndex 可以很友善的提供線性一緻讀，不過 commitIndex 是需要從 leader 那裡擷取的，多了一輪RPC。

Lease Read：通過租約(lease)保證 leader 的身份，進而省去了 readIndex 每次 heartbeat 确認 leader 身份，性能更好, 但是通過時鐘維護 lease 本身并不是絕對的安全(jraft 中預設配置是 readIndex，因為 readIndex 性能已足夠好)。

2 SOFAJRaft 設計

SOFAJRaft - Raft Node

Node：Raft 分組中的一個節點，連接配接封裝底層的所有服務，使用者看到的主要服務接口，特别是 apply(task) 用于向 raft group 組成的複制狀态機叢集送出新任務應用到業務狀态機。

存儲

• Log 存儲，記錄 raft 使用者送出任務的日志，将從 leader 複制到其他節點上。LogStorage 是存儲實作, LogManager 負責對底層存儲的調用，對調用做緩存、批量送出、必要的檢查和優化。
• Metadata 存儲，元資訊存儲，記錄 raft 實作的内部狀态，比如目前 term、投票給哪個節點等資訊。
• Snapshot 存儲，用于存放使用者的狀态機 snapshot 及元資訊，可選. SnapshotStorage 用于 snapshot 存儲實作，SnapshotExecutor 用于 snapshot 實際存儲、遠端安裝、複制的管理。

狀态機

• StateMachine：使用者核心邏輯的實作，核心是 onApply(Iterator) 方法，應用通過 Node#apply(task) 送出的日志到業務狀态機。
• FSMCaller：封裝對業務 StateMachine 的狀态轉換的調用以及日志的寫入等，一個有限狀态機的實作, 做必要的檢查、請求合并送出和并發處理等。

複制

• Replicator：用于 leader 向 followers 複制日志，也就是 raft 中的 AppendEntries 調用，包括心跳存活檢查等。
• ReplicatorGroup：用于單個 raft group 管理所有的 replicator，必要的權限檢查和派發。

RPC 子產品用于節點之間的網絡通訊

• RPC Server：内置于 Node 内的 RPC 伺服器，接收其他節點或者用戶端發過來的請求, 轉交給對應服務處理。
• RPC Client：用于向其他節點發起請求，例如投票、複制日志、心跳等。

KV Store：SOFAJRaft 隻是一個 lib，KV Store 是 SOFAJRaft 的一個典型的應用場景，把它放進圖中以便更好的了解 SOFAJRaft。

SOFAJRaft - Raft Group

SOFAJRaft - Multi Raft Group

3 SOFAJRaft 實作細節

高效的線性一緻讀

什麼是線性一緻讀?

所謂線性一緻讀，一個簡單的例子就是在 t1 的時刻我們寫入了一個值, 那麼在 t1 之後， 我們一定能讀到這個值，不可能讀到 t1 之前的舊值 (想想 java 中的 volatile 關鍵字，說白了線性一緻讀就是在分布式系統中實作 volatile 語義)。

上圖Client A、B、C、D均符合線性一緻讀，其中 D 看起來是 stale read，其實并不是， D 請求橫跨了3個階段，而讀可能發生在任意時刻，是以讀到 1 或 2 都行。

重要：接下來的讨論均基于一個大前提，就是業務狀态機的實作必須是滿足線性一緻性的, 簡單說就是也要具有 java volatile 的語義。

1）直接點，是否可以直接從目前 leader 節點讀？

怎麼确定目前的 leader 真的是 leader(網絡分區)？

2）最簡單的實作方式：讀請求走一遍 raft 協定

有什麼問題?

• 不僅有日志寫盤開銷，還有日志複制的 RPC 開銷，在讀比重較大的系統中是無法接受的
• 還多了一堆的 raft “讀日志”

3）ReadIndex Read

這是 raft 論文中提到過的一種優化方案，具體來說：

• 将目前自己 log 的 commit index 記錄到一個 local 變量 ReadIndex 裡面。
• 向其他節點發起一次 heartbeat，如果大多數節點傳回了對應的 heartbeat response，那麼 leader 就能夠确定現在自己仍然是 leader (證明了自己是自己)。
• Leader 等待自己的狀态機執行，直到 apply index 超過了 ReadIndex，這樣就能夠安全的提供 Linearizable Read 了, 也不必管讀的時刻是否 leader 已飄走 (思考：為什麼需要等到 apply index 超過了 ReadIndex 才可以執行讀請求？)。
• Leader 執行 read 請求，将結果傳回給 Client。

通過ReadIndex，也可以很容易在 followers 節點上提供線性一緻讀：

• Follower 節點向 leader 請求最新的 ReadIndex。
• Leader執行上面 i ~ iii 的過程(确定自己真的是 leader)，并傳回 ReadIndex 給 follower。
• Follower 等待自己的 apply index 超過了 ReadIndex (有什麼問題？慢節點？)。
• Follower 執行 read 請求，将結果傳回給 client。

ReadIndex小結：

• 相比較于走 raft log 的方式，ReadIndex 讀省去了磁盤的開銷，能大幅度提升吞吐，結合 SOFAJRaft 的 batch + pipeline ack + 全異步機制，三副本的情況下 leader 讀的吞吐接近于 RPC 的上限。
• 延遲取決于多數派中最慢的一個 heartbeat response，理論上對于降低延時的效果不會非常顯著。

4）Lease Read

Lease read 與 ReadIndex 類似，但更進一步，不僅省去了 log，還省去了網絡互動。它可以大幅提升讀的吞吐也能顯著降低延時。

基本的思路是 leader 取一個比 election timeout 小的租期(最好小一個數量級)，在租約期内不會發生選舉，這就確定了 leader 不會變，是以可以跳過 ReadIndex 的第二步， 也就降低了延時。可以看到, Lease read 的正确性和時間是挂鈎的，是以時間的實作至關重要，如果漂移嚴重，這套機制就會有問題。

實作方式：

• 定時 heartbeat 獲得多數派響應, 确認 leader 的有效性 (在 SOFAJRaft 中預設的 heartbeat 間隔是 election timeout 的十分之一)。
• 在租約有效時間内，可以認為目前 leader 是 raft group 内的唯一有效 leader，可忽略 ReadIndex 中的 heartbeat 确認步驟(2)。
• Leader 等待自己的狀态機執行，直到 apply index 超過了 ReadIndex，這樣就能夠安全的提供 Linearizable Read 了。

5）更進一步：Wait Free

到此為止 lease 省去了 ReadIndex 的第 2 步(heartbeat)，實際上還能再進一步，省去第 3 步。

我們想想前面的實作方案的本質是什麼? 目前節點的狀态機達到“讀”這一刻的時間點 相同或者更新的狀态。

那麼更嚴格一點的限制就是：目前時刻，目前節點的狀态機就是最新的。

問題來了，leader 節點的狀态機能保證一定是最新的嗎？

• 首先 leader 節點的 log 一定是最新的，即使新選舉産生的 leader，它也一定包含全部的 commit log，但它的狀态機卻可能落後于舊的 leader。
• 但是在 leader 應用了自己目前 term 的第一條 log 之後，它的狀态機就一定是最新的。
• 是以可以得出結論：當 leader 已經成功應用了自己 term 的第一條 log 之後，不需要再取 commit index，也不用等狀态機，直接讀，一定是線性一緻讀。

小結：Wait Free 機制将最大程度的降低讀延遲，SOFAJRaft 暫未實作 wait free 這一優化，不過已經在計劃中。

在 SOFAJRaft 中發起一次線性一緻讀請求：

// KV 存儲實作線性一緻讀
public void readFromQuorum(String key, AsyncContext asyncContext) {
    // 請求 ID 作為請求上下文傳入
    byte[] reqContext = new byte[4];
    Bits.putInt(reqContext, 0, requestId.incrementAndGet());
    // 調用 readIndex 方法, 等待回調執行
    this.node.readIndex(reqContext, new ReadIndexClosure() {

        @Override
        public void run(Status status, long index, byte[] reqCtx) {
            if (status.isOk()) {
                try {
                    // ReadIndexClosure 回調成功, 可以從狀态機讀取最新資料傳回
                    // 如果你的狀态實作有版本概念, 可以根據傳入的日志 index 編号做讀取
                    asyncContext.sendResponse(new ValueCommand(fsm.getValue(key)));
                } catch (KeyNotFoundException e) {
                    asyncContext.sendResponse(GetCommandProcessor.createKeyNotFoundResponse());
                }
            } else {
                // 特定情況下, 比如發生選舉, 該讀請求将失敗
                asyncContext.sendResponse(new BooleanCommand(false, status.getErrorMsg()));
            }
        }
    });
}           

四 SOFAJRaft 應用場景

1 SOFAJRaft 可以做什麼

• 選舉
• 分布式鎖服務，比如 zookeeper
• 高可靠的元資訊管理

分布式存儲系統，如分布式消息隊列、分布式檔案系統、分布式塊系統等等。

2 使用者案例

• AntQ Streams QCoordinator：使用 SOFAJRaft 在 coordinator 叢集内做選舉、元資訊存儲等功能。
• Schema Registry：高可靠 schema 管理服務，類似 kafka schema registry。
• SOFA 服務注冊中心元資訊管理子產品：IP 資料資訊注冊，要求寫資料達到各個節點一緻， 并且在少數派節點挂掉時保證不影響資料正常存儲。
• RheaKV：基于 SOFAJRaft 和 rocksDB 實作的嵌入式、分布式、高可用、強一緻的 KV 存儲類庫。

3 簡單實踐：基于 SOFAJRaft 設計一個簡單的 KV Store

到目前為止，我們似乎還沒看到 SOFAJRaft 作為一個 lib 有什麼特别之處, 因為 SOFAJRaft 能辦到的 zk，etcd 似乎基本上也都可以辦到, 那麼 SOFAJRaft 算不算重複造輪子？

為了說明 SOFAJRaft 具有很好的想象空間以及擴充能力，下面再介紹一個基于 SOFAJRaft 的複雜一些的實踐。

4 複雜一點的實踐：基于 SOFAJRaft 的 Rhea KV 的設計

功能名詞

• PD：全局的中心總控節點, 負責整個叢集的排程, 不需要自管理的叢集可不啟用 PD (一個PD可管理多個叢集，基于 clusterId 隔離)。
• Store：叢集中的一個實體存儲節點，一個 store 包含一個或多個 region。
• Region：最小的 KV 資料單元，每個 region 都有一個左閉右開的區間 [startKey, endKey)，可根據請求流量/負載/資料量大小等名額自動分裂以及自動副本搬遷。

特點

• 嵌入式
• 強一緻性
• 自驅動：自診斷，自優化，自決策，自恢複。以上幾點(尤其2, 3)基本都是依托于 SOFAJRaft 自身的功能來實作。

招聘

我們是螞蟻智能監控技術中台的存儲團隊，我們正在使用 Rust/Go/Java 建構高性能、低成本具備實時分析能力的新一代時序資料庫，歡迎轉崗或者推薦，聯系人家純：[email protected]

參考資料 https://raft.github.io/ https://raft.github.io/slides/raftuserstudy2013.pdf https://github.com/brpc/braft/blob/master/docs/cn/raft_protocol.md https://pingcap.com/blog-cn/linearizability-and-raft/ https://aphyr.com/posts/313-strong-consistency-models https://zhuanlan.zhihu.com/p/51063866 Github 代碼倉庫： https://github.com/sofastack/sofa-jraft

免費領取電子書

《阿裡雲技術面試紅寶書》

本書公開30道阿裡雲技術面試真題，含雲原生、大資料、IoT、資料庫等領域，精準回顧相關知識點及考察要點，間接地與技術大牛們學習，溫故知新！

掃碼加阿裡妹好友，回複“紅寶書”擷取吧~(若掃碼無效，可直接添加alimei4、alimei5、alimei6、alimei7)