Raft分布式一致性算法怎么实现

本篇内容介绍了"Raft分布式一致性算法怎么实现"的有关知识，在实际案例的操作过程中，不少人都会遇到这样的困境，接下来就让小编带领大家学习一下如何处理这些情况吧！希望大家仔细阅读，能够学有所成！

• CAP定理

1. Consistency:一致性

2. Availability:可用性

3. Partition-tolerance:分区容错性

CAP定理指出，在异步网络模型中，不存在一个系统可以同时满足上述3个属性。换句话说，分布式系统必须舍弃其中的一个属性。对于需要在分布式条件下运行的系统来说，如何在一致性、可用性和分区容错性中取舍，或者说要弱化哪一个属性，是首先要考虑的问题。

对于高可用性的系统来说，往往会保留强一致性。但对于强一致性的系统来说，有一类专门解决这种问题的算法--共识算法。"共识"的意思是保证所有的参与者都有相同的认知(可以理解为强一致性)。共识算法本身可以依据是否有恶意节点分为两类，大部分时候共识算法指的是没有恶意节点的那一类，即系统中的节点不会向其他节点发送恶意请求，比如欺骗请求。共识算法中最有名的是Paxos算法。其次是Raft和ZAB算法(Zookeeper中的实现)

• Raft核心算法

Raft算法的核心是选举和日志复制。

当多台服务器同时对外服务时，服务器如何同步变更成了一个问题。一般是采用主从模型，即一个主服务器(Leader)，多个从服务器(Follower),所有请求都通过Leader服务器处理，Follower服务器只负责备份数据。但假设Leader服务器宕机了，那么Follower服务器中哪个服务器成为新的Leader服务器呢？理论上所有的Follower服务器的副本数据应该是和Leader服务器一致的，但是由于数据延迟、发送顺序不一致等问题，导致某个时刻每个Follower服务器拥有的数据有可能不一样。由此产生的问题需要从以下两方面进行处理。

1. 使用日志写入，而不是直接修改，保证到Follower服务器的同步请求有序而且能够重新计算当前状态，也就是日志状态机模型。

2. 写入时，过半服务器写入成功才算整体成功，也就是Quorum机制。

• 日志状态机模型

在状态机模型中，日志从上往下不断追加，当前状态的任何时间点都可以从索引为1的日志开始计算。有了状态机模型后，分布式一致性的问题就转换成了如何保证所有参与的节点按照同一顺序写入的问题。

• 基于Quorum机制的写入

在一些master/slave模式中，有些master并不关心slave的复制进度。master只负责不断写入自己的日志，通过某些传输方式把变更同步给slave服务器。而在一些严格的全量复制中，当所有的slave服务器全部同步之后，master服务器才会继续写入。主从复制在master服务器宕机之后数据会丢失，而全量复制则性能非常差。相比之下，过半写入的Quorum机制既可以减少数据丢失的风险，性能也不会太差。

现在假设有3台服务器，节点A、B、C。此时正在向这三台服务器写入值，此时节点A的值是2(最新值)，而节点B和C的值都是旧值1.此时当客户端向这个集群取值的时候，如果读取任意两个节点的数据，客户端读取到的数据版本有以下可能。

1. 节点A和B:2与1

2. 节点A和C:2与1

3. 节点B和C:1与1

此时我们可以看到，当读取到B和C的时候，客户端没有读取到最新数据。

此时B节点也写入了新值2，此时我们称为过半写入完成。

当客户端向这个集群任意两个节点取值的时候。

1. 节点A和B:2与2

2. 节点A和C:2与1

3. 节点B和C:2与1

由以上结果我们可以看到，当过半写入的时候，无论哪一种情况，客户端都能读取到最新的值。对于master/slave或者leader/follower模型的分布式系统来说，客户端并不能直接访问所有节点，但是对于系统内的服务器节点来说，可以通过比较各自持有的日志来决定谁成为新的Leader节点，在此过程中，过半写入的数据往往是有效的数据。

• 基于日志比较的选举

假设Leader节点宕机，那么如何从剩下的服务器节点中选举新的Leader节点呢？一般情况下，肯定是希望选择拥有最新数据的节点。

