[源码解析]之-ElasticJob如何做到单分片任务只在一个节点运行的?

约定不等于承诺〃 2023-01-13 14:59 470阅读 0赞

*这篇文章是之前体验elastic-job时搭了一个elasticjob多节点环境，当时对elasticjob是如何做到多节点中只运行一个定时任务挺疑惑的，所以就跟着源码看了看并做了一个简单的记录。*

*在了解与体验了spring cloud data flow后觉得elasticjob优秀是优秀，但与spring cloud data flow相比能更好地运行在kuernetes上总觉得不怎么适用于当时公司的定时任务场景，于是这篇文章也就没有发出来。*

*但最近由于公司项目由于数据存储那块原有架构感觉快扛不住了，于是了解与学习了些关于ClickHouse的相关知识，在了解到clickhouse中有的MPP特性时了解到它也是通过zookeeper实现的。这让我想到了ElasticJob无中心节点的特性，也让我觉得原来**不管是什么类型的应用，在实现原理方面总是有着如此多相似的地方**，同时我觉得当初去看elasticjob的源码也还是很有价值和意义的，尽管最后没有使用它。*

--------------------

## 单分片任务执行过程 ##

在使用elasticjob的过程中有一点感觉很神奇：**定义了一个不需要分片的定时任务，然后部署了多个jar包(也就是多个节点)，但在运行过程中看到只有一个节点会运行那个定时任务**

怀着对上面这个问题的疑惑，我决定来跟踪一下elastic-job的源码，看一下它底层是如何实现**单分片任务只在一个节点运行**的原理

### elasticjob-lite-spring-boot-starter中job初始化过程 ###

测试的环境仍然是上一篇文章[《定时任务框架elastic-job快速体验与思考》][elastic-job]中的环境，项目是用的spring-boot框架来搭建的，通过引入elasticjob-lite-spring-boot-starter依赖来进行实现的，所以探究elastic-job的源码就从这个依赖开始吧！

对于spring-boot-starter项目的入口直接到项目中的**spring.factories**配置文件就行了，elasticjob-lite-spring-boot-starter项目的中spring.factories配置文件内容如下:

org.springframework.boot.autoconfigure.EnableAutoConfiguration=\
      org.apache.shardingsphere.elasticjob.lite.spring.boot.job.ElasticJobLiteAutoConfiguration

随后进入ElasticJobLiteAutoConfiguration查看即可：

/**
     * ElasticJob-Lite auto configuration.
     */
    @Configuration(proxyBeanMethods = false)
    @AutoConfigureAfter(DataSourceAutoConfiguration.class)
    @ConditionalOnProperty(name = "elasticjob.enabled", havingValue = "true", matchIfMissing = true)
    @Import({ElasticJobRegistryCenterConfiguration.class, ElasticJobTracingConfiguration.class, ElasticJobSnapshotServiceConfiguration.class})
    @EnableConfigurationProperties(ElasticJobProperties.class)
    public class ElasticJobLiteAutoConfiguration {
        
        @Configuration(proxyBeanMethods = false)
        @Import({ElasticJobBootstrapConfiguration.class, ScheduleJobBootstrapStartupRunner.class})
        protected static class ElasticJobConfiguration {
        }
    }

这个AutoConfiguration的关联类很多，找准主线很重要，先查看定时任务在哪里定义的

根据类的名称直接从ElasticJobBootstrapConfiguration.class入手

## ElasticJobBootstrapConfiguration ##

ElasticJobBootstrapConfiguration中有一个PostConstrct容器初始化方法，就直接看这个方法即可：

/**
         * Create job bootstrap instances and register them into container.
         */
        @PostConstruct
        public void createJobBootstrapBeans() {
            ElasticJobProperties elasticJobProperties = applicationContext.getBean(ElasticJobProperties.class);
            SingletonBeanRegistry singletonBeanRegistry = ((ConfigurableApplicationContext) applicationContext).getBeanFactory();
            CoordinatorRegistryCenter registryCenter = applicationContext.getBean(CoordinatorRegistryCenter.class);
            TracingConfiguration<?> tracingConfig = getTracingConfiguration();
            constructJobBootstraps(elasticJobProperties, singletonBeanRegistry, registryCenter, tracingConfig);
        }

找到constructJobBootstraps方法，看一下这里面是如何定义job的

