Elastic-Job原理--任务调度处理（四）

野性酷女 2022-05-10 07:16 262阅读 0赞

在上一篇博客[Elastic-Job原理--任务分片策略（三）][Elastic-Job_--]我们已经了解了Elastic-Job的任务分片策略，这篇博客我们了解学习一下Elastic-Job是如何执行分片任务的。

首先，Elastic-Job的定时任务执行机制还是基于quartz开发的，因此Elastic-Job实现了Quartz的任务接口Job实现了LiteJob，来根据定时任务规则执行定时任务。

实现Quartz的接口Job

public final class LiteJob implements Job {
        
        @Setter
        private ElasticJob elasticJob;
        
        @Setter
        private JobFacade jobFacade;
        
        @Override
        public void execute(final JobExecutionContext context) throws JobExecutionException {
            JobExecutorFactory.getJobExecutor(elasticJob, jobFacade).execute();
        }
    }

根据任务类型JobExecutorFactory会获取任务执行器，然后调用execute方法

/**
         * 获取作业执行器.
         *
         * @param elasticJob 分布式弹性作业
         * @param jobFacade 作业内部服务门面服务
         * @return 作业执行器
         */
        @SuppressWarnings("unchecked")
        public static AbstractElasticJobExecutor getJobExecutor(final ElasticJob elasticJob, final JobFacade jobFacade) {
            if (null == elasticJob) {
                return new ScriptJobExecutor(jobFacade);
            }
            if (elasticJob instanceof SimpleJob) {
                return new SimpleJobExecutor((SimpleJob) elasticJob, jobFacade);
            }
            if (elasticJob instanceof DataflowJob) {
                return new DataflowJobExecutor((DataflowJob) elasticJob, jobFacade);
            }
            throw new JobConfigurationException("Cannot support job type '%s'", elasticJob.getClass().getCanonicalName());
        }

Elastic-Job提供了任务执行器抽象类AbstractElasticJobExecutor，在AbstractElasticJobExecutor中会获取任务分片信息及任务失败转移等处理操作。

在execute中会获取所有的分片信息，及一系列的处理操作。

/**
         * 执行作业.
         */
        public final void execute() {
            try {
                jobFacade.checkJobExecutionEnvironment();
            } catch (final JobExecutionEnvironmentException cause) {
                jobExceptionHandler.handleException(jobName, cause);
            }
    		//获取分片信息
            ShardingContexts shardingContexts = jobFacade.getShardingContexts();
            if (shardingContexts.isAllowSendJobEvent()) {
                jobFacade.postJobStatusTraceEvent(shardingContexts.getTaskId(), State.TASK_STAGING, String.format("Job '%s' execute begin.", jobName));
            }
            if (jobFacade.misfireIfRunning(shardingContexts.getShardingItemParameters().keySet())) {
                if (shardingContexts.isAllowSendJobEvent()) {
                    jobFacade.postJobStatusTraceEvent(shardingContexts.getTaskId(), State.TASK_FINISHED, String.format(
                            "Previous job '%s' - shardingItems '%s' is still running, misfired job will start after previous job completed.", jobName, 
                            shardingContexts.getShardingItemParameters().keySet()));
                }
                return;
            }
            try {
                jobFacade.beforeJobExecuted(shardingContexts);
                //CHECKSTYLE:OFF
            } catch (final Throwable cause) {
                //CHECKSTYLE:ON
                jobExceptionHandler.handleException(jobName, cause);
            }
    		//执行分片任务
            execute(shardingContexts, JobExecutionEvent.ExecutionSource.NORMAL_TRIGGER);
            while (jobFacade.isExecuteMisfired(shardingContexts.getShardingItemParameters().keySet())) {
                jobFacade.clearMisfire(shardingContexts.getShardingItemParameters().keySet());
                execute(shardingContexts, JobExecutionEvent.ExecutionSource.MISFIRE);
            }
            jobFacade.failoverIfNecessary();
            try {
                jobFacade.afterJobExecuted(shardingContexts);
                //CHECKSTYLE:OFF
            } catch (final Throwable cause) {
                //CHECKSTYLE:ON
                jobExceptionHandler.handleException(jobName, cause);
            }
        }

在execute中会记录一些任务的状态信息，然后执行process方法

