Elastic job leader选举 源码分析

问题3 当某一实例 挂掉时候 ，集群的leader 会报异常 ，当任务完成的时候 ， 只会重新分片 ， 如果 任务没有完成 类似于 问题2 ， leader 报异常后 会报超时异常 。

完成超时异常 ，同问题2 。这里我们讲一讲 leader 什么时候 会重新选举

leader 相关的服务主要在 LeaderService 中 ， 我们可以看到 ，有3个地方调用了 electLeader 方法进行选举。

image.png

具体分析

SchedulerFacade 的registerStartUpInfo(boolean) 方法中调用了选举方法 leaderService.electLeader();

    /**
     * 注册作业启动信息.
     * 
     * @param enabled 作业是否启用
     */
    public void registerStartUpInfo(final boolean enabled) {
        listenerManager.startAllListeners();
        leaderService.electLeader();
        serverService.persistOnline(enabled);
        instanceService.persistOnline();
        shardingService.setReshardingFlag();
        monitorService.listen();
        if (!reconcileService.isRunning()) {
            reconcileService.startAsync();
        }
    }

实例启动的时候 会进行一次选举 , 会创建创建一临时节点 位置 leader/election/instance 。

LeaderService isLeaderUntilBlock()调用了 选举方法leaderService.electLeader();

 /**
     * 判断当前节点是否是主节点.
     * 
     * <p>
     * 如果主节点正在选举中而导致取不到主节点, 则阻塞至主节点选举完成再返回.
     * </p>
     * 
     * @return 当前节点是否是主节点
     */
    public boolean isLeaderUntilBlock() {
        while (!hasLeader() && serverService.hasAvailableServers()) {
            log.info("Leader is electing, waiting for {} ms", 100);
            BlockUtils.waitingShortTime();
            if (!JobRegistry.getInstance().isShutdown(jobName) && serverService.isAvailableServer(JobRegistry.getInstance().getJobInstance(jobName).getIp())) {
                electLeader();
            }
        }
        return isLeader();
    }

当有leader 的时候 ， 集群没有合适实例 isLeaderUntilBlock 中的选举方法 是不会执行的 。
在满足以下条件下, 会执行electLeader
1 没有 leader !hasLeader()
2 有合适的服务器 serverService.hasAvailableServers()
3 当前job 没有 shutdown !JobRegistry.getInstance().isShutdown(jobName)
4 当前服务器是合适的 serverService.isAvailableServer

isLeaderUntilBlock 会在 job 执行 ，获取分片上下文的时候 执行 ,如果当前集群 ，分片未结束，blockUntilShardingCompleted方法 会阻塞 ，等待分片结束

   /**
     * 如果需要分片且当前节点为主节点, 则作业分片.
     * 
     * <p>
     * 如果当前无可用节点则不分片.
     * </p>
     */
    public void shardingIfNecessary() {
        List<JobInstance> availableJobInstances = instanceService.getAvailableJobInstances();
        if (!isNeedSharding() || availableJobInstances.isEmpty()) {
            return;
        }
        if (!leaderService.isLeaderUntilBlock()) {
            blockUntilShardingCompleted();
            return;
        }

ElectionListenerManager.LeaderElectionJobListener 中的 dataChanged方法中调用了 选举方法

    class LeaderElectionJobListener extends AbstractJobListener {
        
        @Override
        protected void dataChanged(final String path, final Type eventType, final String data) {
            if (!JobRegistry.getInstance().isShutdown(jobName) && (isActiveElection(path, data) || isPassiveElection(path, eventType))) {
                leaderService.electLeader();
            }
        }

该listener 会监听 zk ， jobname 级别tree node 变化 ， 如果当前job 没有shutdown ， 满足主动选举 或者被动选举 任意条件 会 进行选举 。

主动选举代码如下 ， 当前集群没有leader ， 当前server没有disable 情况下
会返回true
比如 当前job 没有shutdown ，没有leader ， 本地路径 /javaSimpleJob/servers/192.168.200.151 与path 一致 ， 节点 data 不等于disable 可以参加选举 ，和事件无关哈

         private boolean isActiveElection(final String path, final String data) {
            return !leaderService.hasLeader() && isLocalServerEnabled(path, data);
        }

被动选举代码如下 如果leader crash ， 如果当前实例可用 ，isPassiveElection 会返回 true 可以参加选举

        private boolean isPassiveElection(final String path, final Type eventType) {
            return isLeaderCrashed(path, eventType) && serverService.isAvailableServer(JobRegistry.getInstance().getJobInstance(jobName).getIp());
        }

isLeaderCrashed 判断当前event path 是否是 /leader/election/instance下的节点 ， eventType 是否NODE_REMOVED 如果都满足 ，return true
...
private boolean isLeaderCrashed(final String path, final Type eventType) {
return leaderNode.isLeaderInstancePath(path) && Type.NODE_REMOVED == eventType;
}
...