SOFAJRaft源码阅读-模块启动过程

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

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

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")

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

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

  //添加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));
    }

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

public static void setRpcServer(RpcServer rpcServer) {
        CounterGrpcHelper.rpcServer = rpcServer;
}

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

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

    	//一系列判空操作
		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");

public TimerManager(int workerNum, String name) {
        this.executor = ThreadPoolUtil.newScheduledBuilder() //
            .poolName(name) //
            .coreThreads(workerNum) //
            .enableMetric(true) //
            .threadFactory(new NamedThreadFactory(name, true)) //
            .build();
    }

    	//设置投票计时器
        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;
                }
            }
        };

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

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

// 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;
        }

    	// 将当前的状态设置为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;