淺談分佈式一致性：Raft 與 SOFAJRaft

簡介： 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 自身的功能來實現。

原文連接本文爲阿里雲原創內容，未經容許不得轉載。