編寫你的第一個 Java 版 Raft 分佈式 KV 存儲

前言

本文旨在講述如何使用 Java 語言實現基於 Raft 算法的，分佈式的，KV 結構的存儲項目。該項目的背景是爲了深刻理解 Raft 算法，從而深入理解分佈式環境下數據強一致性該如何實現；該項目的目標是：在複雜的分佈式環境中，多個存儲節點可以保證數據強一致性。

項目地址：github.com/stateIs0/lu…

歡迎 star ：）

什麼是 Java 版 Raft 分佈式 KV 存儲

Raft 算法大部分人都已經瞭解，也有不少實現，從 GitHub 上來看，彷佛 Golang 語言實現的較多，比較有名的，例如 etcd。而 Java 版本的，在生產環境大規模使用的實現則較少；

同時，他們的設計目標大部分都是命名服務，即服務註冊發現，也就是說，他們一般都是基於 AP 實現，就像 DNS，DNS 是一個命名服務，同時也不是一個強一致性的服務。

比較不一樣的是 Zookeeper，ZK 常被你們用來作命名服務，但他更多的是一個分佈式服務協調者。

而上面的這些都不是存儲服務，雖然也均可以作一些存儲工做。甚至像 kafka，能夠利用 ZK 實現分佈式存儲。

回到咱們這邊。

這次咱們語言部分使用 Java，RPC 網絡通訊框架使用的是螞蟻金服 SOFA-Bolt，底層 KV 存儲使用的是 RocksDB，其中核心的 Raft 則由咱們本身實現（若是不本身實現，那這個項目沒有意義）。 注意，該項目將捨棄一部分性能和可用性，以追求儘量的強一致性。

爲何要費盡心力重複造輪子

小時候，咱們閱讀關於高可用的文章時，最後都會提到一個問題：服務掛了怎麼辦？

一般有 2 種回答：

1. 若是是無狀態服務，那麼絕不影響使用。
2. 若是是有狀態服務，能夠將狀態保存到一個別的地方，例如 Redis。若是 Redis 掛了怎麼辦？那就放到 ZK。

不少中間件，都會使用 ZK 來保證狀態一致，例如 codis，kafka。由於使用 ZK 可以幫咱們節省大量的時間。但有的時候，中間件的用戶以爲引入第三方中間件很麻煩，那麼中間件開發者會嘗試本身實現一致性，例如 Redis Cluster， TiDB 等。

而一般本身實現，都會使用 Raft 算法，那有人問，爲何不使用"更牛逼的" paxos 算法？對不起，這個有點難，至少目前開源的、生產環境大規模使用的 paxos 算法實現尚未出現，只聽過 Google 或者 alibaba 在其內部實現過，具體是什麼樣子的，這裏咱們就不討論了。

回到咱們的話題，爲何重複造輪子？從 3 個方面來回答：

1. 有的時候 ZK 和 etcd 並不能解決咱們的問題，或者像上面說的，引入其餘的中間件部署起來太麻煩也過重。
2. 徹底處於好奇，好奇爲何 Raft 能夠保證一致性（這一般能夠經過汗牛充棟的文章來獲得解答）？可是到底該怎麼實現？
3. 分佈式開發的要求，做爲開發分佈式系統的程序員，若是可以更深入的理解分佈式系統的核心算法，那麼對如何合理設計一個分佈式系統將大有益處。

好，有了以上 3 個緣由，咱們就有足夠的動力來造輪子了，接下來就是如何造的問題了。

編寫前的 Raft 理論基礎

任何實踐都是理論先行。若是你對 Raft 理論已經很是熟悉，那麼能夠跳過此節，直接看實現的步驟。

Raft 爲了算法的可理解性，將算法分紅了 4 個部分。

1. leader 選舉
2. 日誌複製
3. 成員變動
4. 日誌壓縮

同 zk 同樣，leader 都是必須的，全部的寫操做都是由 leader 發起，從而保證數據流向足夠簡單。而 leader 的選舉則經過比較每一個節點的邏輯時間（term）大小，以及日誌下標（index）的大小。

剛剛說 leader 選舉涉及日誌下標，那麼就要講日誌複製。日誌複製能夠說是 Raft 核心的核心，說簡單點，Raft 就是爲了保證多節點之間日誌的一致。當日志一致，咱們能夠認爲整個系統的狀態是一致的。這個日誌你能夠理解成 mysql 的 binlog。

