SOFAStack Scalable Open Financial  Architecture Stack 是螞蟻金服自主研發的金融級分散式架構，包含了構建金融級雲原生架構所需的各個元件，是在金融場景裡錘鍊出來的最佳實踐。

本文為《剖析 | SOFAJRaft 實現原理》第二篇，本篇作者米麒麟，來自陸金所。《剖析 | SOFAJRaft 實現原理》系列由 SOFA 團隊和原始碼愛好者們出品，專案代號：SOFA:JRaftLab/，目前領取已經完成，感謝大家的參與。

SOFAJRaft 是一個基於 Raft 一致性演算法的生產級高效能 Java 實現，支援 MULTI-RAFT-GROUP，適用於高負載低延遲的場景。

SOFAJRaft ：https://gitee.com/sofastack/sofa-jraft

前言

SOFAJRaft-RheaKV 是基於 SOFAJRaft 和 RocksDB 實現的嵌入式、分散式、高可用、強一致的 KV 儲存類庫，SOFAJRaft 是基於 Raft 一致性演算法的生產級高效能 Java 實現，支援 Multi-Raft-Group。SOFAJRaft-RheaKV 叢集主要包括三個核心元件：PD，Store 和 Region。本文將圍繞 SOFAJRaft-RheaKV 架構設計，儲存概覽，核心模組，使用場景以及基於 Raft 實現等方面剖析 SOFAJRaft-RheaKV 基於 SOFAJRaft 實現原理，闡述如何使用 Raft 協議支援 KV 儲存類庫功能特性：

• SOFAJRaft-RheaKV 基礎架構如何設計？核心元件負責哪些功能？模組內部處理流程是怎樣？
• 基於 SOFAJRaft 如何使用 Raft 實現 SOFAJRaft-RheaKV 強一致性和自驅動等特性？ 

SOFAJRaft-RheaKV 概覽

SOFAJRaft-RheaKV 是一個輕量級的分散式的嵌入式的 KV 儲存 Library， RheaKV 包含在 SOFAJRaft 專案裡，是 SOFAJRaft 的子模組。SOFAJRaft-RheaKV 定位是嵌入式 jar 包方式嵌入到應用中，涵蓋以下功能特性：

• 強一致性，基於 Multi-Raft 分散式一致性協議保證資料可靠性和一致性；
• 自驅動，自診斷，自優化，自決策，自恢復；
• 可監控基於節點自動上報到 PD 的元資訊和狀態資訊；
• 基本 API get/put/delete 和跨分割槽 scan/batch put，distributed lock 等。

架構設計

SOFAJRaft-RheaKV 儲存類庫主要包括 PD，Store 和 Region 三個核心元件，支援輕量級的狀態/元資訊儲存以及叢集同步，分散式鎖服務使用場景：

• PD 是全域性的中心總控節點，負責整個叢集的排程管理，維護 RegionRouteTable 路由表。一個 PDServer 管理多個叢集，叢集之間基於 clusterId 隔離；PD Server 需要單獨部署，很多場景其實並不需要自管理，RheaKV 也支援不啟用 PD，不需要自管理的叢集可不啟用 PD，設定 PlacementDriverOptions 的 fake選項為 true 即可。
• Store 是叢集中的一個物理儲存節點，一個 Store 包含一個或多個 Region。
• Region 是最小的 KV 資料單元，可理解為一個數據分割槽或者分片，每個 Region 都有一個左閉右開的區間 [startKey, endKey)，能夠根據請求流量/負載/資料量大小等指標自動分裂以及自動副本搬遷。Region 有多個副本 Replication 構建 Raft Groups 儲存在不同的 Store 節點，通過 Raft 協議日誌複製功能資料同步到同 Group 的全部節點。

儲存設計

SOFAJRaft-RheaKV 儲存層為可插拔設計，實現 RawKVStore 儲存介面，目前 StoreEngine 儲存引擎支援 MemoryDB 和 RocksDB 兩種實現：

• MemoryRawKVStore：MemoryDB 基於 ConcurrentSkipListMap 實現，有更好的效能，但是單機儲存容量受記憶體限制；
• RocksRawKVStore：RocksDB 在儲存容量上只受磁碟限制，適合更大資料量的場景。

SOFAJRaft-RheaKV 儲存引擎基於 MemoryDB 和 RocksDB 實現 KV 儲存入口：

