分布式作业系统 Elastic-Job-Cloud 源码分析 —— 作业调度（一）

// ProducerManager.java
/**
* 调度作业.
* 
* @param jobConfig 作业配置
*/
public void schedule(final CloudJobConfiguration jobConfig) {
   // 应用 或 作业 被禁用，不调度
   if (disableAppService.isDisabled(jobConfig.getAppName()) || disableJobService.isDisabled(jobConfig.getJobName())) {
       return;
   }
   if (CloudJobExecutionType.TRANSIENT == jobConfig.getJobExecutionType()) { // 瞬时作业
       transientProducerScheduler.register(jobConfig);
   } else if (CloudJobExecutionType.DAEMON == jobConfig.getJobExecutionType()) { // 常驻作业
       readyService.addDaemon(jobConfig.getJobName());
   }
}

// ReadyService.java
/**
* 将常驻作业放入待执行队列.
*
* @param jobName 作业名称
*/
public void addDaemon(final String jobName) {
   if (regCenter.getNumChildren(ReadyNode.ROOT) > env.getFrameworkConfiguration().getJobStateQueueSize()) {
       log.warn("Cannot add daemon job, caused by read state queue size is larger than {}.", env.getFrameworkConfiguration().getJobStateQueueSize());
       return;
   }
   Optional<CloudJobConfiguration> cloudJobConfig = configService.load(jobName);
   if (!cloudJobConfig.isPresent() || CloudJobExecutionType.DAEMON != cloudJobConfig.get().getJobExecutionType() || runningService.isJobRunning(jobName)) {
       return;
   }
   // 添加到待执行队列
   regCenter.persist(ReadyNode.getReadyJobNodePath(jobName), "1");
}

// ReadyNode.java
final class ReadyNode {

    static final String ROOT = StateNode.ROOT + "/ready";

    private static final String READY_JOB = ROOT + "/%s"; // %s = ${JOB_NAME}
}

// TransientProducerScheduler.java
void start() {
   scheduler = getScheduler();
   try {
       scheduler.start();
   } catch (final SchedulerException ex) {
       throw new JobSystemException(ex);
   }
}

private Scheduler getScheduler() {
   StdSchedulerFactory factory = new StdSchedulerFactory();
   try {
       factory.initialize(getQuartzProperties());
       return factory.getScheduler();
   } catch (final SchedulerException ex) {
       throw new JobSystemException(ex);
   }
}

private Properties getQuartzProperties() {
   Properties result = new Properties();
   result.put("org.quartz.threadPool.class", SimpleThreadPool.class.getName());
   result.put("org.quartz.threadPool.threadCount", Integer.toString(Runtime.getRuntime().availableProcessors() * 2)); // 线程池数量
   result.put("org.quartz.scheduler.instanceName", "ELASTIC_JOB_CLOUD_TRANSIENT_PRODUCER");
   result.put("org.quartz.plugin.shutdownhook.class", ShutdownHookPlugin.class.getName());
   result.put("org.quartz.plugin.shutdownhook.cleanShutdown", Boolean.TRUE.toString());
   return result;
}

// TransientProducerScheduler.java
private final TransientProducerRepository repository;

synchronized void register(final CloudJobConfiguration jobConfig) {
   String cron = jobConfig.getTypeConfig().getCoreConfig().getCron();
   // 添加 cron 作业集合
   JobKey jobKey = buildJobKey(cron);
   repository.put(jobKey, jobConfig.getJobName());
   // 调度 作业
   try {
       if (!scheduler.checkExists(jobKey)) {
           scheduler.scheduleJob(buildJobDetail(jobKey), buildTrigger(jobKey.getName()));
       }
   } catch (final SchedulerException ex) {
       throw new JobSystemException(ex);
   }
}

调用 TransientProducerRepository#put(...) 以 Quartz JobKey 为主键聚合作业。
final class TransientProducerRepository {/**
 * cron 作业集合
 * key：作业Key
 */
private final ConcurrentHashMap<JobKey, List<String>> cronTasks = new ConcurrentHashMap<>(256, 1);

synchronized void put(final JobKey jobKey, final String jobName) {
    remove(jobName);
    List<String> taskList = cronTasks.get(jobKey);
 if (null == taskList) {
        taskList = new CopyOnWriteArrayList<>();
        taskList.add(jobName);
        cronTasks.put(jobKey, taskList);
 return;
    }
 if (!taskList.contains(jobName)) {
        taskList.add(jobName);
    }
}

}

调用 #buildJobDetail(...) 创建 Quartz Job 信息。实现代码如下：
private JobDetail buildJobDetail(final JobKey jobKey) {
    JobDetail result = JobBuilder.newJob(ProducerJob.class) // ProducerJob.java
            .withIdentity(jobKey).build();
    result.getJobDataMap().put("repository", repository);
    result.getJobDataMap().put("readyService", readyService);
 return result;
}

JobBuilder#newJob(…) 的参数是 ProducerJob，下文会讲解到。
调用 #buildTrigger(...) 创建 Quartz Trigger。实现代码如下：
private Trigger buildTrigger(final String cron) {
 return TriggerBuilder.newTrigger()
           .withIdentity(cron)
           .withSchedule(CronScheduleBuilder.cronSchedule(cron) // cron
           .withMisfireHandlingInstructionDoNothing())
           .build();
}

public static final class ProducerJob implements Job {

   private TransientProducerRepository repository;

   private ReadyService readyService;

   @Override
   public void execute(final JobExecutionContext context) throws JobExecutionException {
       List<String> jobNames = repository.get(context.getJobDetail().getKey());
       for (String each : jobNames) {
           readyService.addTransient(each);
       }
   }
}

调用 TransientProducerRepository#get(...) 方法，获得该 Job 对应的作业集合。实现代码如下：
final class TransientProducerRepository {/**
 * cron 作业集合
 * key：作业Key
 */
private final ConcurrentHashMap<JobKey, List<String>> cronTasks = new ConcurrentHashMap<>(256, 1);

List<String> get(final JobKey jobKey) {
    List<String> result = cronTasks.get(jobKey);
 return null == result ? Collections.<String>emptyList() : result;
}

}

调用 ReadyService#addTransient(...) 方法，添加瞬时作业到待执行作业队列。实现代码如下：
/**
* 将瞬时作业放入待执行队列.
* 
* @param jobName 作业名称
*/
public void addTransient(final String jobName) {
 //
 if (regCenter.getNumChildren(ReadyNode.ROOT) > env.getFrameworkConfiguration().getJobStateQueueSize()) {
       log.warn("Cannot add transient job, caused by read state queue size is larger than {}.", env.getFrameworkConfiguration().getJobStateQueueSize());
 return;
   }
 //
   Optional<CloudJobConfiguration> cloudJobConfig = configService.load(jobName);
 if (!cloudJobConfig.isPresent() || CloudJobExecutionType.TRANSIENT != cloudJobConfig.get().getJobExecutionType()) {
 return;
   }
 // 
   String readyJobNode = ReadyNode.getReadyJobNodePath(jobName);
   String times = regCenter.getDirectly(readyJobNode);
 if (cloudJobConfig.get().getTypeConfig().getCoreConfig().isMisfire()) {
       regCenter.persist(readyJobNode, Integer.toString(null == times ? 1 : Integer.parseInt(times) + 1));
   } else {
       regCenter.persist(ReadyNode.getReadyJobNodePath(jobName), "1");
   }
}