从0-1实现简易Raft分布式共识算法

一、Raft前置简介

Raft目前是最著名的分布式共识性算法，被广泛的应用在各种分布式框架、组件中，如Redis、RocketMq、Kafka、Nacos（CP）等

根据Raft论文，可将Raft拆分为如下4个功能模块：

领导者选举
日志同步、心跳
持久化
日志压缩，快照（本文未实现）

这4个模块彼此并不完全独立，如日志的同步情况左右着领导者选举，快照也影响着日志同步等等；为了前后的递进性，对于一些功能的实现，可能会出现改动和优化，比如日志同步实现后，在数据持久化部分又会对同步做一些优化，提高主、从节点日志冲突解决的性能。

这里就不再过多的介绍了，看本文之前请先简单了解一下Raft算法，提供如下资料：

Raft算法在线动画演示
图解分布式共识算法Paxos协议
浅谈分布式一致性：Raft 与 SOFAJRaft
深入剖析共识性算法 Raft
深入解读Raft算法与etcd工程实现
Raft一致性算法论文-译文
SOFA-JRaft：蚂蚁金服的Raft算法实现库（JAVA版）

本文实现不完全和Raft论文一致，做了不少改动，核心思想不变，请悉知！！

二、功能流程简介

你看完上述资料，应该对Raft有一个基本了解了，本文我们实现了一个Raft算法下的简易版的KV存储，我将它拆分成一下几个角色：

RPC模块：复制各节点间的信息传递，如心跳、日志、选举等等

节点模块：节点有三种状态leader、follow、candidate，每种状态下所要做的事是不一样的

状态机：负责节点状态的变更，日志持久化一致性处理，投票一致性处理

定时任务：leader需要定时发送心跳，follow需要定时检测leader是否存活等

日志模块：日志需要持久化在本地文件，还需要给其他节点同步

以上几个角色相互配合，实现以下几个主要功能流程：

1.选举流程

实现细节下面深究，这里暂不过多介绍，简单了解一下大致流程，大体就是：

Follow节点发现Leader节点挂了，则升级为Candidate节点发起投票
其他Follow节点收到投票请求后，根据条件判断是否投票给它，True或者False
Candidate一旦收到的投票通过请求过半，则升级为Leader
升级Leader后发送心跳，阻止其他Follow变成Candidate

在这里插入图片描述

2.心跳流程

注意：这里和原文有区别，我将心跳和日志做了拆分，不再耦合了，因为我觉得在没有客户端请求的情况下，记录这些心跳日志没有意义，在没有数据日志或者说数据日志水平都是一样的情况下，谁做Leader我觉得都OK

实现细节下面深究，这里暂不过多介绍，简单了解一下大致流程，大体就是：

Leader会定时发送心跳请求给Follow，告诉它我还活着，防止它篡位
Follow收到心跳后返回一个心跳响应
Leader收到的心跳响应没有过半则自动降级成为Follow停止对外服务

（为什么要心跳响应，还要自动降级？后面咱们细说）

在这里插入图片描述

3.KV客户端请求流程

因为我们要做的是一个简易版KV嘛，那肯定有客户端发送命令嘛对不对：

客户端发送SET或者GET命令，集群返回成功或者数据
发送SET命令，只有Leader会处理，同步给其他Follow，然后根据结果返回成功还是失败
发送GET命令，目前也只有Leader会处理，返回对应数据，没有就null（GET没有日志产生）

