查看原文
其他

分布式作业系统 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) &gt; 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走你

好看吗?↓

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

    文章有问题?点此查看未经处理的缓存