com.alipay.sofa.jraft.rhea.storage.RawKVStore
com.alipay.sofa.jraft.rhea.storage.MemoryRawKVStore
com.alipay.sofa.jraft.rhea.storage.RocksRawKVStore

SOFAJRaft-RheaKV 資料強一致性依靠 SOFAJRaft 同步資料到其他副本 Replication, 每個資料變更都會落地為一條 Raft 日誌, 通過 Raft 協議日誌複製功能將資料安全可靠地同步到同 Raft Group 的全部節點裡。

核心設計

SOFAJRaft-RheaKV 核心模組包括 KV 模組[RheaKVStore 基於 RegionRouteTable 路由表使用 RaftRawKVStore 儲存 KeyValue]，PD 模組[PlacementDriverServer 基於 StoreHeartbeat/RegionHeartbeat 心跳平衡節點分割槽 Leader 以及分裂]。

KV 模組內部處理

• RheaKVStore：最上層 User API，預設實現為 DefaultRheaKVStore， RheaKVStore 為純非同步實現，所以通常阻塞呼叫導致的客戶端出現瓶頸，理論上不會在 RheaKV 上遭遇，DefaultRheaKVStore 實現包括請求路由、Request 分裂、Response 聚合以及失敗重試等功能。
• PlacementDriverClient：非必須，作為與 PlacementDriver Server 叢集溝通的客戶端，通過它獲取叢集完整資訊包括但不僅限於"請求路由表"，對於無 PD 場景， RheaKV 提供 Fake PD Client。
• RegionRouteTable：分片邏輯基於 RegionRouteTable 路由表結構，最適合的資料結構便是跳錶或者二叉樹(最接近匹配項查詢)。作為本地路由表快取元件，RegionRouteTable 根據 KV 請求的具體失敗原因來決策是否從 PD Server 叢集重新整理資料，並且提供對單個 Key、多個 Key 列表以及 Key Range 進行計算返回對應的分割槽 ID。選擇 Region 的 StartKey 作為 RegionRouteTable 的 Key ，主要取決於 Region Split 的方式，父 Region 分裂成兩個子 Region 導致其中一個子 Region 的 StartKey 為 SplitKey。
• LoadBalancer：在提供 Follower 線性一致讀的配置下有效，目前僅支援 RR 策略。
• RheaKVRpcService：針對 KV 儲存服務的 RPC Client 客戶端封裝，實現 Failover 邏輯。
• RegionKVService：KV Server 服務端的請求處理服務，一個 StoreEngine 中包含很多 RegionKVService, 每個 RegionKVService 對應一個 Region，只處理本身 Region 範疇內的請求。
• MetricsRawKVStore：攔截請求做指標度量。
• RaftRawKVStore：RheaKV 的 Raft 入口，從這裡開始 Raft 流程。
• KVStoreStateMachine：實現 Raft 狀態機。
• RocksRawKVStore：原始的 RocksDB API 封裝， 目前 RheaKV 也支援可插拔的 MemoryDB 儲存實現。

PD 模組內部處理

PD 模組主要參考 TIKV 的設計理念，目前只實現自動平衡所有節點的分割槽 Leader 以及自動分裂。

• PlacementDriverClient -> MetadataClient：MetadataClient 負責從 PD 獲取叢集元資訊以及註冊元資訊。
• StoreEngine -> HeartbeatSender：
  1.  HeartbeatSender 負責傳送當前儲存節點的心跳，心跳中包含一些狀態資訊，心跳一共分為兩類：StoreHeartbeat 和 RegionHeartbeat；
  2. PD 不斷接受 RheaKV 叢集這兩類心跳訊息，PD 在對 Region Leader 的心跳回復裡面包含具體排程指令，再以這些資訊作為決策依據。除此之外，PD 還應該可以通過管理介面接收額外的運維指令，用來人為執行更準確的決策。
  3. 兩類心跳包含的狀態資訊詳細內容如下：
  • StoreHeartbeat 包括儲存節點 Store 容量，Region 數量，Snapshot 數量以及寫入/讀取資料量等 StoreStats 統計明細；
  • RegionHeartbeat 包括 Region 的 Leader 位置，掉線 Peer 列表，暫時不 Work 的 Follower 以及寫入/讀取資料量/Key 的個數等 RegionStats 統計明細。
• Pipeline：是針對心跳上報 Stats 的計算以及儲存處理流水線，處理單元 Handler 可插拔非常方便擴充套件。
• MetadataStore：負責叢集元資訊儲存以及查詢，儲存方面基於內嵌的 RheaKV。