Raft 經過各類補丁，保證了日誌複製的正確性。

Raft leader 節點會將客戶端的請求都封裝成日誌，發送到各個 follower 中，若是集羣中超過一半的 follower 回覆成功，那麼這個日誌就能夠被提交（commit），這個 commit 能夠理解爲 ACID 的 D ，即持久化。當日志被持久化到磁盤，後面的事情就好辦了。

而第三點則是爲了節點的擴展性。第四點是爲了性能。相比較 leader 選舉和 日誌複製，不是那麼的重要，能夠說，若是沒有成員變動和日誌壓縮，也能夠搞出一個可用的 Raft 分佈式系統，但沒有 leader 選舉和日誌複製，是萬萬不能的。

所以，本文和本項目將重點放在 leader 選舉和日誌複製。

以上，就簡單說明了 Raft 的算法，關於 Raft 算法更多的文章，請參考本人博客中的其餘文章（包含官方各個版本論文和 PPT & 動畫 & 其餘博客文章），博客地址：thinkinjava.cn

實現的步驟

實現目標：基於 Raft 論文實現 Raft 核心功能，即 Leader 選舉 & 日誌複製。

Raft 核心組件包括：一致性模塊，RPC 通訊，日誌模塊，狀態機。

技術選型：

• 一致性模塊，是 Raft 算法的核心實現，經過一致性模塊，保證 Raft 集羣節點數據的一致性。這裏咱們須要本身根據論文描述去實現。
• RPC 通訊，可使用 HTTP 短鏈接，也能夠直接使用 TCP 長鏈接，考慮到集羣各個節點頻繁通訊，同時節點一般都在一個局域網內，所以咱們選用 TCP 長鏈接。而 Java 社區長鏈接框架首選 Netty，這裏咱們選用螞蟻金服網絡通訊框架 SOFA-Bolt（基於 Netty），便於快速開發。
• 日誌模塊，Raft 算法中，日誌實現是基礎，考慮到時間因素，咱們選用 RocksDB 做爲日誌存儲。
• 狀態機，能夠是任何實現，其實質就是將日誌中的內容進行處理。能夠理解爲 Mysql binlog 中的具體數據。因爲咱們是要實現一個 KV 存儲，那麼能夠直接使用日誌模塊的 RocksDB 組件。

以上。咱們能夠看到，得益於開源世界，咱們開發一個 Raft 存儲，只須要編寫一個「一致性模塊」就好了，其餘模塊都有現成的輪子可使用，真是美滋滋。

接口設計：

上面咱們說了 Raft 的幾個核心功能，事實上，就能夠理解爲接口。因此咱們定義如下幾個接口：

1. Consensus， 一致性模塊接口
2. LogModule，日誌模塊接口
3. StateMachine， 狀態機接口
4. RpcServer & RpcClient， RPC 接口
5. Node，同時，爲了聚合上面的幾個接口，咱們須要定義一個 Node 接口，即節點，Raft 抽象的機器節點。
6. LifeCycle， 最後，咱們須要管理以上組件的生命週期，所以須要一個 LifeCycle 接口。

接下來，咱們須要詳細定義核心接口 Consensus。咱們根據論文定義了 2 個核心接口：

/** * 請求投票 RPC * * 接收者實現： * * 若是term < currentTerm返回 false （5.2 節） * 若是 votedFor 爲空或者就是 candidateId，而且候選人的日誌至少和本身同樣新，那麼就投票給他（5.2 節，5.4 節） */
    RvoteResult requestVote(RvoteParam param);

    /** * 附加日誌(多個日誌,爲了提升效率) RPC * * 接收者實現： * * 若是 term < currentTerm 就返回 false （5.1 節） * 若是日誌在 prevLogIndex 位置處的日誌條目的任期號和 prevLogTerm 不匹配，則返回 false （5.3 節） * 若是已經存在的日誌條目和新的產生衝突（索引值相同可是任期號不一樣），刪除這一條和以後全部的 （5.3 節） * 附加任何在已有的日誌中不存在的條目 * 若是 leaderCommit > commitIndex，令 commitIndex 等於 leaderCommit 和 新日誌條目索引值中較小的一個 */
    AentryResult appendEntries(AentryParam param);
複製代碼

請求投票 & 附加日誌。也就是咱們的 Raft 節點的核心功能，leader 選舉和 日誌複製。實現這兩個接口是 Raft 的關鍵所在。

