从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(){
// 变为leader
RaftNodeInfo.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);
// 任期+1
RaftNodeInfo.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)) {
// 投票通过 升级为leader
StateMachines.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();
// 投票过程中 可能又收到了心跳或者日志,状态已经变为follow
if (!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);
}
// 投票过程中 可能状态又已经变为follow
if (!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的存储设计是真的厉害,做完这个又收获不少