private void constructJobBootstraps(final ElasticJobProperties elasticJobProperties, final SingletonBeanRegistry singletonBeanRegistry,
                                            final CoordinatorRegistryCenter registryCenter, final TracingConfiguration<?> tracingConfig) {
            for (Map.Entry<String, ElasticJobConfigurationProperties> entry : elasticJobProperties.getJobs().entrySet()) {
                ElasticJobConfigurationProperties jobConfigurationProperties = entry.getValue();
                Preconditions.checkArgument(null != jobConfigurationProperties.getElasticJobClass()
                                || !Strings.isNullOrEmpty(jobConfigurationProperties.getElasticJobType()),
                        "Please specific [elasticJobClass] or [elasticJobType] under job configuration.");
                Preconditions.checkArgument(null == jobConfigurationProperties.getElasticJobClass()
                                || Strings.isNullOrEmpty(jobConfigurationProperties.getElasticJobType()),
                        "[elasticJobClass] and [elasticJobType] are mutually exclusive.");
                if (null != jobConfigurationProperties.getElasticJobClass()) {
                    registerClassedJob(entry.getKey(), entry.getValue().getJobBootstrapBeanName(), singletonBeanRegistry, registryCenter, tracingConfig, jobConfigurationProperties);
                } else if (!Strings.isNullOrEmpty(jobConfigurationProperties.getElasticJobType())) {
                    registerTypedJob(entry.getKey(), entry.getValue().getJobBootstrapBeanName(), singletonBeanRegistry, registryCenter, tracingConfig, jobConfigurationProperties);
                }
            }
        }

通过源码可知此方法遍历了配置文件中的所有job配置，然后根据job的类型分别调用各自的registerJob方法。这里我们看一下常用的registerClassedJob方法

private void registerClassedJob(final String jobName, final String jobBootstrapBeanName, final SingletonBeanRegistry singletonBeanRegistry, final CoordinatorRegistryCenter registryCenter,
                                        final TracingConfiguration<?> tracingConfig, final ElasticJobConfigurationProperties jobConfigurationProperties) {
            JobConfiguration jobConfig = jobConfigurationProperties.toJobConfiguration(jobName);
            Optional.ofNullable(tracingConfig).ifPresent(jobConfig.getExtraConfigurations()::add);
            ElasticJob elasticJob = applicationContext.getBean(jobConfigurationProperties.getElasticJobClass());
            if (Strings.isNullOrEmpty(jobConfig.getCron())) {
                Preconditions.checkArgument(!Strings.isNullOrEmpty(jobBootstrapBeanName), "The property [jobBootstrapBeanName] is required for One-off job.");
                singletonBeanRegistry.registerSingleton(jobBootstrapBeanName, new OneOffJobBootstrap(registryCenter, elasticJob, jobConfig));
            } else {
                String beanName = !Strings.isNullOrEmpty(jobBootstrapBeanName) ? jobBootstrapBeanName : jobConfig.getJobName() + "ScheduleJobBootstrap";
                singletonBeanRegistry.registerSingleton(beanName, new ScheduleJobBootstrap(registryCenter, elasticJob, jobConfig));
            }
        }

对于有配置cron表达式的方法会执行下面这行代码

> singletonBeanRegistry.registerSingleton(beanName, new ScheduleJobBootstrap(registryCenter, elasticJob, jobConfig));

通过它会在spring容器中定义一个ScheduleJobBootstrap的bean，通过代码我们也可知：每个job都会创建一个对应的ScheduleJobBootstrap

## ScheduleJobBootstrap ##

通过上面的分析我们知道ScheduleJobBootstrap应该是一个很重要的类，继续看一下它的构造方法

public ScheduleJobBootstrap(final CoordinatorRegistryCenter regCenter, final ElasticJob elasticJob, final JobConfiguration jobConfig) {
            jobScheduler = new JobScheduler(regCenter, elasticJob, jobConfig);
        }

接着看一下JobScheduler类的构造方法