而後再看 LogModule 接口，這個自由發揮，考慮日誌的特色，我定義瞭如下幾個接口：

void write(LogEntry logEntry);

LogEntry read(Long index);

void removeOnStartIndex(Long startIndex);

LogEntry getLast();

Long getLastIndex();

複製代碼

分別是寫，讀，刪，最後是兩個關於 Last 的接口，在 Raft 中，Last 是一個很是關鍵的東西，所以我這裏單獨定義了 2個方法，雖然看起來不是很好看 ：）

狀態機接口，在 Raft 論文中，將數據保存到狀態機，做者稱之爲應用，那麼咱們也這麼命名，說白了，就是將已成功提交的日誌應用到狀態機中：

/** * 將數據應用到狀態機. * * 原則上,只需這一個方法(apply). 其餘的方法是爲了更方便的使用狀態機. * @param logEntry 日誌中的數據. */
    void apply(LogEntry logEntry);

    LogEntry get(String key);

    String getString(String key);

    void setString(String key, String value);

    void delString(String... key);
    
複製代碼

第一個 apply 方法，就是 Raft 論文經常說起的方法，即將日誌應用到狀態機中，後面的幾個方法，都是我爲了方便獲取數據設計的，能夠不用在乎，甚至於，這幾個方法不存在也不影響 Raft 的實現，但影響 KV 存儲的實現，試想：一個系統只有保存功能，沒有獲取功能，要你何用？。

RpcClient 和 RPCServer 沒什麼好講的，其實就是 send 和 receive。

而後是 Node 接口，Node 接口也是 Raft 沒有定義的，咱們依靠本身的理解定義了幾個接口：

/** * 設置配置文件. * * @param config */
    void setConfig(NodeConfig config);

    /** * 處理請求投票 RPC. * * @param param * @return */
    RvoteResult handlerRequestVote(RvoteParam param);

    /** * 處理附加日誌請求. * * @param param * @return */
    AentryResult handlerAppendEntries(AentryParam param);

    /** * 處理客戶端請求. * * @param request * @return */
    ClientKVAck handlerClientRequest(ClientKVReq request);

    /** * 轉發給 leader 節點. * @param request * @return */
    ClientKVAck redirect(ClientKVReq request);
複製代碼

首先，一個 Node 確定須要配置文件，因此有一個 setConfig 接口， 而後，確定須要處理「請求投票」和「附加日誌」，同時，還須要接收用戶，也就是客戶端的請求（否則數據從哪來？），因此有 handlerClientRequest 接口，最後，考慮到靈活性，咱們讓每一個節點均可以接收客戶端的請求，但 follower 節點並不能處理請求，因此須要重定向到 leader 節點，所以，咱們須要一個重定向接口。

最後是生命週期接口，這裏咱們簡單定義了 2 個，有須要的話，再另外加上組合接口：

void init() throws Throwable;

    void destroy() throws Throwable;
複製代碼

好，基本的接口定義完了，後面就是實現了。實現纔是關鍵。

Leader 選舉的實現

選舉，其實就是一個定時器，根據 Raft 論文描述，若是超時了就須要從新選舉，咱們使用 Java 的定時任務線程池進行實現，實現以前，須要肯定幾個點：

1. 選舉者必須不是 leader。
2. 必須超時了才能選舉，具體超時時間根據你的設計而定,注意，每一個節點的超時時間不能相同，應當使用隨機算法錯開（Raft 關鍵實現），避免無謂的死鎖。
3. 選舉者優先選舉本身,將本身變成 candidate。
4. 選舉的第一步就是把本身的 term 加一。
5. 而後像其餘節點發送請求投票 RPC，請求參數參照論文，包括自身的 term，自身的 lastIndex，以及日誌的 lastTerm。同時，請求投票 RPC 應該是並行請求的。
6. 等待投票結果應該有超時控制，若是超時了，就不等待了。
7. 最後，若是有超過半數的響應爲 success，那麼就須要當即變成 leader ，併發送心跳阻止其餘選舉。
8. 若是失敗了，就須要從新選舉。注意，這個期間，若是有其餘節點發送心跳，也須要馬上變成 follower，不然，將死循環。

具體代碼，可參見 github.com/stateIs0/lu…

上面說的，實際上是 Leader 選舉中，請求者的實現，那麼接收者如何實現呢？接收者在收到「請求投票」 RPC 後，須要作如下事情：

