SOFAJRaft源码阅读（壹）-模块启动过程
阳昊
本篇文章旨在分析SOFAJRaft中jraft-example模块的启动过程，由于SOFAJRaft在持续开源的过程中，所以无法保证示例代码永远是最新的，要是有较大的变动或者纰漏、错误的地方，欢迎大家留言讨论。
@Author：Akai-yuan
@博客地址：Akai-yuan的博客网站
更新时间：2023年1月20日
写在前面：
SOFAJRaft 是一个基于 RAFT 一致性算法的生产级高性能 Java 实现。
所以在阅读源码之前，你需要了解RAFT算法的一些基础知识，之前整合过一篇Raft算法的博客，里面包括了Leader选举、日志复制、成员变更等问题。详情可以查看这篇文章：Raft算法
另外，你还需要了解一些SOFAJRaft所依赖的框架，例如gRPC、Disruptor等等
其中Disruptor框架根据美团技术团队文章整合成一篇博客：SOFAJRaft源码阅读-框架Disruptor浅析
SOFAJRaft-Start Our Journey
1. 开门见山：main方法概览
我们在启动CounterServer的main方法的时候，首先进行参数校验，然后将传入的String[]类型参数args分别转化为日志存储路径、集群名字、当前节点IP和端口、集群节点IP和端口，并将这些启动参数注入到NodeOptions的属性中。
然后我们发现，他调用了CounterServer构造器和CounterGrpcHelper的blockUntilShutdown方法阻止进程退出。CounterServer构造器将是我们重点了解的内容。
blockUntilShutdown方法实际上就是利用反射获取server实例的shutdown和awaitTermination方法，并将这两个方法添加到钩子函数中阻止进程退出：
Runtime.getRuntime().addShutdownHook(shutdownHook)
关于钩子函数，可以查看我的博客： ShutdownHook如何优雅的停机
附上main方法的源码如下：
public static void main(final String[] args) throws IOException {
//参数长度校验
if (args.length != 4) {
System.out
.println("Usage : java com.alipay.sofa.jraft.example.counter.CounterServer {dataPath} {groupId} {serverId} {initConf}");
System.out
.println("Example: java com.alipay.sofa.jraft.example.counter.CounterServer /tmp/server1 counter 127.0.0.1:8081 127.0.0.1:8081,127.0.0.1:8082,127.0.0.1:8083");
System.exit(1);
}
//日志存储路径
final String dataPath = args[0];
//集群名字
final String groupId = args[1];
//当前节点IP和端口
final String serverIdStr = args[2];
//集群节点IP和端口
final String initConfStr = args[3];
final NodeOptions nodeOptions = new NodeOptions();
// 设置选举超时时间1秒
nodeOptions.setElectionTimeoutMs(1000);
// 关闭 CLI 服务
nodeOptions.setDisableCli(false);
// snapshot设置为每30秒一次
nodeOptions.setSnapshotIntervalSecs(30);
// 转化PeerId对象
final PeerId serverId = new PeerId();
// 转化失败抛出异常
if (!serverId.parse(serverIdStr)) {
throw new IllegalArgumentException("Fail to parse serverId:" + serverIdStr);
}
final Configuration initConf = new Configuration();
//将raft分组加入到Configuration的peers数组中
if (!initConf.parse(initConfStr)) {
throw new IllegalArgumentException("Fail to parse initConf:" + initConfStr);
}
// 设置初始集群配置
nodeOptions.setInitialConf(initConf);
// 启动raft server
final CounterServer counterServer = new CounterServer(dataPath, groupId, serverId, nodeOptions);
System.out.println("Started counter server at port:"
+ counterServer.getNode().getNodeId().getPeerId().getPort());
// GrpcServer 需要阻止进程退出
CounterGrpcHelper.blockUntilShutdown();
}
2. 对象转换：初始化PeerId对象
引子：在main方法中，我们可以看到，程序将String类型参数转换成了PeerId对象，那么接下来我们需要探究转换的具体过程。
在转换当前节点并初始化为一个PeerId对象的过程中，调用了PeerId中的parse方法：
public boolean parse(final String s) {
if (StringUtils.isEmpty(s)) {
return false;
}
final String[] tmps = Utils.parsePeerId(s);
if (tmps.length < 2 || tmps.length > 4) {
return false;
}
try {
final int port = Integer.parseInt(tmps[1]);
this.endpoint = new Endpoint(tmps[0], port);
switch (tmps.length) {
case 3:
this.idx = Integer.parseInt(tmps[2]);
break;
case 4:
if (tmps[2].equals("")) {
this.idx = 0;
} else {
this.idx = Integer.parseInt(tmps[2]);
}
this.priority = Integer.parseInt(tmps[3]);
break;
default:
break;
}
this.str = null;
return true;
} catch (final Exception e) {
LOG.error("Parse peer from string failed: {}.", s, e);
return false;
}
}
该方法内部又调用了工具类Utils.parsePeerId,最终达到的效果如下：
其中，a、b分别对应IP和Port端口号，组成了PeerId的EndPoint属性；c指代idx【同一地址中的索引，默认值为0】；d指代priority优先级【节点的本地优先级值，如果节点不支持优先级选择，则该值为-1】。
PeerId.parse("a:b") = new PeerId("a", "b", 0 , -1)
PeerId.parse("a:b:c") = new PeerId("a", "b", "c", -1)
PeerId.parse("a:b::d") = new PeerId("a", "b", 0, "d")
PeerId.parse("a:b:c:d") = new PeerId("a", "b", "c", "d")
3. 渐入佳境：构造CounterServer
引子：在main方法中，我们可以看到，进行初步的参数解析后，调用了CounterServer的构造器，要说这个构造器，第一次看里面的步骤确实会感觉挺复杂的，接下来我们一起分析一下源码。
CounterServer构造器的源码如下：
public CounterServer(final String dataPath, final String groupId, final PeerId serverId,
final NodeOptions nodeOptions) throws IOException {
// 初始化raft data path, 它包含日志、元数据、快照
FileUtils.forceMkdir(new File(dataPath));
// 这里让 raft RPC 和业务 RPC 使用同一个 RPC server, 通常也可以分开
final RpcServer rpcServer = RaftRpcServerFactory.createRaftRpcServer(serverId.getEndpoint());
// GrpcServer need init marshaller
CounterGrpcHelper.initGRpc();
CounterGrpcHelper.setRpcServer(rpcServer);
// 注册业务处理器
CounterService counterService = new CounterServiceImpl(this);
rpcServer.registerProcessor(new GetValueRequestProcessor(counterService));
rpcServer.registerProcessor(new IncrementAndGetRequestProcessor(counterService));
// 初始化状态机
this.fsm = new CounterStateMachine();
// 设置状态机到启动参数
nodeOptions.setFsm(this.fsm);
// 设置存储路径 (包含日志、元数据、快照)
// 日志(必须)
nodeOptions.setLogUri(dataPath + File.separator + "log");
// 元数据(必须)
nodeOptions.setRaftMetaUri(dataPath + File.separator + "raft_meta");
// 快照(可选, 一般都推荐)
nodeOptions.setSnapshotUri(dataPath + File.separator + "snapshot");
// 初始化 raft group 服务框架
this.raftGroupService = new RaftGroupService(groupId, serverId, nodeOptions, rpcServer);
// 启动
this.node = this.raftGroupService.start();
}
接下来仔细说说CounterServer的构造器里面具体做了什么。
4. 追根溯源：创建RpcServer
引子：CountServer构造器中调用的RaftRpcServerFactory.createRaftRpcServer()方法，底层到底是如何构造出一个RpcServer的呢，接下来会和大家讨论createRaftRpcServer()方法的具体实现
RaftRpcServerFactory.createRaftRpcServer(serverId.getEndpoint())方法：
createRaftRpcServer方法目前有createRaftRpcServer(final Endpoint endpoint)和
createRaftRpcServer(final Endpoint endpoint, final Executor raftExecutor,final Executor cliExecutor)两个重载方法，其实不管哪个方法，本质上实现过程都有如下两个步骤：
（1）首先调用了GrpcRaftRpcFactory的createRpcServer方法，这里涉及gRpc构建server的底层知识，有时间会再写一篇文章探究一下gRpc，这里可以简单理解为构建了一个rpc服务端。该方法实现如下：
public RpcServer createRpcServer(final Endpoint endpoint, final ConfigHelper<RpcServer> helper) {
final int port = Requires.requireNonNull(endpoint, "endpoint").getPort();
Requires.requireTrue(port > 0 && port < 0xFFFF, "port out of range:" + port);
final MutableHandlerRegistry handlerRegistry = new MutableHandlerRegistry();
final Server server = ServerBuilder.forPort(port) //
.fallbackHandlerRegistry(handlerRegistry) //
.directExecutor() //
.maxInboundMessageSize(RPC_MAX_INBOUND_MESSAGE_SIZE) //
.build();
final RpcServer rpcServer = new GrpcServer(server, handlerRegistry, this.parserClasses, getMarshallerRegistry());
if (helper != null) {
helper.config(rpcServer);
}
return rpcServer;
}
（2）紧接着调用addRaftRequestProcessors，这个方法为RpcServer添加RAFT和CLI服务核心请求处理器，关于RpcProcessor这个实体类，会在后面的文章中具体分析，这里可以先"不求甚解"。
//添加RAFT和CLI服务请求处理器
public static void addRaftRequestProcessors(final RpcServer rpcServer, final Executor raftExecutor,
final Executor cliExecutor) {
// 添加raft核心处理器
final AppendEntriesRequestProcessor appendEntriesRequestProcessor = new AppendEntriesRequestProcessor(
raftExecutor);
rpcServer.registerConnectionClosedEventListener(appendEntriesRequestProcessor);
rpcServer.registerProcessor(appendEntriesRequestProcessor);
rpcServer.registerProcessor(new GetFileRequestProcessor(raftExecutor));
rpcServer.registerProcessor(new InstallSnapshotRequestProcessor(raftExecutor));
rpcServer.registerProcessor(new RequestVoteRequestProcessor(raftExecutor));
rpcServer.registerProcessor(new PingRequestProcessor());
rpcServer.registerProcessor(new TimeoutNowRequestProcessor(raftExecutor));
rpcServer.registerProcessor(new ReadIndexRequestProcessor(raftExecutor));
// 添加raft cli服务处理器
rpcServer.registerProcessor(new AddPeerRequestProcessor(cliExecutor));
rpcServer.registerProcessor(new RemovePeerRequestProcessor(cliExecutor));
rpcServer.registerProcessor(new ResetPeerRequestProcessor(cliExecutor));
rpcServer.registerProcessor(new ChangePeersRequestProcessor(cliExecutor));
rpcServer.registerProcessor(new GetLeaderRequestProcessor(cliExecutor));
rpcServer.registerProcessor(new SnapshotRequestProcessor(cliExecutor));
rpcServer.registerProcessor(new TransferLeaderRequestProcessor(cliExecutor));
rpcServer.registerProcessor(new GetPeersRequestProcessor(cliExecutor));
rpcServer.registerProcessor(new AddLearnersRequestProcessor(cliExecutor));
rpcServer.registerProcessor(new RemoveLearnersRequestProcessor(cliExecutor));
rpcServer.registerProcessor(new ResetLearnersRequestProcessor(cliExecutor));
}
5. 一探究竟：CounterGrpcHelper
CountServer构造器在初步创建RpcServer后，调用了CounterGrpcHelper.initGRpc()和CounterGrpcHelper.setRpcServer(rpcServer)两个方法，接下来和大家分析这两个方法的实现过程
首先请看initGRpc方法：
RpcFactoryHelper.rpcFactory()实际是调用了GrpcRaftRpcFactory（因为GrpcRaftRpcFactory实现了RaftRpcFactory接口），GrpcRaftRpcFactory中维护了一个ConcurrentHashMap<String, Message> parserClasses 其中【key为各种请求/响应实体的名称，value为对应请求/响应的实例】。
然后通过反射获取到MarshallerHelper的registerRespInstance方法，实际上MarshallerHelper里面维护了一个ConcurrentHashMap<String, Message> messages 其中【key为请求实体的名称，value为对应响应的实例】
public static void initGRpc() {
if ("com.alipay.sofa.jraft.rpc.impl.GrpcRaftRpcFactory".equals(RpcFactoryHelper.rpcFactory().getClass()
.getName())) {
RpcFactoryHelper.rpcFactory().registerProtobufSerializer(CounterOutter.GetValueRequest.class.getName(),
CounterOutter.GetValueRequest.getDefaultInstance());
RpcFactoryHelper.rpcFactory().registerProtobufSerializer(
CounterOutter.IncrementAndGetRequest.class.getName(),
CounterOutter.IncrementAndGetRequest.getDefaultInstance());
RpcFactoryHelper.rpcFactory().registerProtobufSerializer(CounterOutter.ValueResponse.class.getName(),
CounterOutter.ValueResponse.getDefaultInstance());
try {
Class<?> clazz = Class.forName("com.alipay.sofa.jraft.rpc.impl.MarshallerHelper");
Method registerRespInstance = clazz.getMethod("registerRespInstance", String.class, Message.class);
registerRespInstance.invoke(null, CounterOutter.GetValueRequest.class.getName(),
CounterOutter.ValueResponse.getDefaultInstance());
registerRespInstance.invoke(null, CounterOutter.IncrementAndGetRequest.class.getName(),
CounterOutter.ValueResponse.getDefaultInstance());
} catch (Exception e) {
LOG.error("Failed to init grpc server", e);
}
}
}
我们再看setRpcServer方法：
CounterGrpcHelper里面还维护了一个RpcServer实例，CounterGrpcHelper.setRpcServer(rpcServer)实际上会将构造的RpcServer装配到CounterGrpcHelper里面。
public static void setRpcServer(RpcServer rpcServer) {
CounterGrpcHelper.rpcServer = rpcServer;
}
6.乘胜追击：RaftGroupService
在CountServer构造器中，经过上述一系列操作步骤，走到了RaftGroupService构造器中，在构造RaftGroupService实体后，调用了它的start方法，这一步在于初始化 raft group 服务框架
public synchronized Node start(final boolean startRpcServer) {
//如果已经启动了，那么就返回
if (this.started) {
return this.node;
}
//校验serverId和groupId
if (this.serverId == null || this.serverId.getEndpoint() == null
|| this.serverId.getEndpoint().equals(new Endpoint(Utils.IP_ANY, 0))) {
throw new IllegalArgumentException("Blank serverId:" + this.serverId);
}
if (StringUtils.isBlank(this.groupId)) {
throw new IllegalArgumentException("Blank group id:" + this.groupId);
}
//设置当前node的endPoint
NodeManager.getInstance().addAddress(this.serverId.getEndpoint());
//创建node
this.node = RaftServiceFactory.createAndInitRaftNode(this.groupId, this.serverId, this.nodeOptions);
if (startRpcServer) {
//启动远程服务
this.rpcServer.init(null);
} else {
LOG.warn("RPC server is not started in RaftGroupService.");
}
this.started = true;
LOG.info("Start the RaftGroupService successfully.");
return this.node;
}
该方法首先对启动参数进行判断，然后对serverId和groupId等参数进行校验。对RaftGroupService在构造器实例化的，然后为NodeManager添加当前EndPoint到(ConcurrentHashSet）addrSet变量中，接着调用RaftServiceFactory#createAndInitRaftNode实例化Node节点。
每个节点都会启动一个rpc的服务，因为每个节点既可以被选举成Leader也可以给其他节点投票，节点之间需要互相通信。
7.刨根问底：Node节点的创建
以下就是Node节点的一系列创建过程，由于嵌套的层数比较多，所以就全部列举出来了，整个过程简而言之就是，createAndInitRaftNode方法首先调用createRaftNode实例化一个Node的实例NodeImpl，然后调用其init方法进行初始化，主要的配置都是在init方法中完成的。代码如下：
this.node = RaftServiceFactory.createAndInitRaftNode(this.groupId, this.serverId, this.nodeOptions);
public static Node createAndInitRaftNode(final String groupId, final PeerId serverId, final NodeOptions opts) {
final Node ret = createRaftNode(groupId, serverId);
if (!ret.init(opts)) {
throw new IllegalStateException("Fail to init node, please see the logs to find the reason.");
}
return ret;
}
public static Node createRaftNode(final String groupId, final PeerId serverId) {
return new NodeImpl(groupId, serverId);
}
public NodeImpl(final String groupId, final PeerId serverId) {
super();
if (groupId != null) {
Utils.verifyGroupId(groupId);
}
this.groupId = groupId;
this.serverId = serverId != null ? serverId.copy() : null;
this.state = State.STATE_UNINITIALIZED;
this.currTerm = 0;
updateLastLeaderTimestamp(Utils.monotonicMs());
this.confCtx = new ConfigurationCtx(this);
this.wakingCandidate = null;
final int num = GLOBAL_NUM_NODES.incrementAndGet();
LOG.info("The number of active nodes increment to {}.", num);
}
8.一窥到底：Node节点的初始化
鉴于NodeImpl#init方法确实挺长的，所以我打算分成几个部分来展示，以便更直观的分析，避免代码太过冗长影响感官
(1)《参数的赋值与校验》
这段代码首先进行一系列判空操作，然后给各变量赋值，然后进行校验判断一下serverId不能为Utils.IP_ANY（即0.0.0.0），当前的Endpoint必须要在NodeManager里面设置过（serverExists方法用来判断是否设置过，RaftGroupService的start方法可以设置该变量）。
//一系列判空操作
Requires.requireNonNull(opts, "Null node options");
Requires.requireNonNull(opts.getRaftOptions(), "Null raft options");
Requires.requireNonNull(opts.getServiceFactory(), "Null jraft service factory");
//JRaftServiceFactory目前有3个实现类
// 1.BDBLogStorageJRaftServiceFactory
// 2.DefaultJRaftServiceFactory
// 3.HybridLogJRaftServiceFactory
this.serviceFactory = opts.getServiceFactory();
this.options = opts;
this.raftOptions = opts.getRaftOptions();
//基于 Metrics 类库的性能指标统计，具有丰富的性能统计指标，默认为false,不开启度量工具
this.metrics = new NodeMetrics(opts.isEnableMetrics());
this.serverId.setPriority(opts.getElectionPriority());
this.electionTimeoutCounter = 0;
//Utils.IP_ANY = "0.0.0.0"
if (this.serverId.getIp().equals(Utils.IP_ANY)) {
LOG.error("Node can't started from IP_ANY.");
return false;
}
if (!NodeManager.getInstance().serverExists(this.serverId.getEndpoint())) {
LOG.error("No RPC server attached to, did you forget to call addService?");
return false;
}
if (this.options.getAppendEntriesExecutors() == null) {
this.options.setAppendEntriesExecutors(Utils.getDefaultAppendEntriesExecutor());
}
接下来初始化一个全局定时调度管理器：
//定时任务管理器
//此处TIMER_FACTORY获取到的是DefaultRaftTimerFactory
//this.options.isSharedTimerPool()默认为false
//this.options.getTimerPoolSize()取值为Utils.cpus() * 3 > 20 ? 20 : Utils.cpus() * 3
this.timerManager = TIMER_FACTORY.getRaftScheduler(this.options.isSharedTimerPool(),
this.options.getTimerPoolSize(), "JRaft-Node-ScheduleThreadPool");
此处浅析一下TimerManager：
初始化一个线程池，根据传入的参数：
this.options.getTimerPoolSize()==Utils.cpus() * 3 > 20 ? 20 : Utils.cpus() * 3
我们分析得知：如果当前的服务器的cpu线程数的3倍大于20，那么这个线程池的coreSize就是20，否则就是cpu线程数的3倍。
public TimerManager(int workerNum, String name) {
this.executor = ThreadPoolUtil.newScheduledBuilder() //
.poolName(name) //
.coreThreads(workerNum) //
.enableMetric(true) //
.threadFactory(new NamedThreadFactory(name, true)) //
.build();
}
(2)《计时器的初始化》
由于这些计时器的实现比较繁杂，所以具体功能等到后面对应章节再一并梳理。
- voteTimer是用来控制选举的，如果选举超时，当前的节点又是候选者角色，那么就会发起选举。
- electionTimer是预投票计时器。候选者在发起投票之前，先发起预投票，如果没有得到半数以上节点的反馈，则候选者就会识趣的放弃参选。
- stepDownTimer定时检查是否需要重新选举leader。当前的leader可能出现它的Follower可能并没有整个集群的1/2却还没有下台的情况，那么这个时候会定期的检查看leader的Follower是否有那么多，没有那么多的话会强制让leader下台。
- snapshotTimer快照计时器。这个计时器会每隔1小时触发一次生成一个快照。
这些计时器有一个共同的特点就是会根据不同的计时器返回一个在一定范围内随机的时间。返回一个随机的时间可以防止多个节点在同一时间内同时发起投票选举从而降低选举失败的概率。
//设置投票计时器
final String suffix = getNodeId().toString();
String name = "JRaft-VoteTimer-" + suffix;
this.voteTimer = new RepeatedTimer(name, this.options.getElectionTimeoutMs(), TIMER_FACTORY.getVoteTimer(
this.options.isSharedVoteTimer(), name)) {
//处理投票超时
@Override
protected void onTrigger() {
handleVoteTimeout();
}
//在一定范围内返回一个随机的时间戳
@Override
protected int adjustTimeout(final int timeoutMs) {
return randomTimeout(timeoutMs);
}
};
//设置预投票计时器
//当leader在规定的一段时间内没有与 Follower 舰船进行通信时，
// Follower 就可以认为leader已经不能正常担任旗舰的职责，则 Follower 可以去尝试接替leader的角色。
// 这段通信超时被称为 Election Timeout
//候选者在发起投票之前，先发起预投票
name = "JRaft-ElectionTimer-" + suffix;
this.electionTimer = new RepeatedTimer(name, this.options.getElectionTimeoutMs(),
TIMER_FACTORY.getElectionTimer(this.options.isSharedElectionTimer(), name)) {
@Override
protected void onTrigger() {
handleElectionTimeout();
}
//在一定范围内返回一个随机的时间戳
//为了避免同时发起选举而导致失败
@Override
protected int adjustTimeout(final int timeoutMs) {
return randomTimeout(timeoutMs);
}
};
//leader下台的计时器
//定时检查是否需要重新选举leader
name = "JRaft-StepDownTimer-" + suffix;
this.stepDownTimer = new RepeatedTimer(name, this.options.getElectionTimeoutMs() >> 1,
TIMER_FACTORY.getStepDownTimer(this.options.isSharedStepDownTimer(), name)) {
@Override
protected void onTrigger() {
handleStepDownTimeout();
}
};
//快照计时器
name = "JRaft-SnapshotTimer-" + suffix;
this.snapshotTimer = new RepeatedTimer(name, this.options.getSnapshotIntervalSecs() * 1000,
TIMER_FACTORY.getSnapshotTimer(this.options.isSharedSnapshotTimer(), name)) {
private volatile boolean firstSchedule = true;
@Override
protected void onTrigger() {
handleSnapshotTimeout();
}
@Override
protected int adjustTimeout(final int timeoutMs) {
if (!this.firstSchedule) {
return timeoutMs;
}
// Randomize the first snapshot trigger timeout
this.firstSchedule = false;
if (timeoutMs > 0) {
int half = timeoutMs / 2;
return half + ThreadLocalRandom.current().nextInt(half);
} else {
return timeoutMs;
}
}
};
(3)《消费队列Disruptor》
关于Disruptor的内容，后面有时间会写一篇相关的文章进行分享
这里初始化了一个Disruptor作为消费队列，然后校验了metrics是否开启，默认是不开启的
this.configManager = new ConfigurationManager();
//初始化一个disruptor，采用多生产者模式
this.applyDisruptor = DisruptorBuilder.<LogEntryAndClosure>newInstance() //
//设置disruptor大小，默认16384
.setRingBufferSize(this.raftOptions.getDisruptorBufferSize()) //
.setEventFactory(new LogEntryAndClosureFactory()) //
.setThreadFactory(new NamedThreadFactory("JRaft-NodeImpl-Disruptor-", true)) //
.setProducerType(ProducerType.MULTI) //
.setWaitStrategy(new BlockingWaitStrategy()) //
.build();
//设置事件处理器
this.applyDisruptor.handleEventsWith(new LogEntryAndClosureHandler());
//设置异常处理器
this.applyDisruptor.setDefaultExceptionHandler(new LogExceptionHandler<Object>(getClass().getSimpleName()));
// 启动disruptor的线程
this.applyQueue = this.applyDisruptor.start();
//如果开启了metrics统计
if (this.metrics.getMetricRegistry() != null) {
this.metrics.getMetricRegistry().register("jraft-node-impl-disruptor",
new DisruptorMetricSet(this.applyQueue));
}
(4)《功能初始化》
对快照、日志、元数据等功能进行初始化
//fsmCaller封装对业务 StateMachine 的状态转换的调用以及日志的写入等
this.fsmCaller = new FSMCallerImpl();
//初始化日志存储功能
if (!initLogStorage()) {
LOG.error("Node {} initLogStorage failed.", getNodeId());
return false;
}
//初始化元数据存储功能
if (!initMetaStorage()) {
LOG.error("Node {} initMetaStorage failed.", getNodeId());
return false;
}
//对FSMCaller初始化
if (!initFSMCaller(new LogId(0, 0))) {
LOG.error("Node {} initFSMCaller failed.", getNodeId());
return false;
}
//实例化投票箱
this.ballotBox = new BallotBox();
final BallotBoxOptions ballotBoxOpts = new BallotBoxOptions();
ballotBoxOpts.setWaiter(this.fsmCaller);
ballotBoxOpts.setClosureQueue(this.closureQueue);
//初始化ballotBox的属性
if (!this.ballotBox.init(ballotBoxOpts)) {
LOG.error("Node {} init ballotBox failed.", getNodeId());
return false;
}
//初始化快照存储功能
if (!initSnapshotStorage()) {
LOG.error("Node {} initSnapshotStorage failed.", getNodeId());
return false;
}
//校验日志文件索引的一致性
final Status st = this.logManager.checkConsistency();
if (!st.isOk()) {
LOG.error("Node {} is initialized with inconsistent log, status={}.", getNodeId(), st);
return false;
}
//配置管理raft group中的信息
this.conf = new ConfigurationEntry();
this.conf.setId(new LogId());
// if have log using conf in log, else using conf in options
if (this.logManager.getLastLogIndex() > 0) {
checkAndSetConfiguration(false);
} else {
this.conf.setConf(this.options.getInitialConf());
// initially set to max(priority of all nodes)
this.targetPriority = getMaxPriorityOfNodes(this.conf.getConf().getPeers());
}
if (!this.conf.isEmpty()) {
Requires.requireTrue(this.conf.isValid(), "Invalid conf: %s", this.conf);
} else {
LOG.info("Init node {} with empty conf.", this.serverId);
}
初始化replicatorGroup、rpcService以及readOnlyService：
// TODO RPC service and ReplicatorGroup is in cycle dependent, refactor it
this.replicatorGroup = new ReplicatorGroupImpl();
//收其他节点或者客户端发过来的请求，转交给对应服务处理
this.rpcService = new DefaultRaftClientService(this.replicatorGroup, this.options.getAppendEntriesExecutors());
final ReplicatorGroupOptions rgOpts = new ReplicatorGroupOptions();
rgOpts.setHeartbeatTimeoutMs(heartbeatTimeout(this.options.getElectionTimeoutMs()));
rgOpts.setElectionTimeoutMs(this.options.getElectionTimeoutMs());
rgOpts.setLogManager(this.logManager);
rgOpts.setBallotBox(this.ballotBox);
rgOpts.setNode(this);
rgOpts.setRaftRpcClientService(this.rpcService);
rgOpts.setSnapshotStorage(this.snapshotExecutor != null ? this.snapshotExecutor.getSnapshotStorage() : null);
rgOpts.setRaftOptions(this.raftOptions);
rgOpts.setTimerManager(this.timerManager);
// Adds metric registry to RPC service.
this.options.setMetricRegistry(this.metrics.getMetricRegistry());
//初始化rpc服务
if (!this.rpcService.init(this.options)) {
LOG.error("Fail to init rpc service.");
return false;
}
this.replicatorGroup.init(new NodeId(this.groupId, this.serverId), rgOpts);
this.readOnlyService = new ReadOnlyServiceImpl();
final ReadOnlyServiceOptions rosOpts = new ReadOnlyServiceOptions();
rosOpts.setFsmCaller(this.fsmCaller);
rosOpts.setNode(this);
rosOpts.setRaftOptions(this.raftOptions);
//只读服务初始化
if (!this.readOnlyService.init(rosOpts)) {
LOG.error("Fail to init readOnlyService.");
return false;
}
(5)《逻辑变动》
这段代码里会将当前的状态设置为Follower，然后启动快照定时器定时生成快照。
如果当前的集群不是单节点集群需要做一下stepDown，表示新生成的Node节点需要重新进行选举。
最下面有一个if分支，如果当前的jraft集群里只有一个节点，那么个节点必定是leader直接进行选举就好了，所以会直接调用electSelf进行选举。
// 将当前的状态设置为Follower
this.state = State.STATE_FOLLOWER;
if (LOG.isInfoEnabled()) {
LOG.info("Node {} init, term={}, lastLogId={}, conf={}, oldConf={}.", getNodeId(), this.currTerm,
this.logManager.getLastLogId(false), this.conf.getConf(), this.conf.getOldConf());
}
//如果快照执行器不为空，并且生成快照的时间间隔大于0，那么就定时生成快照
if (this.snapshotExecutor != null && this.options.getSnapshotIntervalSecs() > 0) {
LOG.debug("Node {} start snapshot timer, term={}.", getNodeId(), this.currTerm);
this.snapshotTimer.start();
}
//新启动的node需要重新选举
if (!this.conf.isEmpty()) {
stepDown(this.currTerm, false, new Status());
}
if (!NodeManager.getInstance().add(this)) {
LOG.error("NodeManager add {} failed.", getNodeId());
return false;
}
// Now the raft node is started , have to acquire the writeLock to avoid race
// conditions
this.writeLock.lock();
//这个分支表示当前的jraft集群里只有一个节点，那么个节点必定是leader直接进行选举就好了
if (this.conf.isStable() && this.conf.getConf().size() == 1 && this.conf.getConf().contains(this.serverId)) {
// The group contains only this server which must be the LEADER, trigger
// the timer immediately.
electSelf();
} else {
this.writeLock.unlock();
}
return true;
9.写在最后
To be Honest，第一次阅读这种复杂的开源代码，确实非常吃力，但其实仔细去琢磨后你会慢慢痴迷于里面很多优雅、优秀的实现。
在这里，感谢SOFAJRaft的每一位代码贡献者。源码的阅读过程中，的的确确学到了很多东西。我也会继续学习下去，希望能够巩固、深入我对RAFT一致性算法的理解。
类似资料：