Elastic-Job原理--任务失败转移(五)

拼搏现实的明天。 2022-05-06 06:42 454阅读 0赞

在上一篇博客[Elastic-Job原理--任务调度处理（四）][Elastic-Job_--]我们已经了解到Elastic-Job依赖quartz定时任务执行分片任务的过程，这篇博客我们简单了解一下Elastic-Job中当某个服务器节点与注册中心断开连接(无法进行任务执行)时其需要执行的任务转移到其他节点的过程。

首先提供如下类图，与节点任务失败转移相关主要类如下：

![70][]

FailoverService，作业失效转移服务。  
FailoverNode，作业失效转移数据存储路径。  
FailoverListenerManager，作业失效转移监听管理器。

### 一、重新分片 ###

当服务器节点从注册中心zk断开连接时，Elastic-job需要做的一件事情是需要在下次任务执行前进行重新分片，当zk节点数目发生变更时，会引发ListenServersChangedJobListener监听器调用，此监听器会调用shardingService的重新分片标志设置方法，这样再下次任务执行前会重新进行任务分片操作。

/**
     * 当实例节点变更时会调用此监听器
     *
     */
    class ListenServersChangedJobListener extends AbstractJobListener {
            
            @Override
            protected void dataChanged(final String path, final Type eventType, final String data) {
                //如果节点数目发生变更则设置重新分片标志，下次任务执行前会进行重新分片
                if (!JobRegistry.getInstance().isShutdown(jobName) && (isInstanceChange(eventType, path) || isServerChange(path))) {
                    shardingService.setReshardingFlag();
                }
            }
            
            private boolean isInstanceChange(final Type eventType, final String path) {
                return instanceNode.isInstancePath(path) && Type.NODE_UPDATED != eventType;
            }
            
            private boolean isServerChange(final String path) {
                return serverNode.isServerPath(path);
            }
        }

任务重新分片只是解决了下次任务执行时，所有的分片任务都是分布到各个实例中，但是当前失效的任务是如何处理的。

### 二、任务失效转移 ###

所谓失效转移，就是在执行任务的过程中遇见异常的情况，这个分片任务可以在其他节点再次执行。这个和上面的HA不同，对于HA，上面如果任务终止，那么不会在其他任务实例上再次重新执行。Job的失效转移监听来源于FailoverListenerManager中JobCrashedJobListener的dataChanged方法。FailoverListenerManager监听的是zk的instance节点删除事件。如果任务配置了failover等于true，其中某个instance与zk失去联系或被删除，并且失效的节点又不是本身，就会触发失效转移逻辑。首先，在某个任务实例elastic-job会在leader节点下面创建failover节点以及items节点。items节点下会有失效任务实例的原本应该做的分片好。比如，失效的任务实例原来负责分片1和2。那么items节点下就会有名字叫1的子节点，就代表分片1需要转移到其他节点上去运行。如下图：

![70 1][]

当节点任务失效时会调用JobCrashedJobListener监听器，此监听器会根据实例id获取所有的分片，然后调用FailoverService的setCrashedFailoverFlag方法，将每个分片id写到/jobName/leader/failover/items下

/**
     * 任务失效时会调用这个监听器
     */
    class JobCrashedJobListener extends AbstractJobListener {
            
            @Override
            protected void dataChanged(final String path, final Type eventType, final String data) {
                if (isFailoverEnabled() && Type.NODE_REMOVED == eventType && instanceNode.isInstancePath(path)) {
                    String jobInstanceId = path.substring(instanceNode.getInstanceFullPath().length() + 1);
                    if (jobInstanceId.equals(JobRegistry.getInstance().getJobInstance(jobName).getJobInstanceId())) {
                        return;
                    }
                    //会将所有的分片初始化到注册中心中
                    List<Integer> failoverItems = failoverService.getFailoverItems(jobInstanceId);
                    if (!failoverItems.isEmpty()) {
                        for (int each : failoverItems) {
                            failoverService.setCrashedFailoverFlag(each);
                            failoverService.failoverIfNecessary();
                        }
                    } else {
                        for (int each : shardingService.getShardingItems(jobInstanceId)) {
                            failoverService.setCrashedFailoverFlag(each);
                            failoverService.failoverIfNecessary();
                        }
                    }
                }
            }
        }

在FailoverService方法中调用setCrashedFailoverFlag方法将需要任务转移的分片id进行实例化。

/**
         * 设置失效的分片项标记.
         * 
         * @param item 崩溃的作业项
         */
        public void setCrashedFailoverFlag(final int item) {
            if (!isFailoverAssigned(item)) {
                jobNodeStorage.createJobNodeIfNeeded(FailoverNode.getItemsNode(item));
            }
        }

然后接下来调用FailoverService的failoverIfNessary方法，首先判断是否需要失败转移，如果可以需要则只需作业失败转移。