1. 注意，選舉操做應該是串行的，由於涉及到狀態修改，併發操做將致使數據錯亂。也就是說，若是搶鎖失敗，應當當即返回錯誤。
2. 首先判斷對方的 term 是否小於本身，若是小於本身，直接返回失敗。
3. 若是當前節點沒有投票給任何人，或者投的正好是對方，那麼就能夠比較日誌的大小，反之，返回失敗。
4. 若是對方日誌沒有本身大，返回失敗。反之，投票給對方，並變成 follower。變成 follower 的同時，異步的選舉任務在最後從 condidate 變成 leader 以前，會判斷是不是 follower，若是是 follower，就放棄成爲 leader。這是一個兜底的措施。

具體代碼參見 github.com/stateIs0/lu…

到這裏，基本就可以實現 Raft Leader 選舉的邏輯。

注意，咱們上面涉及到的 LastIndex 等參數，尚未實現，但不影響咱們編寫僞代碼，畢竟日誌複製比 leader 選舉要複雜的多，咱們的原則是從易到難。：）

日誌複製的實現

日誌複製是 Raft 實現一致性的核心。

日誌複製有 2 種形式，1種是心跳，一種是真正的日誌，心跳的日誌內容是空的，其餘部分基本相同，也就是說，接收方在收到日誌時，若是發現是空的，那麼他就是心跳。

心跳

既然是心跳，確定就是個定時任務，和選舉同樣。在咱們的實現中，咱們每 5 秒發送一次心跳。注意點：

1. 首先本身必須是 leader 才能發送心跳。
2. 必須知足 5 秒的時間間隔。
3. 併發的向其餘 follower 節點發送心跳。
4. 心跳參數包括自身的 ID，自身的 term，以便讓對方檢查 term，防止網絡分區致使的腦裂。
5. 若是任意 follower 的返回值的 term 大於自身，說明本身分區了，那麼須要變成 follower，並更新本身的 term。而後從新發起選舉。

具體代碼查看：github.com/stateIs0/lu…

而後是心跳接收者的實現，這個就比較簡單了，接收者須要作幾件事情：

1. 不管成功失敗首先設置返回值，也就是將本身的 term 返回給 leader。
2. 判斷對方的 term 是否大於自身，若是大於自身，變成 follower，防止異步的選舉任務誤操做。同時更新選舉時間和心跳時間。
3. 若是對方 term 小於自身，返回失敗。不更新選舉時間和心跳時間。以便觸發選舉。

具體代碼參見：github.com/stateIs0/lu…

說完了心跳，再說說真正的日誌附加。

簡單來講，當用戶向 Leader 發送一個 KV 數據，那麼 Leader 須要將 KV數據封裝成日誌，並行的發送到其餘的 follower 節點，只要在指定的超時時間內，有過半幾點返回成功，那麼久提交（持久化）這條日誌，返回客戶端成功，否者返回失敗。

所以，Leader 節點會有一個 ClientKVAck handlerClientRequest(ClientKVReq request) 接口，用於接收用戶的 KV 數據，同時，會並行向其餘節點複製數據，具體步驟以下：

1. 每一個節點均可能會接收到客戶端的請求，但只有 leader 能處理，因此若是自身不是 leader，則須要轉發給 leader。
2. 而後將用戶的 KV 數據封裝成日誌結構，包括 term，index，command，預提交到本地。
3. 並行的向其餘節點發送數據，也就是日誌複製。
4. 若是在指定的時間內，過半節點返回成功，那麼就提交這條日誌。
5. 最後，更新本身的 commitIndex，lastApplied 等信息。

注意，複製不只僅是簡單的將這條日誌發送到其餘節點，這可能比咱們想象的複雜，爲了保證複雜網絡環境下的一致性，Raft 保存了每一個節點的成功複製過的日誌的 index，即 nextIndex ，所以，若是對方以前一段時間宕機了，那麼，從宕機那一刻開始，到當前這段時間的全部日誌，都要發送給對方。

甚至於，若是對方以爲你發送的日誌仍是太大，那麼就要遞減的減少 nextIndex，複製更多的日誌給對方。注意：這裏是 Raft 實現分佈式一致性的關鍵所在。

具體代碼參見：github.com/stateIs0/lu…

再來看看日誌接收者的實現步驟：