SOFAJRaft-RheaKV 剖析

RheaKV 是基於 SOFAJRaft 實現的嵌入式、分散式、高可用、強一致的 KV 儲存類庫，TiKV 是一個分散式的 KV 系統，採用 Raft 協議保證資料的強一致性，同時使用 MVCC + 2PC 方式實現分散式事務的支援，兩者如何基於 Raft協議實現 KV 儲存？

RheaKV 基於 JRaft 實現

RaftRawKVStore 是 RheaKV 基於 Raft 複製狀態機 KVStoreStateMachine 的 RawKVStore 介面 KV 儲存實現，呼叫 applyOperation(kvOperation,kvStoreClosure) 方法根據讀寫請求申請指定 KVOperation 操作，申請鍵值操作處理邏輯：

1. 檢查當前節點的狀態是否為 STATE_LEADER，如果當前節點不是 Leader 直接失敗通知 Done Closure，通知失敗(NOT_LEADER)後客戶端重新整理 Leader 地址並且重試。Raft 分組 Leader 節點呼叫 Node#apply(task) 提交申請基於鍵值操作的任務到相應 Raft Group，向 Raft Group 組成的複製狀態機叢集提交新任務應用到業務狀態機，Raft Log 形成 Majority 後 StateMachine#onApply(iterator) 介面應用到狀態機的時候會被獲取呼叫。Node 節點構建申請任務日誌封裝成事件釋出回撥，釋出節點服務事件到佇列 applyQueue，依靠 Disruptor 的 MPSC 模型批量消費，對整體吞吐效能有著極大的提升。日誌服務事件處理器以單執行緒 Batch 攢批的消費方式批量執行鍵值儲存申請任務；
2. Raft 副本節點 Node 執行申請任務檢查當前狀態是否為 STATE_LEADER，必須保證 Leader 節點操作申請任務。迴圈遍歷節點服務事件判斷任務的預估任期是否等於當前節點任期，Leader 沒有發生變更的階段內提交的日誌擁有相同的 Term 編號，節點 Node 任期滿足預期則 Raft 協議投票箱 BallotBox 呼叫 appendPendingTask(conf, oldConf, done) 日誌複製之前儲存應用上下文，即基於當前節點配置以及原始配置建立選票 Ballot 新增到選票雙向佇列 pendingMetaQueue；
3. 日誌管理器 LogManager 呼叫底層日誌儲存 LogStorage#appendEntries(entries) 批量提交申請任務日誌寫入 RocksDB，用於 Leader 向 Follower 複製日誌包括心跳存活檢查等。日誌管理器釋出 Leader 穩定狀態回撥 LeaderStableClosure 事件到佇列 diskQueue 即 Disruptor 的 Ring Buffer，穩定狀態回撥事件處理器通過MPSC Queue 模型攢批消費觸發提交節點選票；
4. 投票箱 BallotBox 呼叫 commitAt(firstLogIndex, lastLogIndex, peerId) 方法提交當前 PeerId 節點選票到 Raft Group，更新日誌索引在[first_log_index, last_log_index]範疇。通過 Node#apply(task) 提交的申請任務最終將會複製應用到所有 Raft 節點上的狀態機，RheaKV 狀態機通過繼承 StateMachineAdapter 狀態機介面卡的 KVStoreStateMachine 表示；
5. Raft 狀態機 KVStoreStateMachine 呼叫 onApply(iterator) 方法按照提交順序應用任務列表到狀態機。當 onApply(iterator) 方法返回時認為此批申請任務都已經成功應用到狀態機上，假如沒有完全應用(比如錯誤、異常)將被當做 Critical 級別錯誤報告給狀態機的 onError(raftException) 方法，錯誤型別為 ERROR_TYPE_STATE_MACHINE。Critical 錯誤導致終止狀態機，為什麼這裡需要終止狀態機，非業務邏輯異常的話(比如磁碟滿了等 IO 異常)，代表可能某個節點成功應用到狀態機，但是當前節點卻應用狀態機失敗，是不是代表出現不一致的錯誤？ 解決辦法只能終止狀態機，需要手工介入重啟，重啟後依靠 Snapshot + Raft log 恢復狀態機保證狀態機資料的正確性。提交的任務在 SOFAJRaft 內部用來累積批量提交，應用到狀態機的是 Task迭代器，通過 com.alipay.sofa.jraft.Iterator 介面表示；
6. KVStoreStateMachine 狀態機迭代狀態輸出列表積攢鍵值狀態列表批量申請 RocksRawKVStore 呼叫 batch***(kvStates) 方法執行相應鍵值操作儲存到 RocksDB，為啥 Batch 批量儲存呢？ 刷盤常用伎倆，攢批刷盤優於多次刷盤。通過 RecycleUtil 回收器工具回收狀態輸出列表，其中KVStateOutputList 是 Pooled ArrayList 實現，RecycleUtil 用於釋放列表物件池化複用避免每次建立 List。

