世界热门:SOFAJRaft模块启动过程

本篇文章旨在分析SOFAJRaft中jraft-example模块的启动过程，由于SOFAJRaft在持续开源的过程中，所以无法保证示例代码永远是最新的，要是有较大的变动或者纰漏、错误的地方，欢迎大家留言讨论。@Author：Akai-yuan更新时间：2023年1月20日

SOFAJRaft-Start Our Journey

1. 开门见山：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];    //SOFAJRaft集群的名字    final String groupId = args[1];    //当前节点的ip和端口    final String serverIdStr = args[2];    //集群节点的ip和端口    final String initConfStr = args[3];    final NodeOptions nodeOptions = new NodeOptions();    // for test, modify some params    // 设置选举超时时间为 1 秒    nodeOptions.setElectionTimeoutMs(1000);    // 关闭 CLI 服务    nodeOptions.setDisableCli(false);    // 每隔30秒做一次 snapshot    nodeOptions.setSnapshotIntervalSecs(30);    // 解析参数    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 need block to prevent process exit    CounterGrpcHelper.blockUntilShutdown();}

我们在启动CounterServer的main方法的时候，会将传入的String[]类型参数args分别转化为日志存储的路径、SOFAJRaft集群的名字、当前节点的ip和端口、集群节点的ip和端口，并设值到NodeOptions中，作为当前节点启动的参数。

(资料图)

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. 渐入佳境：构造CountServer

引子：在main方法中，我们可以看到，进行初步的参数解析后，调用了CountServer的构造器，要说这个构造器，第一次看里面的步骤确实会感觉挺复杂的，接下来我们一起分析一下源码。

CountServer构造器的源码如下：

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();    }

接下来仔细说说CountServer的构造器里面具体做了什么。

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 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 parserClasses 其中【key为各种请求/响应实体的名称，value为对应请求/响应的实例】。然后通过反射获取到MarshallerHelper的registerRespInstance方法，实际上MarshallerHelper里面维护了一个ConcurrentHashMap 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的ip和端口        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;    }

这个方法会在一开始的时候对RaftGroupService在构造器实例化的参数进行校验，然后把当前节点的Endpoint添加到NodeManager的addrSet变量中，接着调用RaftServiceFactory#createAndInitRaftNode实例化Node节点。每个节点都会启动一个rpc的服务，因为每个节点既可以被选举也可以投票给其他节点，节点之间需要互相通信，所以需要启动一个rpc服务。

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不能为0.0.0.0，当前的Endpoint必须要在NodeManager里面设置过等等（NodeManager的设置是在RaftGroupService的start方法里）。然后会初始化一个全局的的定时调度管理器TimerManager：

//一系列判空操作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.HybridLogJRaftServiceFactorythis.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.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