1. 和心跳同樣，要先檢查對方 term，若是 term 都不對，那麼就沒什麼好說的了。
2. 若是日誌不匹配，那麼返回 leader，告訴他，減少 nextIndex 重試。
3. 若是本地存在的日誌和 leader 的日誌衝突了，以 leader 的爲準，刪除自身的。
4. 最後，將日誌應用到狀態機，更新本地的 commitIndex，返回 leader 成功。

具體代碼參見：github.com/stateIs0/lu…

到這裏，日誌複製的部分就講完了。

注意，實現日誌複製的前提是，必須有一個正確的日誌存儲系統，即咱們的 RocksDB，咱們在 RocksDB 的基礎上，使用一種機制，維護了 每一個節點 的LastIndex，不管什麼時候何地，都可以獲得正確的 LastIndex，這是實現日誌複製不可獲取的一部分。

驗證「Leader 選舉」和「日誌複製」

寫完了程序，如何驗證是否正確呢？

固然是寫驗證程序。

咱們首先驗證 「Leader 選舉」。其實這個比較好測試。

1. 在 idea 中配置 5 個 application 啓動項,配置 main 類爲 RaftNodeBootStrap 類, 加入 -DserverPort=8775 -DserverPort=8776 -DserverPort=8777 -DserverPort=8778 -DserverPort=8779 系統配置, 表示分佈式環境下的 5 個機器節點.
2. 依次啓動 5 個 RaftNodeBootStrap 節點, 端口分別是 8775，8776， 8777, 8778, 8779.
3. 觀察控制檯, 約 6 秒後, 會發生選舉事件,此時,會產生一個 leader. 而 leader 會馬上發送心跳維持本身的地位.
4. 若是leader 的端口是 8775, 使用 idea 關閉 8775 端口，模擬節點掛掉, 大約 15 秒後, 會從新開始選舉, 而且會在剩餘的 4 個節點中,產生一個新的 leader. 並開始發送心跳日誌。

而後驗證 日誌複製，分爲 2 種狀況：

正常狀態下

1. 在 idea 中配置 5 個 application 啓動項,配置 main 類爲 RaftNodeBootStrap 類, 加入 -DserverPort=8775 -DserverPort=8776 -DserverPort=8777 -DserverPort=8778 -DserverPort=8779
2. 依次啓動 5 個 RaftNodeBootStrap 節點, 端口分別是 8775，8776， 8777, 8778, 8779.
3. 使用客戶端寫入 kv 數據.
4. 殺掉全部節點, 使用 junit test 讀取每一個 rocksDB 的值, 驗證每一個節點的數據是否一致.

非正常狀態下

1. 在 idea 中配置 5 個 application 啓動項,配置 main 類爲 RaftNodeBootStrap 類, 加入 -DserverPort=8775 -DserverPort=8776 -DserverPort=8777 -DserverPort=8778 -DserverPort=8779
2. 依次啓動 5 個 RaftNodeBootStrap 節點, 端口分別是 8775，8776， 8777, 8778, 8779.
3. 使用客戶端寫入 kv 數據.
4. 殺掉 leader （假設是 8775）.
5. 再次寫入數據.
6. 重啓 8775.
7. 關閉全部節點, 讀取 RocksDB 驗證數據一致性.

Summary

本文並無貼不少代碼，若是要貼代碼的話，閱讀體驗將不會很好，而且代碼也不能說明什麼，若是想看具體實現，能夠到 github 上看看，順便給個 star ：）

該項目 Java 代碼約 2500 行，核心代碼估計也就 1000 多行。你甚至能夠說，這是個玩具代碼，但我相信畢玄大師所說，玩具代碼通過優化後，也是能夠變成可在商業系統中真正健壯運行的代碼（hellojava.info/?p=508） ：）

回到咱們的初衷，咱們並不奢望這段代碼可以運行在生產環境中，就像個人另外一個項目 Lu-RPC 同樣。但，經歷了一次編寫可正確運行的玩具代碼的經歷，下次再次編寫工程化的代碼，應該會更加容易些。這點我深有體會。

能夠稍微展開講一下，在寫完 Lu-RPC 項目後，我就接到了開發生產環境運行的限流熔斷框架任務，此時，開發 Lu-RPC 的經歷讓我在開發該框架時，更加的從容和自如：）

再回到 Raft 上面來，雖然上面的測試用例跑過了，程序也通過了我反反覆覆的測試，但不表明這個程序就是 100% 正確的，特別是在複雜的分佈式環境下。若是你對 Raft 有興趣，歡迎一塊兒交流溝通 ：）

項目地址：github.com/stateIs0/lu…