前言

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

項目地址：https://github.com/stateIs0/lu-raft-kv

歡迎 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，否則，將死循環。

具體代碼，可參見 https://github.com/stateIs0/lu-raft-kv/blob/master/lu-raft-kv/src/main/java/cn/think/in/java/impl/DefaultNode.java#L546

上面說的，其實是 Leader 選舉中，請求者的實現，那麽接收者如何實現呢？接收者在收到“請求投票” RPC 後，需要做以下事情：

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

具體代碼參見 https://github.com/stateIs0/lu-raft-kv/blob/master/lu-raft-kv/src/main/java/cn/think/in/java/impl/DefaultConsensus.java#L51

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

註意，我們上面涉及到的 LastIndex 等參數，還沒有實現，但不影響我們編寫偽代碼，畢竟日誌復制比 leader 選舉要復雜的多，我們的原則是從易到難。：）

日誌復制的實現

日誌復制是 Raft 實現一致性的核心。

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

心跳

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

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

具體代碼查看：https://github.com/stateIs0/lu-raft-kv/blob/master/lu-raft-kv/src/main/java/cn/think/in/java/impl/DefaultNode.java#L695

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

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

具體代碼參見：https://github.com/stateIs0/lu-raft-kv/blob/master/lu-raft-kv/src/main/java/cn/think/in/java/impl/DefaultConsensus.java#L109

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

簡單來說，當用戶向 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 實現分布式一致性的關鍵所在。

具體代碼參見：https://github.com/stateIs0/lu-raft-kv/blob/master/lu-raft-kv/src/main/java/cn/think/in/java/impl/DefaultNode.java#L244

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

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

具體代碼參見：https://github.com/stateIs0/lu-raft-kv/blob/master/lu-raft-kv/src/main/java/cn/think/in/java/impl/DefaultConsensus.java#L109

到這裏，日誌復制的部分就講完了。

註意，實現日誌復制的前提是，必須有一個正確的日誌存儲系統，即我們的 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 多行。你甚至可以說，這是個玩具代碼，但我相信畢玄大師所說，玩具代碼經過優化後，也是可以變成可在商業系統中真正健壯運行的代碼（http://hellojava.info/?p=508） ：）

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

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

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

項目地址：https://github.com/stateIs0/lu-raft-kv

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