public JobScheduler(final CoordinatorRegistryCenter regCenter, final ElasticJob elasticJob, final JobConfiguration jobConfig) {
            Preconditions.checkArgument(null != elasticJob, "Elastic job cannot be null.");
            this.regCenter = regCenter;
            //获取出job的listener列表,配置job的listener有啥用还要研究一下
            Collection<ElasticJobListener> jobListeners = getElasticJobListeners(jobConfig);
            //setupFacade准备
            setUpFacade = new SetUpFacade(regCenter, jobConfig.getJobName(), jobListeners);
            String jobClassName = JobClassNameProviderFactory.getProvider().getJobClassName(elasticJob);
            //设置zk的配置信息
            this.jobConfig = setUpFacade.setUpJobConfiguration(jobClassName, jobConfig);
            //SchedulerFacade准备
            schedulerFacade = new SchedulerFacade(regCenter, jobConfig.getJobName());
            //jobFacade准备
            jobFacade = new LiteJobFacade(regCenter, jobConfig.getJobName(), jobListeners, findTracingConfiguration().orElse(null));
            validateJobProperties();
            //创建对应的ElasticJobExecutor
            jobExecutor = new ElasticJobExecutor(elasticJob, this.jobConfig, jobFacade);
            setGuaranteeServiceForElasticJobListeners(regCenter, jobListeners);
            jobScheduleController = createJobScheduleController();
        }

**jobExecutor = new ElasticJobExecutor(elasticJob, this.jobConfig, jobFacade);** 这行代码创建了一个ElasticJobExecutor，ElasticJobExecutor又通过**JobItemExecutorFactory.getExecutor(elasticJob.getClass())** 创建了一个JobItemExecutor

package org.apache.shardingsphere.elasticjob.executor;
    
    import lombok.extern.slf4j.Slf4j;
    import org.apache.shardingsphere.elasticjob.api.ElasticJob;
    import org.apache.shardingsphere.elasticjob.api.JobConfiguration;
    import org.apache.shardingsphere.elasticjob.error.handler.JobErrorHandler;
    import org.apache.shardingsphere.elasticjob.executor.context.ExecutorContext;
    import org.apache.shardingsphere.elasticjob.executor.item.JobItemExecutor;
    import org.apache.shardingsphere.elasticjob.executor.item.JobItemExecutorFactory;
    import org.apache.shardingsphere.elasticjob.infra.env.IpUtils;
    import org.apache.shardingsphere.elasticjob.infra.exception.ExceptionUtils;
    import org.apache.shardingsphere.elasticjob.infra.exception.JobExecutionEnvironmentException;
    import org.apache.shardingsphere.elasticjob.infra.listener.ShardingContexts;
    import org.apache.shardingsphere.elasticjob.tracing.event.JobExecutionEvent;
    import org.apache.shardingsphere.elasticjob.tracing.event.JobExecutionEvent.ExecutionSource;
    import org.apache.shardingsphere.elasticjob.tracing.event.JobStatusTraceEvent.State;
    
    import java.util.Collection;
    import java.util.Map;
    import java.util.concurrent.ConcurrentHashMap;
    import java.util.concurrent.CountDownLatch;
    import java.util.concurrent.ExecutorService;
    
    /**
     * ElasticJob executor.
     */
    @Slf4j
    public final class ElasticJobExecutor {
        
        private final ElasticJob elasticJob;
        
        private final JobFacade jobFacade;
        
        private final JobItemExecutor jobItemExecutor;
        
        private final ExecutorContext executorContext;
        
        private final Map<Integer, String> itemErrorMessages;
        
        public ElasticJobExecutor(final ElasticJob elasticJob, final JobConfiguration jobConfig, final JobFacade jobFacade) {
            this(elasticJob, jobConfig, jobFacade, JobItemExecutorFactory.getExecutor(elasticJob.getClass()));
        }
        
        private ElasticJobExecutor(final ElasticJob elasticJob, final JobConfiguration jobConfig, final JobFacade jobFacade, final JobItemExecutor jobItemExecutor) {
            this.elasticJob = elasticJob;
            this.jobFacade = jobFacade;
            this.jobItemExecutor = jobItemExecutor;
            executorContext = new ExecutorContext(jobFacade.loadJobConfiguration(true));
            itemErrorMessages = new ConcurrentHashMap<>(jobConfig.getShardingTotalCount(), 1);
        }
    
        ……
        ……
        ……
        
    }
        
        
        
        
        
    package org.apache.shardingsphere.elasticjob.executor.item;
    
    import lombok.AccessLevel;
    import lombok.NoArgsConstructor;
    import org.apache.shardingsphere.elasticjob.api.ElasticJob;
    import org.apache.shardingsphere.elasticjob.infra.exception.JobConfigurationException;
    import org.apache.shardingsphere.elasticjob.executor.item.impl.ClassedJobItemExecutor;
    import org.apache.shardingsphere.elasticjob.executor.item.impl.TypedJobItemExecutor;
    import org.apache.shardingsphere.elasticjob.infra.spi.ElasticJobServiceLoader;
    
    import java.util.HashMap;
    import java.util.Map;
    import java.util.Map.Entry;
    import java.util.ServiceLoader;
    
    /**
     * Job item executor factory.
     */
    @SuppressWarnings("rawtypes")
    @NoArgsConstructor(access = AccessLevel.PRIVATE)
    public final class JobItemExecutorFactory {
        
            private static final Map<Class, ClassedJobItemExecutor> CLASSED_EXECUTORS = new HashMap<>();
            
            static {
                ElasticJobServiceLoader.registerTypedService(TypedJobItemExecutor.class);
                //通过spi的方式加载出所有的ClassedJobItemExecutor实现类，然后放到CLASSED_EXECUTORS中
                ServiceLoader.load(ClassedJobItemExecutor.class).forEach(each -> CLASSED_EXECUTORS.put(each.getElasticJobClass(), each));
            }
            /**
             * Get executor.
             * 
             * @param elasticJobClass elastic job class
             * @return job item executor
             */
            @SuppressWarnings("unchecked")
            public static JobItemExecutor getExecutor(final Class<? extends ElasticJob> elasticJobClass) {
                for (Entry<Class, ClassedJobItemExecutor> entry : CLASSED_EXECUTORS.entrySet()) {
                    if (entry.getKey().isAssignableFrom(elasticJobClass)) {
                        return entry.getValue();
                    }
                }
                throw new JobConfigurationException("Can not find executor for elastic job class `%s`", elasticJobClass.getName());
            }
            ……
            ……
            ……
    }