/**
         * 如果需要失效转移, 则执行作业失效转移.
         */
        public void failoverIfNecessary() {
            if (needFailover()) {
                jobNodeStorage.executeInLeader(FailoverNode.LATCH, new FailoverLeaderExecutionCallback());
            }
        }

在needFailover方法会对是否需要失效转移进行判断

private boolean needFailover() {
             // `${JOB_NAME}/leader/failover/items/${ITEM_ID}` 有失效转移的作业分片项
            return jobNodeStorage.isJobNodeExisted(FailoverNode.ITEMS_ROOT) && !jobNodeStorage.getJobNodeChildrenKeys(FailoverNode.ITEMS_ROOT).isEmpty()
                    // 当前作业不在运行中
                    && !JobRegistry.getInstance().isJobRunning(jobName);
        }

条件一：$\{JOB\_NAME\}/leader/failover/items/$\{ITEM\_ID\} 有失效转移的作业分片项。  
条件二：当前作业不在运行中。此条件即是上文提交的作业节点空闲的定义。失效转移： 运行中的作业服务器崩溃不会导致重新分片，只会在下次作业启动时分片。启用失效转移功能可以在本次作业执行过程中，监测其他作业服务器【空闲】，抓取未完成的孤儿分片项执行

在FailoverLeaderExecutionCallback中回调逻辑如下：

（1）也会首先判断是否需要失效转移，

（2）从注册中心获得一个 \`$\{JOB\_NAME\}/leader/failover/items/$\{ITEM\_ID\}\` 作业分片项，

（3）在注册中心节点\`$\{JOB\_NAME\}/sharding/$\{ITEM\_ID\}/failover\` 作业分片项 为 当前作业节点，

（4）然后移除任务转移分片项，

（5）最后调用执行，提交任务

class FailoverLeaderExecutionCallback implements LeaderExecutionCallback {
       
       @Override
       public void execute() {
           // 判断需要失效转移
           if (JobRegistry.getInstance().isShutdown(jobName) || !needFailover()) {
               return;
           }
           // 获得一个 `${JOB_NAME}/leader/failover/items/${ITEM_ID}` 作业分片项
           int crashedItem = Integer.parseInt(jobNodeStorage.getJobNodeChildrenKeys(FailoverNode.ITEMS_ROOT).get(0));
           log.debug("Failover job '{}' begin, crashed item '{}'", jobName, crashedItem);
           // 设置这个 `${JOB_NAME}/sharding/${ITEM_ID}/failover` 作业分片项 为 当前作业节点
           jobNodeStorage.fillEphemeralJobNode(FailoverNode.getExecutionFailoverNode(crashedItem), JobRegistry.getInstance().getJobInstance(jobName).getJobInstanceId());
           // 移除这个 `${JOB_NAME}/leader/failover/items/${ITEM_ID}` 作业分片项
           jobNodeStorage.removeJobNodeIfExisted(FailoverNode.getItemsNode(crashedItem));
           // TODO 不应使用triggerJob, 而是使用executor统一调度 疑问：为什么要用executor统一，后面研究下
           // 触发作业执行
           JobScheduleController jobScheduleController = JobRegistry.getInstance().getJobScheduleController(jobName);
           if (null != jobScheduleController) {
               jobScheduleController.triggerJob();
           }
       }
    }

调用 JobScheduleController\#triggerJob() 方法，立即启动作业。调用该方法，实际作业不会立即执行，而仅仅是进行触发。如果有多个失效转移的作业分片项，多次调用 JobScheduleController\#triggerJob() 方法会不会导致作业是并行执行的？答案是不会，因为一个作业的 Quartz 线程数设置为 1。

同时可以结合博客[Elastic-Job原理--任务调度处理（四）][Elastic-Job_--]任务调度处理流程中，LiteJobFacade有获取分片操作的函数。

@Override
        public ShardingContexts getShardingContexts() {
            //是否失败转移
            boolean isFailover = configService.load(true).isFailover();
            if (isFailover) {
                //获取失败分片
                List<Integer> failoverShardingItems = failoverService.getLocalFailoverItems();
                if (!failoverShardingItems.isEmpty()) {
                    //执行失败分片任务
                    return executionContextService.getJobShardingContext(failoverShardingItems);
                }
            }
            //重新分片
            shardingService.shardingIfNecessary();
            List<Integer> shardingItems = shardingService.getLocalShardingItems();
            //移除已经失败转移执行的任务
            if (isFailover) {
                shardingItems.removeAll(failoverService.getLocalTakeOffItems());
            }
            shardingItems.removeAll(executionService.getDisabledItems(shardingItems));
            return executionContextService.getJobShardingContext(shardingItems);
        }

在getShardingContexts中有专门获取所有的要失败转移需要执行的分片。

[Elastic-Job_--]: https://blog.csdn.net/qq924862077/article/details/82858684
[70]: /images/20220506/fe7f6bf2ea7b4466a564dde660060307.png
[70 1]: /images/20220506/ccb2ee2fff414f6d98106ca14c5399c8.png