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

原创：老艿艿芋道源码 2019-05-13

点击上方“芋道源码”，选择“设为星标”

做积极的人，而不是积极废人！

源码精品专栏

摘要: 原创出处 http://www.iocoder.cn/Elastic-Job/cloud-job-scheduler-and-executor-second/ 「芋道源码」欢迎转载，保留摘要，谢谢！

本文基于 Elastic-Job V2.1.5 版本分享

1. 概述
2. 云作业操作
3. 云作业应用操作

1. 概述

本文主要分享 Elastic-Job-Cloud 云作业应用配置和云作业配置变更对作业调度的影响，作为《Elastic-Job-Cloud 源码分析 —— 作业调度（一）》的补充内容。所以需要你对作业调度已经有一定了解的基础上。

🙂 如果你做作业调度有任何想交流，欢迎加我的公众号( 芋道源码 ) 或微信( wangwenbin-server ) 交流。

你行好事会因为得到赞赏而愉悦
同理，开源项目贡献者会因为 Star 而更加有动力
为 Elastic-Job 点赞！传送门

2. 云作业操作

我们可以使用运维平台或 Restful API 对云作业进行操作。前者是对后者的界面包装，如下图所示：

2.1 注册云作业配置

《Elastic-Job-Cloud 源码分析 —— 作业配置》「3.1.1 操作云作业配置」有详细解析。

2.2 禁用云作业

调用 CloudJobRestfulApi#disable(...) 方法，禁用云作业，实现代码如下：

@POST
@Path("/{jobName}/disable")
public void disable(@PathParam("jobName") final String jobName) {
   if (configService.load(jobName).isPresent()) {
       // 将作业放入禁用队列
       facadeService.disableJob(jobName);
       // 停止调度作业
       producerManager.unschedule(jobName);
   }
}

调用 FacadeService#disableJob(...) 方法，将作业放入禁用作业队列。实现代码如下：
// FacadeService.java public void disableJob(final String jobName) { disableJobService.add(jobName); } // DisableJobService.java public void add(final String jobName) { if (regCenter.getNumChildren(DisableJobNode.ROOT) > env.getFrameworkConfiguration().getJobStateQueueSize()) { log.warn("Cannot add disable job, caused by read state queue size is larger than {}.", env.getFrameworkConfiguration().getJobStateQueueSize()); return; } // 将作业放入禁用队列 String disableJobNodePath = DisableJobNode.getDisableJobNodePath(jobName); if (!regCenter.isExisted(disableJobNodePath)) { regCenter.persist(disableJobNodePath, jobName); } } // DisableJobNode.java final class DisableJobNode {static final String ROOT = StateNode.ROOT + "/disable/job"; private static final String DISABLE_JOB = ROOT + "/%s"; // %s = ${JOB_NAME} }

DisableJobService，禁用作业队列服务。
禁用作业队列存储在注册中心( Zookeeper )的持久数据节点 /${NAMESPACE}/state/disable/job/${JOB_NAME}，存储值为作业名称。使用 zkClient 查看如下：
[zk: localhost:2181(CONNECTED) 6] ls /elastic-job-cloud/state/disable/job [test_job_simple]