理论上，这个拥有最新数据的节点应该有过半节点的支持，也就是说，集群中超过半数的节点(包括这个拥有最新数据的节点自身)的数据不会比这个节点更新。如果不满足这个条件，集群中可能出现"脑裂"现象，比如几个节点拥护一个Leader节点，而另外几个节点拥护另一个Leader节点。

对于如何判断谁的数据更新，可以通过比较来自其他节点的投票请求中的日志索引和自己本地的日志索引来确定。如果自己本地的日志索引比较大，则不支持对方，否则就支持。

根据这个规则，如果三个节点的日志索引都是2，则A会支持B和C以及自己，其他节点相同，每个节点都是三票。为了减少同时成为Leader节点的概率，要求节点不能重复投票，即每个节点只能投一票。

从以上结果可以看出，除了全部自荐，必会有一个节点被选举出来成为Leader。

对于N个节点的集群(N>0)，假设有M个节点成为Leader节点，那么这些节点都需要有过半的支持票，则总票数为

M * N过半

当节点数为奇数时，N过半为(N + 1) / 2

当节点数为偶数时，N过半为N / 2 + 1

而 M * N过半 <= N

要满足该式成立，M(Leader节点数)为1，N过半<= N成立，而M为2的时候

当节点数为奇数时，2 * (N + 1) / 2 = N + 1,而N + 1 <= N是不满足的

当节点数为偶数时，2 * (N / 2 + 1) = N + 2,而N + 2 <= N也是不满足的

以此类推，M >= 2的时候，M * N过半 <= N都是不满足的。

因此最多只能选出1个Leader节点。

• Raft算法中的选举

在Raft算法中，节点有3个角色

1. Leader

2. Candidate(Leader候选人)

3. Follower

在整个集群稳定状态下，Leader节点为一个，它会通过心跳消息与各个Follower节点保持联系。

包括心跳消息在内，Raft算法中使用的消息类型有以下两种。

1. RequestVote,即请求其他节点给自己投票，一般由Candidate节点发出。

2. AppendEntries，用于日志复制，增加条目，在增加日志条目数量为0时作为心跳信息，一般只由Leader节点发出。

• 逻辑时钟term

为了避免服务器时间不一致，系统也可以安全地推进逻辑时间，Raft算法中的选举有一个整形的term参数。这是一个逻辑时钟值，全局递增。它是Lamport Timestamp算法的一个变体。

当多个进程要维护一个全局时间，首先要让每个进程本地有一个全局时间的副本。Lamport Timestamp算法的流程如下

1. 每个进程在事件发生时递增自己本地的时间副本(加1)。

2. 当进程发送消息时，带上自己本地的时间副本。

3. 当进程收到消息时，比较消息中的时间值和自己本地的时间副本，选择比较大的时间值加1，并更新自己的时间副本。

• 选举中的term和角色迁移

Raft算法中主要使用term作为Leader节点的任期号，term是一个递增的参数。

在选举的过程中，节点的角色会有所变化，Raft算法中的角色迁移如上图所示。步骤如下

1. 系统启动时，所有节点都是Follower节点。

2. 当没有收到来自Leader节点心跳消息时，即心跳超时，Follower节点变成Candidate节点，即自荐成为选举的候选人。

3. Candidate节点收到过半的支持后，变成Leader节点。

出现Leader节点以后，Leader节点会发送心跳消息给其他节点，防止其他节点从Follower节点变成Candidate节点。

在上面的第3步中，如果Candidate节点都没有得到过半支持，无法选出Leader节点，此时Candidate节点选举超时，进入下一轮选举。

票数对半的现象在Raft算法中被称为split vote(分割选举)，在偶数个节点的集群中有可能发生。Raft算法使用随机选举超时来降低split vote出现的概率。

• 选举超时

如果一个偶数节点的集群总是出现分割选举，比如上面的4个节点，最坏情况下是无限次2票对2票，那就无法正常选出Leader节点了。Raft算法中错开了Follower成为Candidate的时间点，提高了选出Leader节点的概率。

在选出Leader节点之后，各个节点需要在最短的时间内获取新Leader节点的信息，否则选举超时又会进入一轮选举，即要求心跳消息间隔远远小于最小选举间隔。

节点的选举超时时间在收到心跳消息后会重置。如果不重置，节点会频繁发起选举，系统难以收敛于稳定状态。