private void execute(final ShardingContexts shardingContexts, final JobExecutionEvent.ExecutionSource executionSource) {
            if (shardingContexts.getShardingItemParameters().isEmpty()) {
                if (shardingContexts.isAllowSendJobEvent()) {
                    jobFacade.postJobStatusTraceEvent(shardingContexts.getTaskId(), State.TASK_FINISHED, String.format("Sharding item for job '%s' is empty.", jobName));
                }
                return;
            }
            jobFacade.registerJobBegin(shardingContexts);
            String taskId = shardingContexts.getTaskId();
            if (shardingContexts.isAllowSendJobEvent()) {
                jobFacade.postJobStatusTraceEvent(taskId, State.TASK_RUNNING, "");
            }
            try {
    			//执行分片任务
                process(shardingContexts, executionSource);
            } finally {
                // TODO 考虑增加作业失败的状态，并且考虑如何处理作业失败的整体回路
                jobFacade.registerJobCompleted(shardingContexts);
                if (itemErrorMessages.isEmpty()) {
                    if (shardingContexts.isAllowSendJobEvent()) {
                        jobFacade.postJobStatusTraceEvent(taskId, State.TASK_FINISHED, "");
                    }
                } else {
                    if (shardingContexts.isAllowSendJobEvent()) {
                        jobFacade.postJobStatusTraceEvent(taskId, State.TASK_ERROR, itemErrorMessages.toString());
                    }
                }
            }
        }

在process方法中，会根据分片数量单任务时直接执行，多任务时添加到线程池执行

//根据分片数量依次执行分片任务
    private void process(final ShardingContexts shardingContexts, final JobExecutionEvent.ExecutionSource executionSource) {
            Collection<Integer> items = shardingContexts.getShardingItemParameters().keySet();
    		//如果只存在一个分片则直接执行
            if (1 == items.size()) {
                int item = shardingContexts.getShardingItemParameters().keySet().iterator().next();
                JobExecutionEvent jobExecutionEvent =  new JobExecutionEvent(shardingContexts.getTaskId(), jobName, executionSource, item);
                process(shardingContexts, item, jobExecutionEvent);
                return;
            }
            final CountDownLatch latch = new CountDownLatch(items.size());
    		//根据分片数量依次执行分片任务
            for (final int each : items) {
                final JobExecutionEvent jobExecutionEvent = new JobExecutionEvent(shardingContexts.getTaskId(), jobName, executionSource, each);
                if (executorService.isShutdown()) {
                    return;
                }
                executorService.submit(new Runnable() {
                    
                    @Override
                    public void run() {
                        try {
                            process(shardingContexts, each, jobExecutionEvent);
                        } finally {
                            latch.countDown();
                        }
                    }
                });
            }
            try {
    			//等待所有分片任务执行完毕
                latch.await();
            } catch (final InterruptedException ex) {
                Thread.currentThread().interrupt();
            }
        }

调用子类的process方法根据任务类型去执行

private void process(final ShardingContexts shardingContexts, final int item, final JobExecutionEvent startEvent) {
            if (shardingContexts.isAllowSendJobEvent()) {
                jobFacade.postJobExecutionEvent(startEvent);
            }
            log.trace("Job '{}' executing, item is: '{}'.", jobName, item);
            JobExecutionEvent completeEvent;
            try {
    			//执行分片任务
                process(new ShardingContext(shardingContexts, item));
                completeEvent = startEvent.executionSuccess();
                log.trace("Job '{}' executed, item is: '{}'.", jobName, item);
                if (shardingContexts.isAllowSendJobEvent()) {
                    jobFacade.postJobExecutionEvent(completeEvent);
                }
                // CHECKSTYLE:OFF
            } catch (final Throwable cause) {
                // CHECKSTYLE:ON
                completeEvent = startEvent.executionFailure(cause);
                jobFacade.postJobExecutionEvent(completeEvent);
                itemErrorMessages.put(item, ExceptionUtil.transform(cause));
                jobExceptionHandler.handleException(jobName, cause);
            }
        }

最终根据任务类型，调用AbstractElasticJobExecutor的实现类

![70][]

/**
     * 简单作业执行器.
     * 
     * @author zhangliang
     */
    public final class SimpleJobExecutor extends AbstractElasticJobExecutor {
        
        private final SimpleJob simpleJob;
        
        public SimpleJobExecutor(final SimpleJob simpleJob, final JobFacade jobFacade) {
            super(jobFacade);
            this.simpleJob = simpleJob;
        }
        
        @Override
        protected void process(final ShardingContext shardingContext) {
            simpleJob.execute(shardingContext);
        }
    }

[Elastic-Job_--]: https://blog.csdn.net/qq924862077/article/details/82994083
[70]: /images/20220510/1f993af7d2c24b64865bfe8b520e0563.png