节点间日志的同步持久化后面细说，这里也看的出来为什么分布式体系下CAP不能共存了，你想要高可用，性能好，就必须在请求leader刷盘成功后返回甚至异步刷盘，这就必然导致可能存在数据丢失或者主从数据不一致的情况，如果你想要一致性，就必须在节点日志都同步完成后才返回(下面我们将日志同步流程）

在这里插入图片描述

4.日志同步流程

上面说过了，我们将心跳和日志做了拆分，只有客户端请求SET命令才会产生日志

Leader收到客户端请求后，先预提交到内存中，后发送预提交命令给所有Follow
Follow收到Leader的预提交命令同样先提交到内存，然后响应Leader
Leader一旦收到超过半数的Follow响应则执行刷盘持久化，否则给客户端响应失败
Leader刷盘成功后，给所有Follow发送刷盘请求，然后给客户端响应成功（无需关心Follow刷盘结果）

这就是很典型的CP流程，保持了一致性和数据不丢失，但大大降低了性能（发现没有尽管这样做，依旧可能存在Follow数据丢失的情况，比如：我是新加入的Follow节点、Follow节点刷盘失败等等情况，那该怎么办呢，我们下面接着来补充）

在这里插入图片描述

5.日志校验流程

正如上所说，日志依旧存在丢失的风险，我们需要做一个日志校验定时任务，定时校验日志是否丢失，由于这个和日志的设计息息相关，所以我们后面在细说，这里简单过一遍流程

follow会有一个定时任务，定时Check日志文件，寻找缺失的日志
如果有则拿到缺失的日志发送拉取请求到Leader，获取对应的日志
然后填充进日志文件，这样就一定保持了和Leader日志数据对齐了

难道每次都要从头到尾扫描一次文件吗？当然不是，扫描过的不需要扫描，有checkPoint，每次只是从checkPoint扫描到lastLogIndex

在这里插入图片描述

三、模块简介

1.RPC模块

这里我们采用Netty框架来做，每个节点即是Client又是Server

按原Raft算法来说，一共有以下几种RPC类型的通信：

RequestVote RPC - 请求投票 RPC，由 Candidate 在选举期间发起。
AppendEntries RPC - 附加条目 RPC，由 Leader 发起，用来复制日志和提供一种心跳机制。

但是我将它进行了一个拆分，拆分的更细了：

RequestVoteRPC-请求投票 RPC，由 Candidate 在选举期间发起。
RequestVoteResult-投票响应RPC，由follow投票
HeartBeatRequest-心跳RPC，由leader定时不间断发起
HeartBeatResult-心跳响应RPC，由follow响应
AppendEntriesPreCommit-日志预提交RPC，由leader发起预提交
AppendEntriesPreCommitResult-日志预提交响应RPC，由follow响应
AppendEntriesCommit-日志提交RPC，预提交成功后，leader会发起真正提交的命令
LogIndexPull-日志拉取RPC，follow定时检测发现自身存在日志丢失，向leader主动拉取日志
LogIndexPullResult-日志拉取响应RPC，leader发现follow存在日志缺失，把日志发给follow
ClientRequest-客户端请求RPC，KV存储的客户端，向集群发出的命令
ClientResponse-客户端请求响应RPC，对客户端的响应

分别对应着一个实体类：

在这里插入图片描述

RPC整体的编解码设计，序列化等等，都和我之前写的RPC框架差不多，这里就不在过多的介绍了，有兴趣可以看看我的：如何从0-1手写一个RPC框架

这里只介绍一下相比原来做出的调整，原来RPC框架传输的实体是固定的，而现在多了很多，而且大量涉及到同步请求返回，所以相比原来新增了泛型的处理，如下示例，两行代码就搞定了一次请求：

RpcSession<ClientResponse, ClientRequest> rpcSession = RpcSessionFactory.<ClientResponse, ClientRequest>openSession(serverConfig, clientRequest);
ClientResponse clientResponse = rpcSession.<ClientResponse>syncSend(4000L);

同时支持：同步等待、超时等待、异步三种请求方式：

public interface RpcSession<R,T>{<R> R syncSend();<R> R syncSend(long timeout);void asyncSend();
}

感兴趣的建议自己看看，RPC所在目录和Netty所有Handler如下：

在这里插入图片描述

2.节点模块

节点有三种类型，leader、candidate、follow，所以我这抽象出一个节点接口，三种实现，一个统一对外服务，一个全局节点信息类

一个节点接口

public interface RaftNode {/** 客户端的请求,会产生日志 : 只有leader才会处理，follow返回leader地址，candidate拒绝 */ClientResponse clientRequestHandler(ClientRequest command,List<ServerConfig> serverConfigs) throws ExecutionException, InterruptedException;/** leader发来的log预处理:会先缓存 */AppendEntriesPreCommitResult logPreCommitHandler(AppendEntriesPreCommit appendEntriesPreCommit);/** leader发来的log提交请求 */void logCommitHandler(AppendEntriesCommit appendEntriesCommit);/** follow发来的log拉取请求 */LogIndexPullResult sendLogPullRequest(List<Long> pullLogIndex);/** leader要处理follow的拉取请求 */LogIndexPullResult logPullRequestHandler(LogIndexPull logIndexPull);/** 发起投票 : 只有候选者 才会发起 */boolean callVoteRequest(List<ServerConfig> serverConfigs) throws ExecutionException, InterruptedException;/** 投票请求处理 */RequestVoteResult voteRequestHandler(RequestVoteRPC voteRPC);/** 发起心跳 : 只有领导才会发起心跳 阻止其他节点成为候选人*/boolean callHeartBeatRequest(List<ServerConfig>serverConfigs) throws ExecutionException, InterruptedException;/** 心跳请求处理 : 只有追随者/候选人才会处理*/HeartBeatResult heartBeatHandler(HeartBeatRequest heartBeatRequest);}

三种实现

在这里插入图片描述

一个对外服务

public class RaftNodeService {private static final Logger log = LoggerFactory.getLogger(RaftNodeService.class);// 心跳间隔时间private final static long INTERVAL_TIME = 1500L;private static Map<NodeStatusEnums, RaftNode> raftNodeMap = new ConcurrentHashMap<>(8);static {raftNodeMap.put(NodeStatusEnums.LEADER, new LeaderRaftNode());raftNodeMap.put(NodeStatusEnums.CANDIDATE, new CandidateRaftNode());raftNodeMap.put(NodeStatusEnums.FOLLOW, new FollowRaftNode());}/*** 节点信息初始化*/public static void raftNodeInit(ServerConfig self, List<ServerConfig> clusterConfig) {RaftNodeInfo.getInstance().setSelf(self);RaftNodeInfo.getInstance().setClusterConfig(clusterConfig);RaftNodeInfo.getInstance().setCurrentNodeStatus(NodeStatusEnums.FOLLOW);createElectionTask();}/*** 发送心跳*/public synchronized static void sendHeartBeat() {RaftNode raftNode = raftNodeMap.get(RaftNodeInfo.getInstance().getCurrentNodeStatus());boolean result = false;try {result = raftNode.callHeartBeatRequest(RaftNodeInfo.getInstance().getClusterConfig());} catch (ExecutionException e) {log.debug(" {}: 完了，作为leader发送心跳失败了:{}", RaftNodeInfo.getInstance().getSelf().toString(), e.getMessage(), e);} catch (InterruptedException e) {log.debug(" {}: 完了，作为leader发送心跳失败了:{}", RaftNodeInfo.getInstance().getSelf().toString(), e.getMessage(), e);}if (!result) {// 代表心跳失败了，状态已经变更了// 需要停止心跳，开启心跳检测heartBeatTestDestroy();createElectionTask();}}/*** 心跳处理*/public static void heartBeatHandler(HeartBeatRequest request, Channel channel) {ThreadPoolUtils.nettyServerAsyncPool.execute(() -> {RaftNode raftNode = raftNodeMap.get(RaftNodeInfo.getInstance().getCurrentNodeStatus());channel.writeAndFlush(new RpcRemoteMsg<HeartBeatResult>(raftNode.heartBeatHandler(request)));// 收到了心跳，所以就要停止当前心跳的检测，然后重新开启一个检测任务createElectionTask();});}// 命令合规性校验 目前就get set 随便校验一下public static boolean commandCheck(String command) {if (command == null || !"SET_GET".contains(command.split(" ")[0]) || command.split(" ").length < 2) {return false;}return true;}/*** 客户端的请求，以KV为例 就是set命令 , 这里请求返回就简陋一点*/public static void clientRequestHandler(ClientRequest request, Channel channel) {ThreadPoolUtils.nettyServerAsyncPool.execute(() -> {ClientResponse clientResponse = ClientResponse.builder().build();clientResponse.setRequestId(request.getRequestId());if (!commandCheck(request.getCommand())) {clientResponse.setCode(401);clientResponse.setMsg("命令格式不正确");channel.writeAndFlush(new RpcRemoteMsg<ClientResponse>(clientResponse));return;}// 只有set命令才需要发送日志，get命令直接取数据就行了String[] command = request.getCommand().split(" ");if (command[0].equals("SET")) {RaftNode raftNode = raftNodeMap.get(RaftNodeInfo.getInstance().getCurrentNodeStatus());try {channel.writeAndFlush(new RpcRemoteMsg<ClientResponse>(raftNode.clientRequestHandler(request, RaftNodeInfo.getInstance().getClusterConfig())));return;} catch (ExecutionException e) {log.debug(" {}: 日志提交失败了:{}", request.getCommand(), e.getMessage(), e);clientResponse.setCode(500);clientResponse.setMsg(e.getMessage());channel.writeAndFlush(new RpcRemoteMsg<ClientResponse>(clientResponse));} catch (InterruptedException e) {log.debug(" {}: 日志提交失败了:{}", RaftNodeInfo.getInstance().getSelf().toString(), e.getMessage(), e);clientResponse.setCode(500);clientResponse.setMsg(e.getMessage());channel.writeAndFlush(new RpcRemoteMsg<ClientResponse>(clientResponse));}} else {// get命令直接取值clientResponse.setCode(200);clientResponse.setData(RaftNodeInfo.getInstance().getLogManage().getDataByKey(command[1]));channel.writeAndFlush(new RpcRemoteMsg<ClientResponse>(clientResponse));}});}/*** Log预提交请求*/public static void logPreCommitHandler(AppendEntriesPreCommit request, Channel channel) {ThreadPoolUtils.nettyServerAsyncPool.execute(() -> {RaftNode raftNode = raftNodeMap.get(RaftNodeInfo.getInstance().getCurrentNodeStatus());channel.writeAndFlush(new RpcRemoteMsg<AppendEntriesPreCommitResult>(raftNode.logPreCommitHandler(request)));});}/*** Log提交请求*/public static void logCommitHandler(AppendEntriesCommit request) {ThreadPoolUtils.nettyServerAsyncPool.execute(() -> {RaftNode raftNode = raftNodeMap.get(RaftNodeInfo.getInstance().getCurrentNodeStatus());raftNode.logCommitHandler(request);// 收到了日志，所以就要停止当前心跳的检测，然后重新开启一个检测任务createElectionTask();});}/*** 发起投票*/public synchronized static void sendCallVote() {RaftNode raftNode = raftNodeMap.get(RaftNodeInfo.getInstance().getCurrentNodeStatus());boolean result = false;try {result = raftNode.callVoteRequest(RaftNodeInfo.getInstance().getClusterConfig());} catch (ExecutionException e) {StateMachines.becomeFollow(RaftNodeInfo.getInstance().getCurrentTerm(), null, null);log.debug(" {}: 完了，作为candidate发起投票失败了:{}", RaftNodeInfo.getInstance().getSelf().toString(), e.getMessage(), e);} catch (InterruptedException e) {StateMachines.becomeFollow(RaftNodeInfo.getInstance().getCurrentTerm(), null, null);log.debug(" {}: 完了，作为candidate发起投票失败了:{}", RaftNodeInfo.getInstance().getSelf().toString(), e.getMessage(), e);}if (!result) {// 代表发起投票失败了，状态已经变更了// 需要重新开启一个检测任务createElectionTask();return;}// 投票成功了 需要开启心跳任务createHearBeatTask();}/*** 发起投票请求处理*/public synchronized static void callVoteHandler(RequestVoteRPC requestVoteRPC, Channel channel) {ThreadPoolUtils.nettyServerAsyncPool.execute(() -> {RaftNode raftNode = raftNodeMap.get(RaftNodeInfo.getInstance().getCurrentNodeStatus());channel.writeAndFlush(new RpcRemoteMsg<RequestVoteResult>(raftNode.voteRequestHandler(requestVoteRPC)));});}/*** 发起Log拉取请求*/public synchronized static LogIndexPullResult sendLogPullRequest(List<Long> pullLogIndex) {if (CollectionUtil.isNotEmpty(pullLogIndex)) {RaftNode raftNode = raftNodeMap.get(RaftNodeInfo.getInstance().getCurrentNodeStatus());return raftNode.sendLogPullRequest(pullLogIndex);}return null;}/*** 发起Log拉取请求处理*/public synchronized static void logPullRequestHandler(LogIndexPull logIndexPull, Channel channel) {ThreadPoolUtils.nettyServerAsyncPool.execute(() -> {RaftNode raftNode = raftNodeMap.get(RaftNodeInfo.getInstance().getCurrentNodeStatus());log.debug("{}:拉取日志：{}",channel.remoteAddress(),JSONObject.toJSON(logIndexPull));channel.writeAndFlush(new RpcRemoteMsg<LogIndexPullResult>(raftNode.logPullRequestHandler(logIndexPull)));});}/*** 销毁并创建心跳检测任务*/public static void createElectionTask() {long randomTime = getRandomTime();final long intervalTime = INTERVAL_TIME + randomTime;// 先销毁之前的electionTaskDestroy();//开启新的ScheduledFuture<?> schedule = ThreadPoolUtils.hearBeatAsyncPool.schedule(new ElectionTask(intervalTime), intervalTime, TimeUnit.MILLISECONDS);RaftNodeInfo.getInstance().setElectionTask(schedule);}/*** 销毁并创建心跳任务*/public static void createHearBeatTask() {// 先销毁之前的heartBeatTestDestroy();//开启新的ScheduledFuture<?> schedule = ThreadPoolUtils.hearBeatAsyncPool.scheduleAtFixedRate(new HeartBeatTask(), 0L, INTERVAL_TIME, TimeUnit.MILLISECONDS);RaftNodeInfo.getInstance().setElectionTask(schedule);}public static long getRandomTime() {// 要比心跳慢一点return RandomUtil.randomLong(250L, 1000L);}/*** 销毁心跳检测任务*/public static void electionTaskDestroy() {if (null != RaftNodeInfo.getInstance().getElectionTask()) {RaftNodeInfo.getInstance().getElectionTask().cancel(true);RaftNodeInfo.getInstance().setElectionTask(null);}}/*** 销毁心跳任务*/public static void heartBeatTestDestroy() {if (null != RaftNodeInfo.getInstance().getHeartBeatTask()) {RaftNodeInfo.getInstance().getHeartBeatTask().cancel(true);RaftNodeInfo.getInstance().setHeartBeatTask(null);}}

一个全局节点信息类

public class RaftNodeInfo {/*** 自己*/private ServerConfig self;/*** 集群其他节点信息*/private List<ServerConfig> clusterConfig;/*** 当前节点状态 默认FOLLOW*/private volatile NodeStatusEnums currentNodeStatus = NodeStatusEnums.FOLLOW;/*** 当前节点任期*/private volatile long currentTerm = 0L;/*** 当前leader*/private volatile String currentLeaderId;/*** 最后日志索引 已提交的*/private volatile long lastLogIndex = 0L;/*** 最后的日志任期 这我这没用到*/private volatile long lastLogTerm = 0L;/*** 当前任期给谁投过票*/private volatile String voteFor;/*** 最近更新时间  心跳或者日志更新**/private volatile long lastUpdateTime = 0L;/*** 心跳任务**/private ScheduledFuture heartBeatTask;/*** 心跳检测任务**/private ScheduledFuture electionTask;/*** 日志管理**/private LogManage logManage;/*** 日志文件**/private String logPath;
}

3.状态机

提供节点状态变更、心跳结果处理、投票结果处理、日志一致性处理

public class StateMachines {private static final Logger log = LoggerFactory.getLogger(StateMachines.class);/** 候选人-》leader */public static void becomeLeader(){// 变为leaderRaftNodeInfo.getInstance().setCurrentNodeStatus(NodeStatusEnums.LEADER);// leader设置为自己RaftNodeInfo.getInstance().setCurrentLeader(RaftNodeInfo.getInstance().getSelf().toString());// 票清了RaftNodeInfo.getInstance().setVoteFor(null);}/** follow-》候选人 */public static void becomeCandidate(){// 变为候选人RaftNodeInfo.getInstance().setCurrentNodeStatus(NodeStatusEnums.CANDIDATE);// 任期+1RaftNodeInfo.getInstance().setCallVoteTerm();// 给自己投一票RaftNodeInfo.getInstance().setVoteFor(RaftNodeInfo.getInstance().getSelf().toString());}/** 候选人、leader->follow */public static void becomeFollow(long term,String leaderId,String voteFor){RaftNodeInfo.getInstance().setCurrentNodeStatus(NodeStatusEnums.FOLLOW);RaftNodeInfo.getInstance().setCurrentLeader(leaderId);RaftNodeInfo.getInstance().setCurrentTerm(term);RaftNodeInfo.getInstance().setVoteFor(voteFor);RaftNodeInfo.getInstance().setLastUpdateTime(System.currentTimeMillis());}/** 投票结果一致性处理 */public static boolean voteResultHandler(List<Future<RequestVoteResult>> taskList,Integer nodeNum) throws ExecutionException, InterruptedException {int voteNum = 0;for (Future<RequestVoteResult> future : taskList) {RequestVoteResult voteResult = future.get();// 判断leader是否还存活 存活的话肯定要把我给否了呀if (leaderIsLive(voteResult)) {return false;}if(voteResult!=null){log.debug("投票结果,我的term：{} ，结果：{}",RaftNodeInfo.getInstance().getCurrentTerm(), JSONObject.toJSON(voteResult));}if (null != voteResult && voteResult.isVoteGranted()) {voteNum++;}}if (voteNum != 0 && voteNum >= (nodeNum / 2)) {// 投票通过 升级为leaderStateMachines.becomeLeader();log.debug(" {}: 哈哈哈，我升级为leader啦", RaftNodeInfo.getInstance().getSelf().toString());return true;} else {// 投票不通过，退成follow 继续苟着StateMachines.becomeFollow(RaftNodeInfo.getInstance().getCurrentTerm(), null, null);log.debug(" {}: 完了，这帮人不支持我，等待机会再试", RaftNodeInfo.getInstance().getSelf().toString());return false;}}// 判断leader是否还存活 存活的话肯定要把我给否了呀private static boolean leaderIsLive(RequestVoteResult voteResult) {if (null != voteResult && StrUtil.isNotEmpty(voteResult.getLeaderId())) {// 被leader一票否决，退成follow 继续苟着StateMachines.becomeFollow(voteResult.getTerm(), voteResult.getLeaderId(), null);return true;}return false;}/** 心跳结果一致性处理 */public static boolean heartBeatResultHandler(List<Future<HeartBeatResult>> taskList,Integer nodeNum) throws ExecutionException, InterruptedException {int responseNum = 0;for (Future<HeartBeatResult> future : taskList) {HeartBeatResult heartBeatResult = future.get();if (null != heartBeatResult) {responseNum++;}}if (responseNum != 0 && responseNum >= (nodeNum / 2)) {log.debug("{}: 万众一心，我再接再厉", RaftNodeInfo.getInstance().getSelf().toString());return true;} else {// 没有应答或者应答数量小于一半 就退化为候选者，并停止对外提供服务// 状态变更StateMachines.becomeFollow(RaftNodeInfo.getInstance().getCurrentTerm(), null, null);log.debug("{}: 我找不到追随者了，我暂时停止对外服务", RaftNodeInfo.getInstance().getSelf().toString());return false;}}/** 日志预提交结果 */public static boolean logPreCommitHandler(List<Future<AppendEntriesPreCommitResult>> taskList, Integer nodeNum) throws ExecutionException, InterruptedException {int responseNum = 0;for (Future<AppendEntriesPreCommitResult> future : taskList) {AppendEntriesPreCommitResult preCommitResult = future.get();if (null != preCommitResult && preCommitResult.isSuccess()) {responseNum++;}}return responseNum != 0 && responseNum >= (nodeNum / 2);}
}

4.日志模块

public interface LogManage extends ResourceLifeCycle{/** leader预提交 */long preCommitLog(LogEntity logEntity);/** follow预提交 */void preCommitLog(long preCommitLogId,LogEntity logEntity);/** 缓存移除 */void cacheLogRemove(long cacheLogId);/** leader日志提交 */long commitLog(long cacheLogId);/** follow日志提交 */void commitLog(long cacheLogId,long logIndex);/** follow日志Check */void logIndexCheck();/** 根据日志索引获取日志内容 */LogEntity getLogEntityByIndex(long logIndex, RandomAccessFile file);/** 命令数据处理 */void dataHandler(String command);/** 根据Key获取数据 */String getDataByKey(String key);
}

5.定时任务

在这里插入图片描述

ElectionTask：心跳检测任务，不通过则升级为Candidate
HeartBeatTask：心跳任务，不断给Follow发送心跳，阻止其成为Candidate
LogIndexCheckTask：Follow日志Check定时任务

四、核心流程介绍

其实流程图已经很清楚了，这里挑部分来聊聊

1.选举

目前心跳设置的时间为1500ms，心跳检测的时间为1750ms+0-750ms随机数（之前随机数设置的很短，算上网络延迟等因素，导致两个Candidate同任期的几率非常之高），follow收到心跳会更新lastUpdateTIme，而心跳检测则会检测这个时间到当前时间是否超过检测时间间隔，超过了则会变成candidate发起选举

在这里插入图片描述

CandidateRaftNode：发起选举RPC

选举RPC实体类：

public class RequestVoteRPC extends RpcMsgId implements Serializable {/** 候选人的任期号  */private long term;/** 请求选票的候选人的 Id(ip:selfPort) */private String candidateId;/** 候选人的最后日志条目的索引值 */private long lastLogIndex;/** 候选人最后日志条目的任期号  */private long lastLogTerm;}

选举方法：

    public boolean callVoteRequest(List<ServerConfig> serverConfigs) throws ExecutionException, InterruptedException {if (CollectionUtil.isEmpty(serverConfigs)) {StateMachines.becomeFollow(RaftNodeInfo.getInstance().getCurrentTerm(), null, null);log.error("只有一个节点，还发起什么投票？");return false;}// candidate 会发起投票请求RaftNodeInfo instance = RaftNodeInfo.getInstance();// 投票过程中 可能又收到了心跳或者日志，状态已经变为followif (!NODE_TYPE.equals(RaftNodeInfo.getInstance().getCurrentNodeStatus())) {return false;}log.debug(" {}: 哈哈哈，我发起了投票", RaftNodeInfo.getInstance().getSelf().toString());List<Future<RequestVoteResult>> taskList = new ArrayList<>(serverConfigs.size());// 加上自己的一票 需要 大于= n/2+1// 所以直接 >= n/2 就算通过了// 但是注意此时如果已经存在leader，日志数又不比当前leader大，所以leader还是leader 具有一票否决权for (ServerConfig serverConfig : serverConfigs) {Future<RequestVoteResult> voteResultFuture = ThreadPoolUtils.sendAsyncMsgPool.submit(() -> {// 构建投票RequestVoteRPC voteRPC = RequestVoteRPC.builder().candidateId(instance.getSelf().toString()).term(instance.getCurrentTerm())  // 成为候选 的时候任期就+1了.lastLogIndex(instance.getLastLogIndex()).build();RpcSession<RequestVoteResult, RequestVoteRPC> voteRPCRpcSession = RpcSessionFactory.<RequestVoteResult, RequestVoteRPC>openSession(serverConfig, voteRPC);return voteRPCRpcSession == null ? null : voteRPCRpcSession.syncSend(1000L);});taskList.add(voteResultFuture);}// 投票过程中 可能状态又已经变为followif (!NODE_TYPE.equals(RaftNodeInfo.getInstance().getCurrentNodeStatus())) {return false;}return StateMachines.voteResultHandler(taskList, serverConfigs.size());}

Follow选举响应：

任期比我大我就同意
任期跟我一样，记录的日志比我多而且我没有投过票我也同意

（Follow同一个任期内只能投一票）

    public RequestVoteResult voteRequestHandler(RequestVoteRPC voteRPC) {// follow 需要处理投票请求RaftNodeInfo instance = RaftNodeInfo.getInstance();RequestVoteResult voteResult = RequestVoteResult.builder().term(instance.getCurrentTerm()).build();voteResult.setRequestId(voteRPC.getRequestId());// 1.任期比我大，我直接就同意if (voteRPC.getTerm() > instance.getCurrentTerm()) {return agreeVote(voteResult, voteRPC);}// 2.任期跟我一样，记录的日志比我多 而且 我没有投过票// 我只能投一票if ((voteRPC.getTerm() == instance.getCurrentTerm() && voteRPC.getLastLogIndex() >= instance.getLastLogIndex())&& (instance.getVoteFor() == null || instance.getVoteFor().equals(voteRPC.getCandidateId()))) {return agreeVote(voteResult, voteRPC);}voteResult.setTerm(instance.getCurrentTerm());voteResult.setVoteGranted(false);log.info(" {}: 我身为现任Follow，我不认可你的实力，我不能给你投票：{}", instance.getSelf().toString(), voteRPC.getCandidateId());return voteResult;}private RequestVoteResult agreeVote(RequestVoteResult voteResult, RequestVoteRPC voteRPC) {voteResult.setTerm(RaftNodeInfo.getInstance().getCurrentTerm());voteResult.setVoteGranted(true);RaftNodeInfo.getInstance().setCurrentTerm(voteRPC.getTerm());RaftNodeInfo.getInstance().setVoteFor(voteRPC.getCandidateId());log.info(" {}: 我身为现任Follow，我认可你的实力，我给你投票：{}", RaftNodeInfo.getInstance().getSelf().toString(), voteRPC.getCandidateId());return voteResult;}

Leader响应：

leader有没有可能收到投票？有可能！假设某一个Follow延迟收到心跳或者没有收到心跳就会发起，那leader就会收到它发起的投票，那怎么办？判断任期和日志，任期和日志都比Leader大则Leader需要退位，否则Leader应该具有一票否决权（这样就防止了某个follow无限发起投票，任期无限+1这种情况）

一个candidate任期非常大的时候，其他follow必然会给他投票，那这样就升为leader就导致了同时存在两个leader的情况，所以这时候的当期leader应该具有一票否决权

    public RequestVoteResult voteRequestHandler(RequestVoteRPC voteRPC) {// leader 有可能收到 候选者的投票申请RaftNodeInfo instance = RaftNodeInfo.getInstance();RequestVoteResult requestVoteResult = RequestVoteResult.builder().build();requestVoteResult.setRequestId(voteRPC.getRequestId());// 候选人的任期比我大 而且日志还比我大 说明我已经out了，我需要退位if (voteRPC.getTerm() >= instance.getCurrentTerm() && voteRPC.getLastLogIndex() > instance.getLastLogIndex()) {// 状态变更StateMachines.becomeFollow(voteRPC.getTerm(), voteRPC.getCandidateId(), null);requestVoteResult.setTerm(voteRPC.getTerm());requestVoteResult.setVoteGranted(true);log.info(" {}: 我身为现任leader，我认可你的实力，我下位让贤：{}", instance.getSelf().toString(), voteRPC.getCandidateId());return requestVoteResult;}log.info(" {}: 我身为现任leader，不同你的上任请求：{}", instance.getSelf().toString(), voteRPC.getCandidateId());// 否则就不同意，而且你还得给我老实点requestVoteResult.setTerm(instance.getCurrentTerm());requestVoteResult.setVoteGranted(false);requestVoteResult.setLeaderId(instance.getSelf().toString());return requestVoteResult;}

2.心跳

在这里插入图片描述

心跳这里我做了一个响应降级的操作，其实正常是不需要的，我这的目的是防止网络分区！

假设原本是这样：

在这里插入图片描述

一旦网络分区则会变成这样，导致两个leader的出现，所以这时候心跳的响应就至关重要，一旦响应少于半数，则leader应该自动降级

在这里插入图片描述

LeaderRaftNode：发起心跳

public boolean callHeartBeatRequest(List<ServerConfig> serverConfigs) throws ExecutionException, InterruptedException {if (CollectionUtil.isEmpty(serverConfigs)) {StateMachines.becomeFollow(RaftNodeInfo.getInstance().getCurrentTerm(), null, null);log.debug(" {}: 只有一个leader，还发什么心跳？", RaftNodeInfo.getInstance().getSelf().toString());return false;}List<Future<HeartBeatResult>> taskList = new ArrayList<>(serverConfigs.size());// leader 需要发送心跳 防止网络分区，一旦心跳返回不足 n/2 则自动降级for (ServerConfig serverConfig : serverConfigs) {Future<HeartBeatResult> heartBeatResultFuture = ThreadPoolUtils.sendAsyncMsgPool.submit(() -> {HeartBeatRequest build = HeartBeatRequest.builder().leaderId(RaftNodeInfo.getInstance().getSelf().toString()).leaderLastCommitIndex(RaftNodeInfo.getInstance().getLastLogIndex()).term(RaftNodeInfo.getInstance().getCurrentTerm()).build();RpcSession<HeartBeatResult, HeartBeatRequest> heartBeatRequestRpcSession = RpcSessionFactory.<HeartBeatResult, HeartBeatRequest>openSession(serverConfig, build);return heartBeatRequestRpcSession == null ? null : heartBeatRequestRpcSession.syncSend(200L);});taskList.add(heartBeatResultFuture);}// 响应结果处理return StateMachines.heartBeatResultHandler(taskList, serverConfigs.size());}

3.日志

日志设计的非常之简陋，就不做过多的介绍了，本文目的还是以实现Raft为主，性能问题暂不考虑，不过还是说一下测试结果，因为KV存储，项目启动需要读取数据放入内存，目前读取50m左右文件10w条日志需要8s左右，肯定是不合理的，目前并没有做日志压缩和快照，也没有用零拷贝技术，因为不想搞的太过复杂

在这里插入图片描述

关于日志check，这里放上两种测试常见的结果

在这里插入图片描述

1.新的节点加入，需要拉取一次所有数据

在这里插入图片描述

2.日志中间缺失

在这里插入图片描述

两种情况都是没问题的！

五、遗留的问题

注意：尽管这样还是有几率导致数据丢失的！！！！

再次强调：本文不完全和Raft论文对标，加了不少个人的想法进去，所以在这个过程中都是遇到问题、思考问题、解决问题，这本就是一个学习的过程，目前最大的一个问题就是：

新加入的节点已经收到了Leader的数据，更新的lastCommitIndex，但是还没来得及向Leader同步以前的数据，而这时Leader挂了，所以这时候这个节点就有几率通过投票成为Leader，这时候数据就有几率丢失，文章中可能看不太出来，具体得看看代码，这算是一个很严重的BUG，各位想想可以怎么解决，而Raft又是怎么解决的？

当然可能还有其他问题，各位大佬如果知道的也可以提出来

六、总结

只有深入本质才能顺应发展，在分布式体系下，共识算法是必不可少的，光看不实践就容易眼高手低，当初我看Raft的时候也感觉挺简单的，不就是三种状态做不同的事，然后状态变更嘛，真正一做起来就发现好多细节都需要考虑，这还只是个demo，回头想想RocketMq和kafka的存储设计是真的厉害，做完这个又收获不少