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 中保存一个分布式计数器，该计数器可以递增和获取，并且在所有节点之间保持一致，任何少数节点的挂掉都不会影响对外提供的两个服务：

1. incrmentAndGet(delta) 递增 delta 数值并返回递增后的值。
2. get() 获取最新的值

RPC 请求

jraft 底层使用 bolt 作为通讯框架，定义两个请求

1. 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; }}

1. GetValueRequest，用于获取最新值：

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

应答结果 ValueResponse，包括：

1. success　是否成功
2. value 成功情况下返回的最新值
3. errorMsg 失败情况下的错误信息
4. 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