进入末尾的createJobScheduleController方法，看一下它是如何创建JobScheduleController的

private JobScheduleController createJobScheduleController() {
            JobScheduleController result = new JobScheduleController(createScheduler(), createJobDetail(), getJobConfig().getJobName());
            JobRegistry.getInstance().registerJob(getJobConfig().getJobName(), result);
            registerStartUpInfo();
            return result;
        }

createJobScheduleController方法中创建了JobScheduleController，我们关注一下它的两个构造参数:

*  createScheduler()
 *  createJobDetail()

Scheduler和JobDetail是quartz中的两个job定义时的术语，因为elastic-job也是基于quartz开发的，所以这上面两个方法应该就是job定义的代码了，具体看一下

createScheduler方法

private Scheduler createScheduler() {
            Scheduler result;
            try {
                StdSchedulerFactory factory = new StdSchedulerFactory();
                factory.initialize(getQuartzProps());
                result = factory.getScheduler();
                result.getListenerManager().addTriggerListener(schedulerFacade.newJobTriggerListener());
            } catch (final SchedulerException ex) {
                throw new JobSystemException(ex);
            }
            return result;
        }
        
        private Properties getQuartzProps() {
            Properties result = new Properties();
            result.put("org.quartz.threadPool.class", SimpleThreadPool.class.getName());
            result.put("org.quartz.threadPool.threadCount", "1");
            result.put("org.quartz.scheduler.instanceName", getJobConfig().getJobName());
            result.put("org.quartz.jobStore.misfireThreshold", "1");
            result.put("org.quartz.plugin.shutdownhook.class", JobShutdownHookPlugin.class.getName());
            result.put("org.quartz.plugin.shutdownhook.cleanShutdown", Boolean.TRUE.toString());
            return result;
        }

createJobDetail方法

private JobDetail createJobDetail() {
            JobDetail result = JobBuilder.newJob(LiteJob.class).withIdentity(getJobConfig().getJobName()).build();
            result.getJobDataMap().put(JOB_EXECUTOR_DATA_MAP_KEY, jobExecutor);
            return result;
        }

这个方法很重要，它定义了一个Litejob，Litejob实现了org.quartz.Job类。

package org.apache.shardingsphere.elasticjob.lite.internal.schedule;
    
    import lombok.Setter;
    import org.apache.shardingsphere.elasticjob.executor.ElasticJobExecutor;
    import org.quartz.Job;
    import org.quartz.JobExecutionContext;
    
    /**
     * Lite job class.
     */
    @Setter
    public final class LiteJob implements Job {
        
        private ElasticJobExecutor jobExecutor;
        
        @Override
        public void execute(final JobExecutionContext context) {
            jobExecutor.execute();
        }
        
    }

从代码可以看出每个elasitc-job中的定时任务其实在quartz中仍然是以Litejob执行的，elastic-job中的定时任务与Litejob是一对一的关系,每次elastic-job中定时任务的执行实际上都是执行的LiteJob中的jobExecutor.execute方法