调用 ProducerManager#unschedule(...) 方法，停止调度作业。实现代码如下：
public void unschedule(final String jobName) { // 杀死作业对应的 Mesos 任务们 for (TaskContext each : runningService.getRunningTasks(jobName)) { schedulerDriver.killTask(Protos.TaskID.newBuilder().setValue(each.getId()).build()); } // 将作业从运行时队列删除 runningService.remove(jobName); // 将作业从待运行队列删除 readyService.remove(Lists.newArrayList(jobName)); // 停止作业调度 Optional<CloudJobConfiguration> jobConfig = configService.load(jobName); if (jobConfig.isPresent()) { transientProducerScheduler.deregister(jobConfig.get()); } }

*
调用 SchedulerDriver#killTask(...) 方法，杀死作业对应的 Mesos 任务们，适用常驻作业。Elastic-Job-Cloud-Scheduler 会接收到 Mesos 杀死任务的请求，调用 TaskExecutor#killTask(...)方法，停止任务调度。实现代码如下：
// TaskExecutor.java public final class TaskExecutor implements Executor {@Override public void killTask(final ExecutorDriver executorDriver, final Protos.TaskID taskID) { // 更新 Mesos 任务状态，已杀死。 executorDriver.sendStatusUpdate(Protos.TaskStatus.newBuilder().setTaskId(taskID).setState(Protos.TaskState.TASK_KILLED).build()); // 关闭该 Mesos 任务的调度 DaemonTaskScheduler.shutdown(taskID); } } // DaemonTaskScheduler.java /** * 停止任务调度. * * @param taskID 任务主键 */ public static void shutdown(final Protos.TaskID taskID) { // 移除任务的 Quartz Scheduler Scheduler scheduler = RUNNING_SCHEDULERS.remove(taskID.getValue()); // 关闭任务的 Quartz Scheduler if (null != scheduler) { try { scheduler.shutdown(); } catch (final SchedulerException ex) { throw new JobSystemException(ex); } } }
调用 RunningService#remove(…) 方法，将作业从运行时队列删除。
调用 ReadyService#remove(…) 方法，将作业从待运行队列删除。
调用 TransientProducerScheduler#deregister(...) 方法，停止作业调度，适用瞬时作业。实现代码如下：
synchronized void deregister(final CloudJobConfiguration jobConfig) { // 移除作业 repository.remove(jobConfig.getJobName()); // 若 cron 不再对应有作业调度，移除 Quartz Scheduler 对 cron 对应的 Quartz Job String cron = jobConfig.getTypeConfig().getCoreConfig().getCron(); if (!repository.containsKey(buildJobKey(cron))) { try { scheduler.unscheduleJob(TriggerKey.triggerKey(cron)); } catch (final SchedulerException ex) { throw new JobSystemException(ex); } } }

2.3 启动云作业

调用 CloudJobRestfulApi#enable(...) 方法，启用云作业，实现代码如下：

@DELETE
@Path("/{jobName}/disable")
public void enable(@PathParam("jobName") final String jobName) throws JSONException {
   Optional<CloudJobConfiguration> configOptional = configService.load(jobName);
   if (configOptional.isPresent()) {
       // 将作业移出禁用队列
       facadeService.enableJob(jobName);
       // 重新调度作业
       producerManager.reschedule(jobName);
   }
}

调用 FacadeService#enableJob(...) 方法，将作业移出禁用作业队列。实现代码如下：
// FacadeService.java public void enableJob(final String jobName) { disableJobService.remove(jobName); } // DisableJobService.java public void remove(final String jobName) { regCenter.remove(DisableJobNode.getDisableJobNodePath(jobName)); }
调用 ProducerManager#reschedule(...) 方法，将作业重新调度。实现代码如下：
public void reschedule(final String jobName) { // 停止调度作业 unschedule(jobName); // 调度作业 Optional<CloudJobConfiguration> jobConfig = configService.load(jobName); if (jobConfig.isPresent()) { schedule(jobConfig.get()); } }

调用 #unschedule(…) 方法，停止调度作业。
调用 #schedule(…) 方法，调度作业，在《Elastic-Job-Cloud 源码分析 —— 作业调度（一）》有详细解析。

2.4 更新云作业配置

调用 CloudJobRestfulApi#update(...) 方法，更新云作业配置，实现代码如下：

// CloudJobRestfulApi.java
@PUT
@Path("/update")
@Consumes(MediaType.APPLICATION_JSON)
public void update(final CloudJobConfiguration jobConfig) {
   producerManager.update(jobConfig);
}

// ProducerManager.java
public void update(final CloudJobConfiguration jobConfig) {
   Optional<CloudJobConfiguration> jobConfigFromZk = configService.load(jobConfig.getJobName());
   if (!jobConfigFromZk.isPresent()) {
       throw new JobConfigurationException("Cannot found job '%s', please register first.", jobConfig.getJobName());
   }
   // 修改云作业配置
   configService.update(jobConfig);
   // 重新调度作业
   reschedule(jobConfig.getJobName());
}

调用 ConfigService#update(jobConfig) 方法，修改云作业配置。实现代码如下：
public void update(final CloudJobConfiguration jobConfig) { regCenter.update(CloudJobConfigurationNode.getRootNodePath(jobConfig.getJobName()), CloudJobConfigurationGsonFactory.toJson(jobConfig)); }
调用 #reschedule(…) 方法，重新调度作业。此处的调用是重复的，实际只需 CloudJobConfigurationListener 监听到配置变化，调用 #reschedule(…) 方法即可。

存储在注册中心( Zookeeper )的云作业配置被更新时，云作业配置变更监听( CloudJobConfigurationListener )会监听到，并执行更新相应逻辑，实现代码如下：

public final class CloudJobConfigurationListener implements TreeCacheListener {

    @Override
    public void childEvent(final CuratorFramework client, final TreeCacheEvent event) throws Exception {
        String path = null == event.getData() ? "" : event.getData().getPath();
        if (isJobConfigNode(event, path, Type.NODE_ADDED)) {
            // .... 省略无关代码
        } else if (isJobConfigNode(event, path, Type.NODE_UPDATED)) {
            CloudJobConfiguration jobConfig = getJobConfig(event);
            if (null == jobConfig) {
                return;
            }
            // 从待执行队列中删除相关作业
            if (CloudJobExecutionType.DAEMON == jobConfig.getJobExecutionType()) {
                readyService.remove(Collections.singletonList(jobConfig.getJobName()));
            }
            // 设置禁用错过重执行
            if (!jobConfig.getTypeConfig().getCoreConfig().isMisfire()) {
                readyService.setMisfireDisabled(jobConfig.getJobName());
            }
            // 重新调度作业
            producerManager.reschedule(jobConfig.getJobName());
        } else if (isJobConfigNode(event, path, Type.NODE_REMOVED)) {
            // .... 省略无关代码
        }
    }

}

CloudJobConfigurationListener 实现 TreeCacheListener 实现对 Zookeeper 数据变更的监听。对 TreeCacheListener 感兴趣的同学，可以查看 Apache Curator 相关知识。
调用 ReadyService#remove(…) 方法，将作业从待运行队列删除。TODO，为啥要删除？
调用 ReadyService#setMisfireDisabled(...) 方法，设置禁用错过重执行。实现代码如下：
public void setMisfireDisabled(final String jobName) { Optional<CloudJobConfiguration> cloudJobConfig = configService.load(jobName); if (cloudJobConfig.isPresent() && null != regCenter.getDirectly(ReadyNode.getReadyJobNodePath(jobName))) { regCenter.persist(ReadyNode.getReadyJobNodePath(jobName), "1"); } }

瞬时作业开启 misfire 功能时，当任务执行过久触发 misifre 会不断累积待执行次数。如果关闭 misfire 功能，需要将多次执行次数归 "1"。

调用 ProducerManager#reschedule(…) 方法，重新调度作业。

2.5 注销云作业

调用 CloudJobRestfulApi#deregister(...) 方法，注销云作业，实现代码如下：

// CloudJobRestfulApi.java
@DELETE
@Path("/deregister")
@Consumes(MediaType.APPLICATION_JSON)
public void deregister(final String jobName) {
   producerManager.deregister(jobName);
}

// ProducerManager.java
public void deregister(final String jobName) {
   Optional<CloudJobConfiguration> jobConfig = configService.load(jobName);
   if (jobConfig.isPresent()) {
       //  从作业禁用队列中删除作业
       disableJobService.remove(jobName);
       // 删除云作业
       configService.remove(jobName);
   }
   // 停止调度作业
   unschedule(jobName);
}

调用 DisableJobService#remove(...) 方法，从作业禁用队列中删除作业，实现代码如下：
public void remove(final String jobName) { regCenter.remove(DisableJobNode.getDisableJobNodePath(jobName)); }
调用 CloudJobConfigurationService#remove(...) 方法，删除云作业配置。
调用 #reschedule(…) 方法，重新调度作业。

存储在注册中心( Zookeeper )的云作业配置被删除时，云作业配置变更监听器( CloudJobConfigurationListener )会监听到，并执行删除相应逻辑，实现代码如下：

public final class CloudJobConfigurationListener implements TreeCacheListener {

    private final CoordinatorRegistryCenter regCenter;

    private final ProducerManager producerManager;

    private final ReadyService readyService;

    public CloudJobConfigurationListener(final CoordinatorRegistryCenter regCenter, final ProducerManager producerManager) {
        this.regCenter = regCenter;
        readyService = new ReadyService(regCenter);
        this.producerManager = producerManager;
    }

    @Override
    public void childEvent(final CuratorFramework client, final TreeCacheEvent event) throws Exception {
        String path = null == event.getData() ? "" : event.getData().getPath();
        if (isJobConfigNode(event, path, Type.NODE_ADDED)) {
            // ... 省略无关代码
        } else if (isJobConfigNode(event, path, Type.NODE_UPDATED)) {
            // ... 省略无关代码
        } else if (isJobConfigNode(event, path, Type.NODE_REMOVED)) {
            String jobName = path.substring(CloudJobConfigurationNode.ROOT.length() + 1, path.length());
            producerManager.unschedule(jobName);
        }
    }
}

调用 #reschedule(…) 方法，重新调度作业。

2.6 触发一次云作业

调用 CloudJobRestfulApi#trigger(...) 方法，触发一次云作业，实现代码如下：

@POST
@Path("/trigger")
@Consumes(MediaType.APPLICATION_JSON)
public void trigger(final String jobName) {
   // 常驻作业不允许触发一次作业
   Optional<CloudJobConfiguration> config = configService.load(jobName);
   if (config.isPresent() && CloudJobExecutionType.DAEMON == config.get().getJobExecutionType()) {
       throw new JobSystemException("Daemon job '%s' cannot support trigger.", jobName);
   }
   // 将瞬时作业放入待执行队列
   facadeService.addTransient(jobName);
}

目前常驻作业不支持触发一次云作业。如果想实现该功能，需要 Elastic-Job-Cloud-Scheduler 通过 Mesos 发送自定义消息，通知 Elastic-Job-Cloud-Executor 触发该作业对应的任务们。
调用 FacadeService#addTransient(…) 方法，将瞬时作业放入待执行队列。当且仅当云作业配置 JobCoreConfiguration.misfire = true 时，该作业在待执行队列的执行次数不断累积加一。在《Elastic-Job-Cloud 源码分析 —— 作业调度（一）》「3.2.3 ProducerJob」有详细解析。

3. 云作业应用操作

我们可以使用运维平台或 Restful API 对云作业应用进行操作。前者是对后者的界面包装，如下图所示：

3.1 注册云作业应用

《Elastic-Job-Cloud 源码分析 —— 作业配置》「2.2 操作云作业App配置」有详细解析。

3.2 更新云作业应用配置

调用 CloudAppRestfulApi#update(...) 方法，更新云作业应用配置，实现代码如下：

// CloudAppRestfulApi.java
@PUT
@Consumes(MediaType.APPLICATION_JSON)
public void update(final CloudAppConfiguration appConfig) {
   appConfigService.update(appConfig);
}

// CloudAppConfigurationService.java
public void update(final CloudAppConfiguration appConfig) {
   regCenter.update(CloudAppConfigurationNode.getRootNodePath(appConfig.getAppName()), CloudAppConfigurationGsonFactory.toJson(appConfig));
}

3.3 禁用云作业应用

调用 CloudAppRestfulApi#disable(...) 方法，禁用云作业应用，实现代码如下：

@POST
@Path("/{appName}/disable")
public void disable(@PathParam("appName") final String appName) {
   if (appConfigService.load(appName).isPresent()) {
       // 将应用放入禁用队列
       disableAppService.add(appName);
       // 停止应用对应所有作业的调度
       for (CloudJobConfiguration each : jobConfigService.loadAll()) {
           if (appName.equals(each.getAppName())) {
               producerManager.unschedule(each.getJobName());
           }
       }
   }
}

调用 DisableAppService#add(...) 方法，将应用放入禁用应用队列，实现代码如下：
// DisableAppService.java public class DisableAppService {public void add(final String appName) { if (regCenter.getNumChildren(DisableAppNode.ROOT) > env.getFrameworkConfiguration().getJobStateQueueSize()) { log.warn("Cannot add disable app, caused by read state queue size is larger than {}.", env.getFrameworkConfiguration().getJobStateQueueSize()); return; } String disableAppNodePath = DisableAppNode.getDisableAppNodePath(appName); if (!regCenter.isExisted(disableAppNodePath)) { regCenter.persist(disableAppNodePath, appName); } } } // DisableAppNode.java final class DisableAppNode {static final String ROOT = StateNode.ROOT + "/disable/app"; private static final String DISABLE_APP = ROOT + "/%s"; // %s = ${APP_NAME} }

DisableAppService，禁用应用队列服务。
禁用应用队列存储在注册中心( Zookeeper )的持久数据节点 /${NAMESPACE}/state/disable/app/${APP_NAME}，存储值为应用名。使用 zkClient 查看如下：
[zk: localhost:2181(CONNECTED) 6] ls /elastic-job-cloud/state/disable/app [example_app]

遍历应用对应所有作业，调用 ProducerManager#unschedule(…) 方法，停止作业调度。

3.4 启用云作业应用

调用 CloudAppRestfulApi#enable(...) 方法，启用云作业应用，实现代码如下：

// CloudAppRestfulApi.java
@DELETE
@Path("/{appName}/disable")
public void enable(@PathParam("appName") final String appName) throws JSONException {
   if (appConfigService.load(appName).isPresent()) {
       // 从禁用应用队列中删除应用
       disableAppService.remove(appName);
       // 重新开启应用对应所有作业的调度
       for (CloudJobConfiguration each : jobConfigService.loadAll()) {
           if (appName.equals(each.getAppName())) {
               producerManager.reschedule(each.getJobName());
           }
       }
   }
}

// DisableAppService.java
public void remove(final String appName) {
   regCenter.remove(DisableAppNode.getDisableAppNodePath(appName));
}

3.5 注销云作业应用

调用 CloudAppRestfulApi#deregister(...) 方法，注销云作业应用，实现代码如下：

@DELETE
@Path("/{appName}")
@Consumes(MediaType.APPLICATION_JSON)
public void deregister(@PathParam("appName") final String appName) {
   if (appConfigService.load(appName).isPresent()) {
       // 移除应用和应用对应所有作业的配置
       removeAppAndJobConfigurations(appName);
       // 停止应用对应的执行器( Elastic-Job-Cloud-Scheduler )
       stopExecutors(appName);
   }
}

调用 #removeAppAndJobConfigurations(...) 方法，移除应用和应用对应所有作业的配置，实现代码如下：
private void removeAppAndJobConfigurations(final String appName) { // 注销(移除)应用对应所有作业的配置 for (CloudJobConfiguration each : jobConfigService.loadAll()) { if (appName.equals(each.getAppName())) { producerManager.deregister(each.getJobName()); } } // 从禁用应用队列中删除应用 disableAppService.remove(appName); // 删除云作业App配置 appConfigService.remove(appName); }
调用 #stopExecutors(...) 方法，停止应用对应的执行器( Elastic-Job-Cloud-Scheduler )，实现代码如下：
// CloudAppRestfulApi.java private void stopExecutors(final String appName) { try { Collection<ExecutorStateInfo> executorBriefInfo = mesosStateService.executors(appName); for (ExecutorStateInfo each : executorBriefInfo) { producerManager.sendFrameworkMessage(ExecutorID.newBuilder().setValue(each.getId()).build(), SlaveID.newBuilder().setValue(each.getSlaveId()).build(), "STOP".getBytes()); } } catch (final JSONException ex) { throw new JobSystemException(ex); } } // ProducerManager.java public void sendFrameworkMessage(final ExecutorID executorId, final SlaveID slaveId, final byte[] data) { schedulerDriver.sendFrameworkMessage(executorId, slaveId, data); }@Override public void frameworkMessage(final ExecutorDriver executorDriver, final byte[] bytes) { if (null != bytes && "STOP".equals(new String(bytes))) { log.error("call frameworkMessage executor stopped."); executorDriver.stop(); } }

调用 SchedulerDriver#sendFrameworkMessage(…) 方法，通过 Mesos 向云作业应用对应的执行器们( Elastic-Job-Cloud-Executor ) 发送消息为 "STOP" 从而关闭执行器。Elastic-Job-Cloud-Executor 会接收到 Mesos 消息，调用 TaskExecutor#frameworkMessage(…) 方法，关闭自己。实现代码如下：

欢迎加入我的知识星球，一起探讨架构，交流源码。加入方式，长按下方二维码噢：

已在知识星球更新源码解析如下：

人若无名

专心练剑

2019走你

好看吗？↓

高三女生醉酒后被强奸致死？检方回应

常德悲剧：让谴责无差别杀戮之声更加响亮一点

2024【公共营养师】培训报名通道已开启，不限学历，23岁及以上可报！还能领2000补贴

【惊】"以为要写遗书"! 飞温哥华航班遇炸弹气旋华人崩溃大哭连空姐都吐了; 客机颠簸盘旋3小时

魏加宁：日本之所以能走出“大衰退”，靠的是不断改革，而不是所谓“积极的财政政策”

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

1. 概述

2. 云作业操作

2.1 注册云作业配置

2.2 禁用云作业

2.3 启动云作业

2.4 更新云作业配置

2.5 注销云作业

2.6 触发一次云作业

3. 云作业应用操作

3.1 注册云作业应用

3.2 更新云作业应用配置

3.3 禁用云作业应用

3.4 启用云作业应用

3.5 注销云作业应用

您可能也对以下帖子感兴趣

高三女生醉酒后被强奸致死？检方回应

常德悲剧：让谴责无差别杀戮之声更加响亮一点

2024【公共营养师】培训报名通道已开启，不限学历，23岁及以上可报！还能领2000补贴

【惊】"以为要写遗书"! 飞温哥华航班遇炸弹气旋 华人崩溃大哭 连空姐都吐了; 客机颠簸盘旋3小时

魏加宁：日本之所以能走出“大衰退”，靠的是不断改革，而不是所谓“积极的财政政策”

生成图片，分享到微信朋友圈

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

1. 概述

2. 云作业操作

2.1 注册云作业配置

2.2 禁用云作业

2.3 启动云作业

2.4 更新云作业配置

2.5 注销云作业

2.6 触发一次云作业

3. 云作业应用操作

3.1 注册云作业应用

3.2 更新云作业应用配置

3.3 禁用云作业应用

3.4 启用云作业应用

3.5 注销云作业应用

您可能也对以下帖子感兴趣

【惊】"以为要写遗书"! 飞温哥华航班遇炸弹气旋华人崩溃大哭连空姐都吐了; 客机颠簸盘旋3小时