SOFAJRaft 是一個基于 RAFT 一致性算法的生產級高性能 Java 實現，支持 MULTI-RAFT-GROUP，適用于高負載低延遲的場景。 使用 SOFAJRaft 你可以專注于自己的業務領域，由 SOFAJRaft 負責處理所有與 RAFT 相關的技術難題，并且 SOFAJRaft 非常易于使用，你可以通過幾個示例在很短的時間內掌握它。

功能特性

• Leader 選舉
• 日志復制和恢復
• 快照和日志壓縮
• 集群線上配置變更，增加節點、刪除節點、替換節點等
• 主動變更 Leader，用于重啟維護，Leader 負載平衡等
• 對稱網絡分區容忍性
• 非對稱網絡分區容忍性
• 容錯性，少數派故障，不影響系統整體可用性
• 多數派故障時手動恢復集群可用
• 高效的線性一致讀，ReadIndex/LeaseRead
• 流水線復制
• 內置了基于 Metrics 類庫的性能指標統計，有豐富的性能統計指標
• 通過了 Jepsen 一致性驗證測試
• JRaft 中包含了一個嵌入式的分布式 KV 實現

介紹

本介紹內容來自 braft 文檔，原文鏈接請參見這里。braft 的關于算法和應用本身的文檔非常優秀，由于 jraft 脫胎自 braft，我們強烈推薦閱讀上述文檔以了解 raft 算法的基本原理和應用。

分布式一致性

分布式一致性 (distributed consensus) 是分布式系統中最基本的問題，用來保證一個分布式系統的可靠性以及容災能力。簡單的來講，就是如何在多個機器間對某一個值達成一致, 并且當達成一致之后，無論之后這些機器間發生怎樣的故障，這個值能保持不變。 抽象定義上， 一個分布式系統里的所有進程要確定一個值 v，如果這個系統滿足如下幾個性質， 就可以認為它解決了分布式一致性問題, 分別是:

• Termination: 所有正常的進程都會決定 v 具體的值，不會出現一直在循環的進程。
• Validity: 任何正常的進程確定的值 v', 那么 v' 肯定是某個進程提交的。比如隨機數生成器就不滿足這個性質。
• Agreement: 所有正常的進程選擇的值都是一樣的。

一致性狀態機

對于一個無限增長的序列 a[1, 2, 3…], 如果對于任意整數 i, a[i] 的值滿足分布式一致性，這個系統就滿足一致性狀態機的要求。 基本上所有的系統都會有源源不斷的操作, 這時候單獨對某個特定的值達成一致是不夠的。為了真實系統保證所有的副本的一致性，通常會把操作轉化為 write-ahead-log(簡稱WAL)。然后讓系統的所有副本對WAL保持一致，這樣每個進程按照順序執行WAL里的操作，就能保證最終的狀態是一致的。

RAFT

RAFT 是一種新型易于理解的分布式一致性復制協議，由斯坦福大學的 Diego Ongaro 和 John Ousterhout 提出，作為 RAMCloud 項目中的中心協調組件。Raft 是一種 Leader-Based 的 Multi-Paxos 變種，相比 Paxos、Zab、View Stamped Replication 等協議提供了更完整更清晰的協議描述，并提供了清晰的節點增刪描述。 Raft 作為復制狀態機，是分布式系統中最核心最基礎的組件，提供命令在多個節點之間有序復制和執行，當多個節點初始狀態一致的時候，保證節點之間狀態一致。系統只要多數節點存活就可以正常處理，它允許消息的延遲、丟棄和亂序，但是不允許消息的篡改(非拜占庭場景)。

Raft 可以解決分布式理論中的 CP，即一致性和分區容忍性，并不能解決 Available 的問題。其中包含分布式系統中一些通常的功能：

• Leader Election
• Log Replication
• Membership Change
• Log Compaction

RAFT 可以做什么

通過 RAFT 提供的一致性狀態機，可以解決復制、修復、節點管理等問題，極大的簡化當前分布式系統的設計與實現，讓開發者只關注于業務邏輯，將其抽象實現成對應的狀態機即可。基于這套框架，可以構建很多分布式應用：