再看到 **result.getJobDataMap().put(JOB\_EXECUTOR\_DATA\_MAP\_KEY, jobExecutor)** 这行，它往刚刚定义quartz的Litejob中传入了jobExecutor

对于jobExecutor的构造细节先不看，直接看它的execute方法

org.apache.shardingsphere.elasticjob.executor.ElasticJobExecutor\#execute()

/**
         * Execute job.
         */
        public void execute() {
            JobConfiguration jobConfig = jobFacade.loadJobConfiguration(true);
            executorContext.reloadIfNecessary(jobConfig);
            JobErrorHandler jobErrorHandler = executorContext.get(JobErrorHandler.class);
            try {
                jobFacade.checkJobExecutionEnvironment();
            } catch (final JobExecutionEnvironmentException cause) {
                jobErrorHandler.handleException(jobConfig.getJobName(), cause);
            }
            //获取分片上下文
            ShardingContexts shardingContexts = jobFacade.getShardingContexts();
            jobFacade.postJobStatusTraceEvent(shardingContexts.getTaskId(), State.TASK_STAGING, String.format("Job '%s' execute begin.", jobConfig.getJobName()));
            if (jobFacade.misfireIfRunning(shardingContexts.getShardingItemParameters().keySet())) {
                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.", jobConfig.getJobName(),
                        shardingContexts.getShardingItemParameters().keySet()));
                return;
            }
            try {
                jobFacade.beforeJobExecuted(shardingContexts);
                //CHECKSTYLE:OFF
            } catch (final Throwable cause) {
                //CHECKSTYLE:ON
                jobErrorHandler.handleException(jobConfig.getJobName(), cause);
            }
            //执行定时任务
            execute(jobConfig, shardingContexts, ExecutionSource.NORMAL_TRIGGER);
            while (jobFacade.isExecuteMisfired(shardingContexts.getShardingItemParameters().keySet())) {
                jobFacade.clearMisfire(shardingContexts.getShardingItemParameters().keySet());
                execute(jobConfig, shardingContexts, ExecutionSource.MISFIRE);
            }
            jobFacade.failoverIfNecessary();
            try {
                jobFacade.afterJobExecuted(shardingContexts);
                //CHECKSTYLE:OFF
            } catch (final Throwable cause) {
                //CHECKSTYLE:ON
                jobErrorHandler.handleException(jobConfig.getJobName(), cause);
            }
        }

从上面可以看出，elastic-job中定时任务的执行实际上执行的是ElasticJobExecutor中的execute方法，然后通过**ShardingContexts shardingContexts = jobFacade.getShardingContexts();** 这行获取出一个分片的上下文信息,随后传入到有参的execute方法中,看一个execute的有参方法：

private void execute(final JobConfiguration jobConfig, final ShardingContexts shardingContexts, final ExecutionSource executionSource) {
            //如果没有分片参数则不进行执行
            if (shardingContexts.getShardingItemParameters().isEmpty()) {
                jobFacade.postJobStatusTraceEvent(shardingContexts.getTaskId(), State.TASK_FINISHED, String.format("Sharding item for job '%s' is empty.", jobConfig.getJobName()));
                return;
            }
            jobFacade.registerJobBegin(shardingContexts);
            String taskId = shardingContexts.getTaskId();
            jobFacade.postJobStatusTraceEvent(taskId, State.TASK_RUNNING, "");
            try {
                //定时任务处理
                process(jobConfig, shardingContexts, executionSource);
            } finally {
                // TODO Consider increasing the status of job failure, and how to handle the overall loop of job failure
                jobFacade.registerJobCompleted(shardingContexts);
                if (itemErrorMessages.isEmpty()) {
                    jobFacade.postJobStatusTraceEvent(taskId, State.TASK_FINISHED, "");
                } else {
                    jobFacade.postJobStatusTraceEvent(taskId, State.TASK_ERROR, itemErrorMessages.toString());
                }
            }
        }

查看ElasticJobExecutor的有参execute方法可知：如果shardingContexts的getShardingItemParameters是空的话，则不进行实际的定时任务执行；否则调用process方法执行定时任务，process方法代码如下：