假设选举超时时间间隔为3-4秒，心跳间隔1秒，则节点会以类似于下面的方式不断修改实际选举超时时间。

1. 节点以Follower角色启动，随机选择选举超时时间为3.3秒，即3.3秒后系统会发起选举。

2. 节点启动1秒后，收到来自Leader节点的心跳消息，节点重新随机选择一个选举超时时间(假设是3.4秒)，并修改下一次选举时间为现在时间的3.4秒后。

3. 节点启动2秒后，再次收到来自Leader节点的心跳消息，节点再次随机选择一个选举超时时间(假设是4秒)，并修改下一次选举时间为现在时间的4秒后。

只要Leader持续不断地发送心跳消息，Follower节点就不会成为Candidate角色并发起选举。

• Raft算法中的日志复制

所有来自客户端的数据变更请求都会被当作一个日志条目追加到节点日志中。日志条目分为以下两种状态

1. 已追加但是尚未持久化

2. 已持久化

Raft算法中的节点会维护一个已持久化的日志条目索引--commitIndex。小于等于commitIndex的日志条目被认为是已提交，或者说是有效的日志条目(已持久化)，否则就是尚未持久化的数据。在系统启动时commitIndex为0.

• 复制进度

为了跟踪各节点的复制进度，Leader负责记录各个节点的nextIndex(下一个需要复制日志条目的索引)和matchIndex(已匹配日志索引)。

选出Leader节点后，Leader节点会重置各节点的nextIndex和matchIndex。matchIndex设为0，nextIndex设置为Leader节点的下一条日志条目的索引，通过和各节点之间发送AppendEntries消息来更新nextIndex和matchIndex。当系统达到稳定状态时，Leader跟踪的各个节点的matchIndex与Leader的commitIndex一致，nextIndex与Leader节点的下一条日志的索引一致。

当客户端向Leader节点发送数据变更请求时，Leader节点会先向自己的日志中加一条日志，但是不提交(不增加commitIndex)。此时Leader节点通过AppendEntries消息向其他节点同步数据，消息包含了最新追加的日志。当超过半数节点(包含Leader节点自己)追加新日志成功之后，Leader节点会持久化日志并推进commitIndex，然后再次通过AppendEntries消息通知其他节点持久化日志。AppendEntries消息除了包含需要复制的日志条目外，还有Leader节点最新的commitIndex。Follower节点参考Leader节点的commitIndex推进自己的commitIndex,也就是持久化日志。如果追加日志成功的节点没有过半，Leader节点不会推进自己的commitIndex，也不会要求其他节点推进commitIndex。在Leader节点推进commitIndex的同时，状态机执行日志中的命令，并把计算后的结果返回客户端。虽然在上图中，Follower节点都持久化完成后才开始计算结果，但实际上Raft算法允许Follower的日志持久化和状态机应用日志同时进行。换句话说，只要节点的commitIndex推进了，那么表示状态机应用哪条日志的lastApplied也可以同时推进。

假如不确认过半追加，碰到"脑裂"或者网络分区的情况下，会出现严重不一致问题。

以5个服务器节点的系统为例，5个节点分别为A、B、C、D、E

1. 一开始Leader节点为A，其他节点都是Follower.

2. 在某个时间点，A、B两个节点与C、D、E 3个节点产生网络分区。网络分区时，节点A无法与节点B以外的节点通信。

3. 节点B依旧接收得到A的心跳消息，所以不会变成Candidate。

4. 节点C、D、E 收不到来自节点A的心跳消息，进行了选举，假设C节点成为了新的Leader。

5. 客户端连接节点A和C分别写入，因为Leader节点并不确认过半写入，所以会导致节点A和C各自增加不同的日志。

6. 当网络分区恢复时，由于分区内节点A、B和分区内节点C、D、E 各自的日志冲突，因此无法合并。

但如果上述过程中，Leader节点确认过半追加后再推进commitIndex，节点A不会持久化日志，并且在网络分区恢复后，分区内节点C、D、E 的日志可以正确复制到分区节点A、B 上，保证数据一致性。

现在我们来设计相关的类

为了保证节点的唯一性，我们有一个节点的ID