• 分布式鎖服務，比如 Zookeeper
• 分布式存儲系統，比如分布式消息隊列、分布式塊系統、分布式文件系統、分布式表格系統等
• 高可靠元信息管理，比如各類 Master 模塊的 HA

Counter 例子詳解

本文檔主要介紹一個基于 jraft 的分布式計數器的例子。

場景

在多個節點(機器)組成的一個 raft group 中保存一個分布式計數器，該計數器可以遞增和獲取，并且在所有節點之間保持一致，任何少數節點的掛掉都不會影響對外提供的兩個服務：

incrmentAndGet(delta) 遞增 delta 數值并返回遞增后的值。

get() 獲取最新的值

RPC 請求

jraft 底層使用 bolt 作為通訊框架，定義兩個請求

IncrementAndGetRequest，用于遞增

public class IncrementAndGetRequest implements Serializable { private static final long serialVersionUID = -5623664785560971849L; private long delta; public long getDelta() { return this.delta; } public void setDelta(long delta) { this.delta = delta; }}

GetValueRequest，用于獲取最新值：

public class GetValueRequest implements Serializable { private static final long serialVersionUID = 9218253805003988802L; public GetValueRequest() { super(); }}

應答結果 ValueResponse，包括：

success　是否成功

value 成功情況下返回的最新值

errorMsg 失敗情況下的錯誤信息

redirect　發生了重新選舉，需要跳轉的新的leader節點。

public class ValueResponse implements Serializable { private static final long serialVersionUID = -4220017686727146773L; private long value; private boolean success; /** * redirect peer id */ private String redirect; private String errorMsg; public String getErrorMsg() { return this.errorMsg; } public void setErrorMsg(String errorMsg) { this.errorMsg = errorMsg; } ......}

IncrementAndAddClosure 用于 Leader 服務端接收 IncrementAndGetRequest 請求后的回調處理：

public class IncrementAndAddClosure implements Closure { private CounterServer counterServer; private IncrementAndGetRequest request; private ValueResponse response; private Closure done; // 網絡應答callback public IncrementAndAddClosure(CounterServer counterServer, IncrementAndGetRequest request, ValueResponse response, Closure done) { super(); this.counterServer = counterServer; this.request = request; this.response = response; this.done = done; } @Override public void run(Status status) { // 返回應答給客戶端 if (this.done != null) { done.run(status); } } public IncrementAndGetRequest getRequest() { return this.request; } public void setRequest(IncrementAndGetRequest request) { this.request = request; } public ValueResponse getResponse() { return this.response; }}

服務端

狀態機 CounterStateMachine

首先持有一個初始值：

public class CounterStateMachine extends StateMachineAdapter { /** * counter value */ private AtomicLong value = new AtomicLong(0);

實現核心的 onApply(iterator) 方法，應用用戶提交的請求到狀態機：

@Override public void onApply(Iterator iter) { // 遍歷日志 while (iter.hasNext()) { long delta = 0; IncrementAndAddClosure closure = null; // done 回調不為null，必須在應用日志后調用，如果不為 null，說明當前是leader。 if (iter.done() != null) { // 當前是leader，可以直接從 IncrementAndAddClosure 中獲取 delta，避免反序列化 closure = (IncrementAndAddClosure) iter.done(); delta = closure.getRequest().getDelta(); } else { // 其他節點應用此日志，需要反序列化 IncrementAndGetRequest，獲取 delta ByteBuffer data = iter.getData(); try { IncrementAndGetRequest request = Codecs.getSerializer(Codecs.Hessian2).decode(data.array(), IncrementAndGetRequest.class.getName()); delta = request.getDelta(); } catch (CodecException e) { LOG.error("Fail to decode IncrementAndGetRequest

總結

以上是生活随笔為你收集整理的java 状态机_基于 RAFT 一致性算法的 Java 实现 SOFAJRaft的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。