Elastic-Job源码分析之JobScheduler类分析

小程序：扫一扫查出行
【扫一扫了解最新限行尾号】
复制小程序

JobScheduler这个类是EJ中比较核心的一个类，我们现在开始解析这个类。

一、构造器

首先我们看一下JobScheduler的几个构造器。

    private JobScheduler(final CoordinatorRegistryCenter regCenter, final LiteJobConfiguration liteJobConfig, final JobEventBus jobEventBus, final ElasticJobListener... elasticJobListeners) {

        JobRegistry.getInstance().addJobInstance(liteJobConfig.getJobName(), new JobInstance());

        this.liteJobConfig = liteJobConfig;

        this.regCenter = regCenter;

        List<ElasticJobListener> elasticJobListenerList = Arrays.asList(elasticJobListeners);

        setGuaranteeServiceForElasticJobListeners(regCenter, elasticJobListenerList);

        schedulerFacade = new SchedulerFacade(regCenter, liteJobConfig.getJobName(), elasticJobListenerList);

        jobFacade = new LiteJobFacade(regCenter, liteJobConfig.getJobName(), Arrays.asList(elasticJobListeners), jobEventBus);

    }

这个构造器比较清晰，第一个参数regCenter表示的是注册中心，这里我们用的是zk作为我们的注册中心，所以这块的配置我们一般从xml配置文件中读取即可。我们按照官方文档配置即可。

    <!--配置作业注册中心 -->

    <reg:zookeeper id="regCenter" server-lists="ip1:2181,ip2:2181"

                   namespace="your-job-name" base-sleep-time-milliseconds="1000" max-sleep-time-milliseconds="3000"

                   max-retries="3"/>

然后从xml文件中取出该配置即可。

ApplicationContext context = new ClassPathXmlApplicationContext("classpath:spring/applicationContext-job.xml");

		ZookeeperRegistryCenter zookeeperRegistryCenter = context.getBean(ZookeeperRegistryCenter.class);

第二个参数liteJobConfig表示的是作业的一些配置，这块我们后续再细细了解。

第三个参数jobEventBus表示作业运行痕迹总线。如果想把作业运行的内容写到DB中，我们需要用到另一个构造器，同时定义自己的JobEventConfiguration，目前来说实现这个接口的只有一个类JobEventRdbConfiguration，通过这个可以将作业运行的痕迹进行持久化到DB的操作。

第四个参数elasticJobListeners表示一些监听器，这里我们可以定义一些监听器，具体的监听器我们可以参考官方文档。

二、初始化

定义好我们的作业后，怎么将我们的作业运行起来呢？这块就需要用到我们的初始化作业的方法init()。

public void init() {

    JobRegistry.getInstance().setCurrentShardingTotalCount(liteJobConfig.getJobName(), liteJobConfig.getTypeConfig().getCoreConfig().getShardingTotalCount());

    JobScheduleController jobScheduleController = new JobScheduleController(createScheduler(), createJobDetail(liteJobConfig.getTypeConfig().getJobClass()), liteJobConfig.getJobName());

    JobRegistry.getInstance().registerJob(liteJobConfig.getJobName(), jobScheduleController, regCenter);

    schedulerFacade.registerStartUpInfo(liteJobConfig);

    jobScheduleController.scheduleJob(liteJobConfig.getTypeConfig().getCoreConfig().getCron());

    }

初始化主要做了以下几个动作：

设置当前分片总数setCurrentShardingTotalCount
创建调度器createScheduler
创建任务详情createJobDetail
将作业在zk上面进行注册registerJob
注册作业启动信息registerStartUpInfo
开始进行作业调度scheduleJob

这几个动作是一个作业能够最终被调度的关键，我们来一一分析。

2.1 createScheduler

    private Scheduler createScheduler() {

        Scheduler result;

        try {

            StdSchedulerFactory factory = new StdSchedulerFactory();

            factory.initialize(getBaseQuartzProperties());//TODO 每次新建任务都会创建一个线程，会导致线程数过多

            result = factory.getScheduler();

        } catch (final SchedulerException ex) {

            throw new JobSystemException(ex);

        }

        return result;

    }