/** * 节点ID */@AllArgsConstructor@Getterpublic class NodeId implements Serializable {//节点的ID值，一经确定不可改变    //可以简单为A、B、C....    @NonNull    private final String value;    public static NodeId of(String value) {return new NodeId(value);    }@Override    public boolean equals(Object o) {if (this == o) return true;        if (o == null || !(o instanceof NodeId)) return false;        NodeId nodeId = (NodeId) o;        return value.equals(nodeId.value);    }@Override    public int hashCode() {return value.hashCode();    }@Override    public String toString() {return value;    }}

集群成员表

/** * 集群成员表 */public class NodeGroup {//当前节点ID    private NodeId selfId;    //成员映射表    private Map memberMap;    /**     * 单节点构造函数     * @param endpoint     */    public NodeGroup(NodeEndpoint endpoint) {this(Collections.singleton(endpoint),endpoint.getId());    }/**     * 多节点构造函数     * @param endpoints     * @param selfId     */    public NodeGroup(Collection endpoints,NodeId selfId) {this.memberMap = buildMemberMap(endpoints);        this.selfId = selfId;    }/**     * 从节点列表中构造成员映射表     * @param endpoints     * @return     */    private Map buildMemberMap(Collection endpoints) {        Map map = new HashMap<>();        endpoints.stream().forEach(endpoint -> map.put(endpoint.getId(),new GroupMember(endpoint)));        if (map.isEmpty()) {throw new IllegalArgumentException("endpoints is empty");        }return map;    }/**     * 重置其他节点的复制进度     * @param nextLogIndex     */    public void resetReplicatingStates(int nextLogIndex) {memberMap.values().stream()                .filter(member -> !member.idEquals(selfId))                .forEach(member -> member.setReplicatingState(new ReplicatingState(nextLogIndex)));    }/**     * 按照节点ID查找成员，找不到时返回空     * @param id     * @return     */    public GroupMember getMember(NodeId id) {return memberMap.get(id);    }/**     * 按照节点呢ID查找成员，找不到时抛出异常     * @param id     * @return     */    public GroupMember findMember(NodeId id) {        GroupMember member = getMember(id);        if (member == null) {throw new IllegalArgumentException("no such node " + id);        }return member;    }/**     * 获取主要节点的数量     * @return     */    public int getCountOfMajor() {return (int) memberMap.values().stream().filter(GroupMember::isMajor).count();    }    /**     * 列出日志复制的对象节点，获取除自己以外的所有节点     * @return     */    public Collection listReplicationTarget() {return memberMap.values().stream()                .filter(m -> !m.idEquals(selfId))                .collect(Collectors.toList());    }/**     * 获取当前节点之外的其他节点     * @return     */    public Set listEndpointExceptSelf() {        Set endpoints = new HashSet<>();        memberMap.values().stream()                .filter(member -> !member.idEquals(selfId))                .forEach(member -> endpoints.add(member.getEndpoint()));        return endpoints;    }}

集群成员

服务器地址

/** * 服务器地址 */@AllArgsConstructor@Getter@ToStringpublic class Address {@NonNull    private final String host; //IP    private final int port; //端口}

/** * 连接节点 */@AllArgsConstructor@Getterpublic class NodeEndpoint {@NonNull    private final NodeId id;    @NonNull    private final Address address;    public NodeEndpoint(NodeId id,String host,int port) {this(id,new Address(host,port));    }}

/** * 集群成员 */@AllArgsConstructorpublic class GroupMember {//连接节点    @Getter    private final NodeEndpoint endpoint;    //复制进度    @Setter    private ReplicatingState replicatingState;    //是否主要成员    @Getter    @Setter    private boolean major;    public GroupMember(NodeEndpoint endpoint) {this(endpoint,null,true);    }//获取下一条日志索引    public int getNextIndex() {return ensureReplicatingState().getNextIndex();    }//获取匹配日志索引    public int getMatchIndex() {return ensureReplicatingState().getMatchIndex();    }/**     * 获取复制进度     * @return     */    private ReplicatingState ensureReplicatingState() {if (replicatingState == null) {throw new IllegalArgumentException("replicating state not set");        }return replicatingState;    }/**     * 判断是否同一个连接节点     * @param id     * @return     */    public boolean idEquals(NodeId id) {return endpoint.getId().equals(id);    }}

/** * 日志复制进度 */@ToStringpublic class ReplicatingState {//下一个需要复制日志条目的索引    @Getter    private int nextIndex;    //匹配的日志条目索引    @Getter    private int matchIndex;    //是否开始复制    @Getter    @Setter    private boolean replicating = false;    //最后复制的位置    @Getter    @Setter    private long lastReplicatedAt = 0;    public ReplicatingState(int nextIndex,int matchIndex) {this.nextIndex = nextIndex;        this.matchIndex = matchIndex;    }public ReplicatingState(int nextIndex) {this(nextIndex,0);    }/**     * 回退     * @return     */    public boolean backOffNextIndex() {if (nextIndex > 1) {nextIndex--;            return true;        }return false;    }/**     * 建议是否推进索引     * @param lastEntryIndex     * @return     */    public boolean advice(int lastEntryIndex) {boolean result = matchIndex != lastEntryIndex || nextIndex != lastEntryIndex + 1;        matchIndex = lastEntryIndex;        nextIndex = lastEntryIndex + 1;        return result;    }}

• 选举实现

节点角色

public enum RoleName {FOLLOWER,   //从节点    CANDIDATE,  //选举节点    LEADER      //主节点}

各节点角色的统一抽象类

@AllArgsConstructor@Getterpublic abstract class AbstractNodeRole {//节点角色    private final RoleName name;    //选举周期    protected final int term;    /**     * 取消每个角色对应的选举超时或者日志复制定时任务     */    public abstract void cancelTimeoutOrTask();    public abstract NodeId getLeaderId(NodeId selfId);}

选举超时类

/** * 选举超时 */@RequiredArgsConstructorpublic class ElectionTimeout {//定时任务结果    private final ScheduledFuture scheduledFuture;    public static final ElectionTimeout NONE = new ElectionTimeout(new NullScheduledFuture());    /**     * 取消选举     */    public void cancel() {scheduledFuture.cancel(false);    }@Override    public String toString() {if (scheduledFuture.isCancelled()) {return "ElectionTimeout(state=cancelled)";        }if (scheduledFuture.isDone()) {return "ElectionTimeout(state=done)";        }return "ElectionTimeout(delay=" + scheduledFuture.getDelay(TimeUnit.MILLISECONDS) + "ms)";    }}

从节点角色

/** * 从节点角色 */@ToStringpublic class FollowerNodeRole extends AbstractNodeRole {//投过票的节点    @Getter    private final NodeId votedFor;    //当前主节点    @Getter    private final NodeId leaderId;    //选举超时    private final ElectionTimeout electionTimeout;    public FollowerNodeRole(int term,NodeId votedFor,NodeId leaderId,ElectionTimeout electionTimeout) {super(RoleName.FOLLOWER, term);        this.votedFor = votedFor;        this.leaderId = leaderId;        this.electionTimeout = electionTimeout;    }@Override    public void cancelTimeoutOrTask() {electionTimeout.cancel();    }@Override    public NodeId getLeaderId(NodeId selfId) {return leaderId;    }}

选举节点角色

/** * 选举节点角色 */@ToStringpublic class CandidateNodeRole extends AbstractNodeRole{//票数    @Getter    private final int votesCount;    //选举超时    private final ElectionTimeout electionTimeout;    /**     * 增加任意票构造     * @param term     * @param votesCount     * @param electionTimeout     */    public CandidateNodeRole(int term,int votesCount,ElectionTimeout electionTimeout) {super(RoleName.CANDIDATE, term);        this.votesCount = votesCount;        this.electionTimeout = electionTimeout;    }/**     * 增加1票构造     * @param term     * @param electionTimeout     */    public CandidateNodeRole(int term,ElectionTimeout electionTimeout) {this(term,1,electionTimeout);    }/**     * 票数+1     * @param electionTimeout     * @return     */    public CandidateNodeRole increaseVotesCount(ElectionTimeout electionTimeout) {this.electionTimeout.cancel();        return new CandidateNodeRole(term,votesCount + 1,electionTimeout);    }@Override    public void cancelTimeoutOrTask() {electionTimeout.cancel();    }@Override    public NodeId getLeaderId(NodeId selfId) {return null;    }}

主节点角色

/** * 日志复制任务 */@Slf4j@RequiredArgsConstructorpublic class LogReplicationTask {private final ScheduledFuture scheduledFuture;    public static final LogReplicationTask NONE = new LogReplicationTask(new NullScheduledFuture());    public void cancel() {log.debug("cancel log replication task");        scheduledFuture.cancel(false);    }@Override    public String toString() {return "LogReplicationTask{delay=" + scheduledFuture.getDelay(TimeUnit.MILLISECONDS) + "}";    }}

/** * 主节点角色 */@ToStringpublic class LeaderNodeRole extends AbstractNodeRole {//日志复制任务    private final LogReplicationTask logReplicationTask;    public LeaderNodeRole(int term,LogReplicationTask logReplicationTask) {super(RoleName.LEADER, term);        this.logReplicationTask = logReplicationTask;    }@Override    public void cancelTimeoutOrTask() {logReplicationTask.cancel();    }@Override    public NodeId getLeaderId(NodeId selfId) {return selfId;    }}

定时调度接口

/** * 定时器 */public interface Scheduler {/**     * 创建日志复制定时任务     * @param task     * @return     */    LogReplicationTask scheduleLogReplicationTask(Runnable task);    /**     * 创建选举超时器     * @param task     * @return     */    ElectionTimeout scheduleElectionTimeout(Runnable task);    /**     * 关闭定时器     * @throws InterruptedException     */    void stop() throws InterruptedException;}

定时调度接口实现类

/** * 节点配置类 */@Datapublic class NodeConfig {//最小选举超时时间    private int minElectionTimeout = 3000;    //最大选举超时时间内    private int maxElectionTimeout = 4000;    //初次日志复制延迟时间    private int logReplicationDelay = 0;    //日志复制间隔    private int logReplicationInterval = 1000;}

@Slf4jpublic class DefaultScheduler implements Scheduler {//最小选举超时时间    private final int minElectionTimeout;    //最大选举超时时间内    private final int maxElectionTimeout;    //初次日志复制延迟时间    private final int logReplicationDelay;    //日志复制间隔    private final int logReplicationInterval;    //随机数生成器    private final Random electionTimeoutRandom;    //定时任务线程池    private final ScheduledExecutorService scheduledExecutorService;    public DefaultScheduler(int minElectionTimeout,int maxElectionTimeout,int logReplicationDelay,                            int logReplicationInterval) {if (minElectionTimeout <= 0 || maxElectionTimeout <= 0                || minElectionTimeout > maxElectionTimeout) {throw new IllegalArgumentException("election timeout should not be 0 or min > max");        }if (logReplicationDelay < 0 || logReplicationInterval <= 0) {throw new IllegalArgumentException("log replication delay < 0 or log replication interval <= 0");        }this.minElectionTimeout = minElectionTimeout;        this.maxElectionTimeout = maxElectionTimeout;        this.logReplicationDelay = logReplicationDelay;        this.logReplicationInterval = logReplicationInterval;        electionTimeoutRandom = new Random();        scheduledExecutorService = Executors.newSingleThreadScheduledExecutor(r -> new Thread(r,"scheduler"));    }public DefaultScheduler(NodeConfig config) {this(config.getMinElectionTimeout(), config.getMaxElectionTimeout(), config.getLogReplicationDelay(),                config.getLogReplicationInterval());    }@Override    public LogReplicationTask scheduleLogReplicationTask(Runnable task) {        ScheduledFuture scheduledFuture = scheduledExecutorService.scheduleWithFixedDelay(task,                logReplicationDelay,logReplicationInterval,TimeUnit.MILLISECONDS);        return new LogReplicationTask(scheduledFuture);    }@Override    public ElectionTimeout scheduleElectionTimeout(Runnable task) {int timeout = electionTimeoutRandom.nextInt(maxElectionTimeout - minElectionTimeout)                + minElectionTimeout;        ScheduledFuture scheduledFuture = scheduledExecutorService.schedule(task,timeout, TimeUnit.MILLISECONDS);        return new ElectionTimeout(scheduledFuture);    }@Override    public void stop() throws InterruptedException {log.debug("stop scheduler");        scheduledExecutorService.shutdown();        scheduledExecutorService.awaitTermination(1,TimeUnit.SECONDS);    }}

/** * 空调度 */public class NullScheduledFuture implements ScheduledFuture