Elastic job 移除leader 源码分析

Elastic job 移除选举方法 leaderService.removeLeader(); 判断是否存在leader ，移除 zk /leader/election/instance node

    /**
     * 删除主节点供重新选举.
     */
    public void removeLeader() {
        jobNodeStorage.removeJobNodeIfExisted(LeaderNode.INSTANCE);
    }

leaderService.removeLeader() 被以下方法调用

image.png

LeaderAbdicationJobListener dataChanged 方法调用了移除选举方法

 class LeaderAbdicationJobListener extends AbstractJobListener {
        
        @Override
        protected void dataChanged(final String path, final Type eventType, final String data) {
            if (leaderService.isLeader() && isLocalServerDisabled(path, data)) {
                leaderService.removeLeader();
            }
        }
        
        private boolean isLocalServerDisabled(final String path, final String data) {
            return serverNode.isLocalServerPath(path) && ServerStatus.DISABLED.name().equals(data);
        }
    }

简单说在当前实例是leader 情况下，且状态被 disable 的情况下，当前leader 会remove leader ，然后会激活被动选举。（如果在job执行过程 disable leader 节点，那么isLeaderUntilBlock 也会执行吧，等待leader选举后再执行job ）

isLocalServerDisabled方法判断 event path 是否是当前servers 路径本地节点，且 event type 是否 disable 都满足返回true

JobShutdownHookPlugin shutdown() 方法调用了移除选举方法

public final class JobShutdownHookPlugin extends ShutdownHookPlugin {
    
    private String jobName;
    
    @Override
    public void initialize(final String name, final Scheduler scheduler, final ClassLoadHelper classLoadHelper) throws SchedulerException {
        super.initialize(name, scheduler, classLoadHelper);
        jobName = scheduler.getSchedulerName();
    }
    
    @Override
    public void shutdown() {
        CoordinatorRegistryCenter regCenter = JobRegistry.getInstance().getRegCenter(jobName);
        if (null == regCenter) {
            return;
        }
        LeaderService leaderService = new LeaderService(regCenter, jobName);
        if (leaderService.isLeader()) {
            leaderService.removeLeader();
        }
        new InstanceService(regCenter, jobName).removeInstance();
    }
}

JobShutdownHookPlugin 会在 ctrl c ， kill 或者 JVM terminating （idea 手动那个不行的）时候激活该job 的 shutdown 方法
首先判断注册中心是否有该jobname ，如果有判断当前是否是leader ，如果是 remove leader ，最后会remove 该job 的 instances 节点如 instances/192.168.200.151@-@7100 节点

ShutdownListenerManager.InstanceShutdownStatusJobListener 在 job 没有shutdown ，没有 paused ，在 instances下的当前节点被移除，判断当前节点没有运行的情况下会执行 schedulerFacade.shutdownInstance(); 然后会判断当前节点是否leader ，如果是，执行 leaderService.removeLeader() 方法。

class InstanceShutdownStatusJobListener extends AbstractJobListener {
        
        @Override
        protected void dataChanged(final String path, final Type eventType, final String data) {
            if (!JobRegistry.getInstance().isShutdown(jobName) && !JobRegistry.getInstance().getJobScheduleController(jobName).isPaused()
                    && isRemoveInstance(path, eventType) && !isReconnectedRegistryCenter()) {
                schedulerFacade.shutdownInstance();
            }
        }
        
        private boolean isRemoveInstance(final String path, final Type eventType) {
            return instanceNode.isLocalInstancePath(path) && Type.NODE_REMOVED == eventType;
        }
        
        private boolean isReconnectedRegistryCenter() {
            return instanceService.isLocalJobInstanceExisted();
        }
    }

    /**
     * 终止作业调度.
     */
    public void shutdownInstance() {
        if (leaderService.isLeader()) {
            leaderService.removeLeader();
        }
        monitorService.close();
        if (reconcileService.isRunning()) {
            reconcileService.stopAsync();
        }
        JobRegistry.getInstance().shutdown(jobName);
    }

Elastic job 移除leader 源码分析

推荐阅读更多精彩内容