这个方法调用了quartz底层的方法，创建了一个调度器。我们看看quartz中的一些配置：

private Properties getBaseQuartzProperties() {

    Properties result = new Properties();

    result.put("org.quartz.threadPool.class", org.quartz.simpl.SimpleThreadPool.class.getName());

    result.put("org.quartz.threadPool.threadCount", "1");

    result.put("org.quartz.scheduler.instanceName", liteJobConfig.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;

}

我们可以看到一个比较重要的参数：org.quartz.threadPool.threadCount，这里设置为了1，也就是说每个任务，EJ默认都会创建一个线程来进行调度，所以如果想动态的创建任务，你就要考虑好你的内存够不够这些线程使用了。

2.2 createJobDetail

private JobDetail createJobDetail(final String jobClass) {

    JobDetail result = JobBuilder.newJob(LiteJob.class).withIdentity(liteJobConfig.getJobName()).build();

    result.getJobDataMap().put(JOB_FACADE_DATA_MAP_KEY, jobFacade);

    Optional<ElasticJob> elasticJobInstance = createElasticJobInstance();

    if (elasticJobInstance.isPresent()) {

        result.getJobDataMap().put(ELASTIC_JOB_DATA_MAP_KEY, elasticJobInstance.get());

    } else if (!jobClass.equals(ScriptJob.class.getCanonicalName())) {

        try {

            result.getJobDataMap().put(ELASTIC_JOB_DATA_MAP_KEY, Class.forName(jobClass).newInstance());

        } catch (final ReflectiveOperationException ex) {

            throw new JobConfigurationException("Elastic-Job: Job class '%s' can not initialize.", jobClass);

        }

    }

    return result;

}

这个方法创建了任务的一些属性。

首先调用JobBuilder的构造器方法，生成了一个详细的任务。
然后将作业的门面放入到作业内部的一个作业数据map中，相当于是一个缓存。
调用createElasticJobInstance创建实例，生成一个Optional的类，这个Optional是Guava中的一个类，后续咱们再研究。我们只需要知道创建了一个实例就行了，实例放在了Optional这个容器中，而Optional可以做一些额外的判断。我们可以联想一下，利用Optional来处理null？
判断实例是否创建成功，通过isPresent()方法来判断。Optional的这个方法判断实例是否为空。
- 如果不为空，就把实例放入到缓存map中。
- 否则，调用反射，把实例放入缓存map中。如果异常的话，直接抛出异常，说明实例化失败。

2.3 registerJob

看方法名我们就能知道，这个方法的作用是注册我们的任务。

/**

 * 添加作业调度控制器.

 *

 * @param jobName 作业名称

 * @param jobScheduleController 作业调度控制器

 * @param regCenter 注册中心

 */

public void registerJob(final String jobName, final JobScheduleController jobScheduleController, final CoordinatorRegistryCenter regCenter) {

    schedulerMap.put(jobName, jobScheduleController);

    regCenterMap.put(jobName, regCenter);

    regCenter.addCacheData("/" + jobName);

}

这步的操作，把作业控制调度器放入到调度缓存中，把任务信息放入到regCenter的缓存中。另外还把任务名称加入到本地缓存中。

2.4 registerStartUpInfo

这个方法是注册一些任务的启动信息，这块的步骤比较多，需要细细理解一下。

    public void registerStartUpInfo(final boolean enabled) {

        listenerManager.startAllListeners();

        leaderService.electLeader();

        serverService.persistOnline(enabled);

        instanceService.persistOnline();

        shardingService.setReshardingFlag();

        monitorService.listen();

        if (!reconcileService.isRunning()) {

            reconcileService.startAsync();

        }

    }

第一步，启动所有监听器。

    public void startAllListeners() {

        electionListenerManager.start();//主节点选举

        shardingListenerManager.start();//分片

        failoverListenerManager.start();//失效转移

        shutdownListenerManager.start();//关闭

        triggerListenerManager.start();//触发器

        rescheduleListenerManager.start();//重新调度

        guaranteeListenerManager.start();//保证分布式任务全部开始和结束状态

        jobNodeStorage.addConnectionStateListener(regCenterConnectionStateListener);//注册中心与任务节点的连接状态

    }