RheaKV 基於狀態機 KVStoreStateMachine 的 RaftRawKVStore 儲存 Raft 實現入口：

com.alipay.sofa.jraft.rhea.storage.RaftRawKVStore

RheaKV 執行在每個 Raft 節點上面的狀態機 KVStoreStateMachine 實現入口：

com.alipay.sofa.jraft.rhea.storage.KVStoreStateMachine 

RheaKV 是一個要保證線性一致性的分散式 KV 儲存引擎，所謂線性一致性，一個簡單的例子是在 T1 的時間寫入一個值，那麼在 T1 之後讀一定能讀到這個值，不可能讀到 T1 之前的值。因為 Raft 協議是為了實現分散式環境下面線性一致性的演算法，所以通過 Raft 非常方便的實現線性 Read，即將任何的讀請求走一次 Raft Log，等 Log 日誌提交之後在 apply 的時候從狀態機裡面讀取值，一定能夠保證此讀取到的值是滿足線性要求的。因為每次 Read 都需要走 Raft 流程，所以效能是非常的低效的，SOFAJRaft 實現 Raft 論文提到 ReadIndex 和 Lease Read 優化，提供基於 Raft 協議的 ReadIndex 演算法的更高效率的線性一致讀實現，ReadIndex 省去磁碟的開銷，結合 SOFAJRaft 的 Batch + Pipeline Ack + 全非同步機制大幅度提升吞吐。RaftRawKVStore 接收 get/multiGet/scan/getSequence 讀請求都使用 Node``#``readIndex``(``requestContext``, ``readIndexClosure``) 發起一次線性一致讀請求，當能夠安全讀取的時候傳入的 ReadIndexClosure 將被呼叫，正常情況從狀態機中讀取資料返回給客戶端，readIndex 讀取失敗嘗試應用鍵值讀操作申請任務於 Leader 節點的狀態機 KVStoreStateMachine，SOFAJRaft 保證讀取的線性一致性。線性一致讀在任何叢集內的節點發起，並不需要強制要求放到 Leader 節點上面，將請求雜湊到叢集內的所有節點上，降低 Leader 節點的讀取壓力。RaftRawKVStore 的 get 讀操作發起一次線性一致讀請求的呼叫：

// KV 儲存實現線性一致讀
public void get(final byte[] key, final boolean readOnlySafe, final KVStoreClosure closure) {
        if (!readOnlySafe) {
            this.kvStore.get(key, false, closure);
            return;
        }
  			// 呼叫 readIndex 方法，等待回撥執行
        this.node.readIndex(BytesUtil.EMPTY_BYTES, new ReadIndexClosure() {

            @Override
            public void run(final Status status, final long index, final byte[] reqCtx) {
                if (status.isOk()) {
                    // ReadIndexClosure 回撥成功，從 RawKVStore 呼叫 get 方法讀取最新資料返回
                    RaftRawKVStore.this.kvStore.get(key, true, closure);
                    return;
                }
                // 特殊情況譬如發生選舉讀請求失敗，嘗試申請 Leader 節點的狀態機
                RaftRawKVStore.this.readIndexExecutor.execute(() -> {
                    if (isLeader()) {
                        LOG.warn("Fail to [get] with 'ReadIndex': {}, try to applying to the state machine.", status);
                        // If 'read index' read fails, try to applying to the state machine at the leader node
                        applyOperation(KVOperation.createGet(key), closure);
                    } else {
                        LOG.warn("Fail to [get] with 'ReadIndex': {}.", status);
                        // Client will retry to leader node
                        new KVClosureAdapter(closure, null).run(status);
                    }
                });
            }
        });
}

TiKV 基於 Raft 實現