private void process(final JobConfiguration jobConfig, final ShardingContexts shardingContexts, final ExecutionSource executionSource) {
            Collection<Integer> items = shardingContexts.getShardingItemParameters().keySet();
            if (1 == items.size()) {
                int item = shardingContexts.getShardingItemParameters().keySet().iterator().next();
                JobExecutionEvent jobExecutionEvent = new JobExecutionEvent(IpUtils.getHostName(), IpUtils.getIp(), shardingContexts.getTaskId(), jobConfig.getJobName(), executionSource, item);
                process(jobConfig, shardingContexts, item, jobExecutionEvent);
                return;
            }
            CountDownLatch latch = new CountDownLatch(items.size());
            for (int each : items) {
                JobExecutionEvent jobExecutionEvent = new JobExecutionEvent(IpUtils.getHostName(), IpUtils.getIp(), shardingContexts.getTaskId(), jobConfig.getJobName(), executionSource, each);
                ExecutorService executorService = executorContext.get(ExecutorService.class);
                if (executorService.isShutdown()) {
                    return;
                }
                executorService.submit(() -> {
                    try {
                        process(jobConfig, shardingContexts, each, jobExecutionEvent);
                    } finally {
                        latch.countDown();
                    }
                });
            }
            try {
                latch.await();
            } catch (final InterruptedException ex) {
                Thread.currentThread().interrupt();
            }
        }

@SuppressWarnings("unchecked")
        private void process(final JobConfiguration jobConfig, final ShardingContexts shardingContexts, final int item, final JobExecutionEvent startEvent) {
            jobFacade.postJobExecutionEvent(startEvent);
            log.trace("Job '{}' executing, item is: '{}'.", jobConfig.getJobName(), item);
            JobExecutionEvent completeEvent;
            try {
                jobItemExecutor.process(elasticJob, jobConfig, jobFacade, shardingContexts.createShardingContext(item));
                completeEvent = startEvent.executionSuccess();
                log.trace("Job '{}' executed, item is: '{}'.", jobConfig.getJobName(), item);
                jobFacade.postJobExecutionEvent(completeEvent);
                // CHECKSTYLE:OFF
            } catch (final Throwable cause) {
                // CHECKSTYLE:ON
                completeEvent = startEvent.executionFailure(ExceptionUtils.transform(cause));
                jobFacade.postJobExecutionEvent(completeEvent);
                itemErrorMessages.put(item, ExceptionUtils.transform(cause));
                JobErrorHandler jobErrorHandler = executorContext.get(JobErrorHandler.class);
                jobErrorHandler.handleException(jobConfig.getJobName(), cause);
            }
        }

看到这里之后，一切就归于平静了

## 总结 ##

在翻看了elasticjob的部分源码后，再回到本文最开始的思考：  
**在elastic-job多节点的环境中，单分片任务只会在一个节点运行呢？**

因为在执行定时任务时会先获取一个ShardingContexts，如果获取出来的ShardingContexts的getShardingItemParameters是空的，则不会实际执行定时任务

那么新的思考又来了，jobNodeStorage中分片信息是何时定义的呢?

*感兴趣的可以继续绕着下面这个类继续查看*

org.apache.shardingsphere.elasticjob.lite.internal.schedule.LiteJobFacade\#getShardingContexts

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);
        }
        
        /**
         * Get sharding items from localhost job server.
         * 
         * @return sharding items from localhost job server
         */
        public List<Integer> getLocalShardingItems() {
            if (JobRegistry.getInstance().isShutdown(jobName) || !serverService.isAvailableServer(JobRegistry.getInstance().getJobInstance(jobName).getIp())) {
                return Collections.emptyList();
            }
            return getShardingItems(JobRegistry.getInstance().getJobInstance(jobName).getJobInstanceId());
        }
        
        
        /**
         * Get sharding items.
         *
         * @param jobInstanceId job instance ID
         * @return sharding items
         */
        public List<Integer> getShardingItems(final String jobInstanceId) {
            JobInstance jobInstance = new JobInstance(jobInstanceId);
            if (!serverService.isAvailableServer(jobInstance.getIp())) {
                return Collections.emptyList();
            }
            List<Integer> result = new LinkedList<>();
            int shardingTotalCount = configService.load(true).getShardingTotalCount();
            for (int i = 0; i < shardingTotalCount; i++) {
                if (jobInstance.getJobInstanceId().equals(jobNodeStorage.getJobNodeData(ShardingNode.getInstanceNode(i)))) {
                    result.add(i);
                }
            }
            return result;
        }

[elastic-job]: https://puhaiyang.blog.csdn.net/article/details/113925399