TiDB 是 PingCAP 公司設計的開源分散式 HTAP (Hybrid Transactional and Analytical Processing) 資料庫，TiDB 叢集主要包括三個核心元件：TiDB Server，PD Server 和 TiKV Server。TiKV Server 負責儲存資料，從外部看 TiKV 是一個分散式的提供事務的 Key-Value 儲存引擎。儲存資料的基本單位是 Region，每個 Region 負責儲存一個 Key Range（從 StartKey 到 EndKey 的左閉右開區間）的資料，每個 TiKV 節點負責多個 Region。TiKV 使用 Raft 協議做複製，保持資料的一致性和容災。副本以 Region 為單位進行管理，不同節點上的多個 Region 構成一個 Raft Group，互為副本。資料在多個 TiKV 之間的負載均衡由 PD 排程，這裡也是以 Region 為單位進行排程。TiKV 利用 Raft 來做資料複製，每個資料變更都會落地為一條 Raft 日誌，通過 Raft 的日誌複製功能，將資料安全可靠地同步到 Group 的多數節點。TiKV 整體架構包括 Placement Driver，Node，Store 以及 Region 元件：

• Placement Driver : Placement Driver (PD) 負責整個叢集的管理排程。
• Node : Node 可以認為是一個實際的物理機器，每個 Node 負責一個或者多個 Store。
• Store : Store 使用 RocksDB 進行實際的資料儲存，通常一個 Store 對應一塊硬碟。
• Region : Region 是資料移動的最小單元，對應的是 Store 裡面一塊實際的資料區間。每個 Region 有多個副本（Replica），每個副本位於不同的 Store ，而這些副本組成了一個 Raft group。

TiKV 使用 Raft 一致性演算法來保證資料的安全，預設提供的是三個副本支援，這三個副本形成了一個 Raft Group。當 Client 需要寫入 TiKV 資料的時候，Client 將操作傳送給 Raft Leader，在 TiKV 裡面稱做 Propose，Leader 將操作編碼成一個 Entry，寫入到自己的 Raft Log 裡面，稱做 Append。Leader 也會通過 Raft 演算法將 Entry 複製到其他的 Follower 上面，叫做 Replicate。Follower 收到這個 Entry 之後也會同樣進行 Append 操作，順帶告訴 Leader Append 成功。當 Leader 發現此 Entry 已經被大多數節點 Append，認為此 Entry 已經是 Committed 的，然後將 Entry 裡面的操作解碼出來，執行並且應用到狀態機裡面，叫做 Apply。TiKV 提供 Lease Read，對於 Read 請求直接發給 Leader，如果 Leader 確定自身的 Lease 沒有過期，那麼直接提供 Read 服務不用執行一次 Raft 流程。如果 Leader 發現 Lease 已經過期，就會強制執行一次 Raft 流程進行續租然後再提供 Read 服務。TiKV 是以 Region 為單位做資料的複製，也就是一個 Region 的資料儲存多個副本，將每一個副本叫做一個 Replica。Replica 之間是通過 Raft 來保持資料的一致，一個 Region 的多個 Replica 儲存在不同的節點上構成一個 Raft Group，其中一個 Replica 作為此 Group 的 Leader，其他的 Replica 作為 Follower。所有的讀和寫都是通過 Leader 進行，再由 Leader 複製給 Follower。

總結

本文圍繞 SOFAJRaft-RheaKV 架構儲存，模組流程以及基於 Raft 實現細節方面闡述 SOFAJRaft-RheaKV 基本原理，剖析 SOFAJRaft-RheaKV 如何使用 JRaft 一致性協議日誌複製功能保證資料的安全和容災，參考 TiKV 基於 Raft 演算法實現了分散式環境資料的強一致性。

系列閱讀

• 螞蟻金服生產級 Raft 演算法庫 SOFAJRaft 儲存模組剖析 | SOFAJRaft 實現原理
• 螞蟻金服開源 SOFAJRaft：生產級 Java Raft 演算法庫

歡迎參加 SOFAMeetup#2 上海站

SOFA Meetup #2 上海站《使用 SOFAStack 快速構建微服務》期待你的參與❤~

5 月 26 日，本週日，SOFAStack 開源核心成員集體出動。本期我們將側重於各個落地的實際場景進行架構解析。

分散式事務 Seata 詳解、與 Spring Cloud 生態的融合案例、使用 SOFAStack 快速構建微服務 Demo 實操、更有最新開源的《讓 AI 像 SQL 一樣簡單 — SQLFlow Demo 》首秀，週日不見不散~

戳連結即可報名：https://tech.antfin.com/communi