ElasticJob - 任务执行过程解析

ElasticJob是基于Quartz实现的弹性分布式任务调度框架，整个任务触发的底层是Quartz来触发。LiteJob框架触发任务执行的开始，下面来分析任务的执行过程。

public final class LiteJob implements Job {@Overridepublic void execute(final JobExecutionContext context) throws JobExecutionException {//根据任务类型获取执行器，SimpleJobExecutorJobExecutorFactory.getJobExecutor(elasticJob, jobFacade).execute();}
}

ElasticJob根据任务的作业类型，获取不同的作业执行器，这里以SimpleJobExecutor为例。它继承了AbstractElasticJobExecutor，它是所有任务执行流程的模板类。

protected AbstractElasticJobExecutor(final JobFacade jobFacade) {this.jobFacade = jobFacade;jobRootConfig = jobFacade.loadJobRootConfiguration(true);jobName = jobRootConfig.getTypeConfig().getCoreConfig().getJobName();executorService = ExecutorServiceHandlerRegistry.getExecutorServiceHandler(jobName, (ExecutorServiceHandler) getHandler(JobProperties.JobPropertiesEnum.EXECUTOR_SERVICE_HANDLER));jobExceptionHandler = (JobExceptionHandler) getHandler(JobProperties.JobPropertiesEnum.JOB_EXCEPTION_HANDLER);itemErrorMessages = new ConcurrentHashMap<>(jobRootConfig.getTypeConfig().getCoreConfig().getShardingTotalCount(), 1);}

在构造方法里面，是根据JobName为每个Job实例初始化了一个任务执行线程池，就是一个普通的ThreadPoolExecutor封装。下面分析下任务执行的流程：

 public void execute() {try {//TODO 检验作业服务器和zk服务器系统时差是否可忍受，默认是不开启检查的jobFacade.checkJobExecutionEnvironment();} catch (final JobExecutionEnvironmentException cause) {jobErrorHandler.handleException(jobConfig.getJobName(), cause);}//TODO 获取分片的上下文信息ShardingContexts shardingContexts = jobFacade.getShardingContexts();jobFacade.postJobStatusTraceEvent(shardingContexts.getTaskId(), State.TASK_STAGING, String.format("Job '%s' execute begin.", jobConfig.getJobName()));//TODO 查看当前任务是否有分片正在执行，就创建节点/sharding/{item}/misfireif (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 {//TODO 任务开始执行，这里会去执行业务配置的监听器ElasticJobListenerjobFacade.beforeJobExecuted(shardingContexts);//CHECKSTYLE:OFF} catch (final Throwable cause) {//CHECKSTYLE:ONjobErrorHandler.handleException(jobConfig.getJobName(), cause);}//TODO 任务的执行处理execute(shardingContexts, ExecutionSource.NORMAL_TRIGGER);//TODO 任务是否有错过执行，重新出发调度一次while (jobFacade.isExecuteMisfired(shardingContexts.getShardingItemParameters().keySet())) {jobFacade.clearMisfire(shardingContexts.getShardingItemParameters().keySet());execute(shardingContexts, ExecutionSource.MISFIRE);}//TODO 执行故障转移，防止未及时故障转移吧，其实有个listener在监控，为什么这里还要手动执行下 故障转移jobFacade.failoverIfNecessary();try {//TODO 任务执行完成，这里是去触发业务配置的监听器的执行jobFacade.afterJobExecuted(shardingContexts);//CHECKSTYLE:OFF} catch (final Throwable cause) {//CHECKSTYLE:ONjobErrorHandler.handleException(jobConfig.getJobName(), cause);}}

Job节点服务器和zookeeper时钟差校验，对应的是Job的maxTimeDiffSeconds最大始终差配置，默认是-1不检查

try {//TODO 检验作业服务器和zk服务器系统时差是否可忍受，默认是不开启检查的jobFacade.checkJobExecutionEnvironment();} catch (final JobExecutionEnvironmentException cause) {jobErrorHandler.handleException(jobConfig.getJobName(), cause);}public void checkMaxTimeDiffSecondsTolerable() throws JobExecutionEnvironmentException {int maxTimeDiffSeconds = load(true).getMaxTimeDiffSeconds();if (-1 == maxTimeDiffSeconds) {return;}long timeDiff = Math.abs(timeService.getCurrentMillis() - jobNodeStorage.getRegistryCenterTime());if (timeDiff > maxTimeDiffSeconds * 1000L) {throw new JobExecutionEnvironmentException("Time different between job server and register center exceed '%s' seconds, max time different is '%s' seconds.", timeDiff / 1000, maxTimeDiffSeconds);}}

获取当前节点服务器的分片信息，这里是分片的关键，在分片那节再细说，这里大概说下做的事情。如果Job配置了故障转移并且存在故障转移分片，优先执行从其它故障节点转移到当前节点服务器的分片。如果需要重新分片，则执行重新分片逻辑，获取分配到当前节点的分片信息。

public ShardingContexts getShardingContexts() {boolean isFailover = configService.load(true).isFailover();if (isFailover) {//TODO 获取故障转移到当前节点的分片信息List<Integer> failoverShardingItems = failoverService.getLocalFailoverItems();if (!failoverShardingItems.isEmpty()) {//TODO 获取故障转移分片信息return executionContextService.getJobShardingContext(failoverShardingItems);}}shardingService.shardingIfNecessary();List<Integer> shardingItems = shardingService.getLocalShardingItems();if (isFailover) {//TODO 删除本节点被故障转移的分片信息shardingItems.removeAll(failoverService.getLocalTakeOffItems());}shardingItems.removeAll(executionService.getDisabledItems(shardingItems));return executionContextService.getJobShardingContext(shardingItems);}

当前节点服务器该job对应的所有分片，是否有执行中的分片， 只要有一个执行中的分片，则对应的所有分片都会被设置为misfire错失执行，然后退出。 就是上次触发的执行没有完成，本次触发不会执行，会等待上次执行完成，防止任务分片在当前节点被重复执行。

//TODO 查看当前任务是否有分片正在执行，就创建节点/sharding/{item}/misfire

  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;}public boolean misfireIfHasRunningItems(final Collection<Integer> items) {if (!hasRunningItems(items)) {return false;}//TODO 只要其中一个分片没有执行完成，就会设置misfiresetMisfire(items);return true;}public boolean hasRunningItems(final Collection<Integer> items) {JobConfiguration jobConfig = configService.load(true);if (!jobConfig.isMonitorExecution()) {return false;}for (int each : items) {//TODO 只要其中一个分片没有执行完成，就会设置misfireif (jobNodeStorage.isJobNodeExisted(ShardingNode.getRunningNode(each))) {return true;}}return false;}

执行任务之前对应的ElasticJobListener，这个是之前在配置Job任务时对应的ElasticJobListener。


try {//TODO 任务开始执行，这里会去执行业务配置的监听器ElasticJobListenerjobFacade.beforeJobExecuted(shardingContexts);//CHECKSTYLE:OFF} catch (final Throwable cause) {//CHECKSTYLE:ONjobErrorHandler.handleException(jobConfig.getJobName(), cause);}

具体任务分片的调度执行，在分片任务执行前，在当前节点服务器把任务在内存中设置为running状态，如果配置了幂等机制，同时也会在zookeeper中创建/sharding/{item}/running临时节点，在服务器宕机该节点就会自动删除了。然后就是把所有的分片任务提交到ThreadPoolExecutor中执行。这里使用的CountDownLatch控制并发等待所有的分片任务都完成，然后就删除任务执行钱创建的/sharding/{item}/running节点。同时如果开启了故障转移机制，就会删除之前创建的故障转移节点/sharding/{item}/failover

 //TODO 任务的执行处理execute(shardingContexts, ExecutionSource.NORMAL_TRIGGER);private void execute(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;}//TODO 将当前任务设置为运行中Running的状态，分为内存状态和zk状态，在zk中创建一个临时节点jobFacade.registerJobBegin(shardingContexts);String taskId = shardingContexts.getTaskId();jobFacade.postJobStatusTraceEvent(taskId, State.TASK_RUNNING, "");try {process(shardingContexts, executionSource);} finally {// TODO Consider increasing the status of job failure, and how to handle the overall loop of job failure//TODO 清除job的running状态和删除故障转移节点/sharding/{item}/failoverjobFacade.registerJobCompleted(shardingContexts);if (itemErrorMessages.isEmpty()) {jobFacade.postJobStatusTraceEvent(taskId, State.TASK_FINISHED, "");} else {jobFacade.postJobStatusTraceEvent(taskId, State.TASK_ERROR, itemErrorMessages.toString());}}}private void process(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(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);if (executorService.isShutdown()) {return;}executorService.submit(() -> {try {process(shardingContexts, each, jobExecutionEvent);} finally {latch.countDown();}});}try {latch.await();} catch (final InterruptedException ex) {Thread.currentThread().interrupt();}}

检查是否有任务分片错过执行，这里的判断是开启了misfire任务错过机制并且分片存在/sharding/{item}/misfire节点。如果开启了任务错过机制，并且有任务错过执行，这里会重新执行一次分片节点任务。任务执行之前清除了/sharding/{item}/misfire节点。

//TODO 任务是否有错过执行，重新出发调度一次while (jobFacade.isExecuteMisfired(shardingContexts.getShardingItemParameters().keySet())) {jobFacade.clearMisfire(shardingContexts.getShardingItemParameters().keySet());execute(shardingContexts, ExecutionSource.MISFIRE);}

在这里再手动执行下故障转移操作， 在这里不知道为啥还需要手动执行下故障转移，在故障转移监听器里面就能监听到，为啥这里还要在任务执行的时候再手动触发下？？？没太理解

//TODO 执行故障转移，防止未及时故障转移吧，其实有个listener在监控，为什么这里还要手动执行下 故障转移
jobFacade.failoverIfNecessary();

任务分片执行完之后，触发业务给任务配置的监听器的执行

 try {//TODO 任务执行完成，这里是去触发业务配置的监听器的执行jobFacade.afterJobExecuted(shardingContexts);//CHECKSTYLE:OFF} catch (final Throwable cause) {//CHECKSTYLE:ONjobErrorHandler.handleException(jobConfig.getJobName(), cause);}

作业执行流程图:

ElasticJob作业执行流程图

ElasticJob - 任务执行过程解析相关推荐

Fiori elements执行过程解析：When click go in table list, odata service is sent
Created by Wang, Jerry, last modified on Mar 22, 2016Go to start of metadata
mysql查询解析过程_MySQL查询执行过程详解
查询是用户通过设置某些查询条件,从表或其他查询中选取全部或者部分数据,以表的形式显示数据供用户浏览.查询是一个独立的.功能强大的.具有计算功能和条件检索功能的数据库对象.MySQL数据库中,MySQL ...
Web APi之过滤器执行过程原理解析【二】（十一）
前言上一节我们详细讲解了过滤器的创建过程以及粗略的介绍了五种过滤器,用此五种过滤器对实现对执行Action方法各个时期的拦截非常重要.这一节我们简单将讲述在Action方法上.控制器上.全局上以及授 ...
android连接ecs sql server_MySQL 执行 SQL 语句的过程解析
前言作为一名菜鸟程序员,我已经工作两年了.从事后端开发工作. 后端开发中,MySQL 可以说是必会的一门技能了.刚开始时我以为数据库就是写写 SQL 语句,做做增删改查业务.直到我写的程序多次导致 ...
mysql源生的sql语句_Django执行源生mysql语句实现过程解析
1.使用extra方法解释:结果集修改器,一种提供额外查询参数的机制说明:依赖model模型使用方式: 用在where后: Book.objects.filter(publisher_id=&q ...
Spring AOP源码解析-拦截器链的执行过程
一.简介在前面的两篇文章中,分别介绍了 Spring AOP 是如何为目标 bean 筛选合适的通知器,以及如何创建代理对象的过程.现在得到了 bean 的代理对象,且通知也以合适的方式插在了目标方 ...
mysql的优化器执行过程（附代码解析，手动注释等）
一.前言这篇文章的诞生很偶然,优化器大家基本都听过,感觉就是只要自己不能理解的东西,都归于mysql优化器就完事了,哈哈.但是优化器到底是什么呢,执行过程是什么样子的呢?博主是在看一篇博文的时候,看 ...
MyBatis核心源码剖析(SqlSession XML解析 Mapper executor SQL执行过程自定义类型处理器缓存日志)
MyBatis核心源码剖析 MyBatis核心源码剖析 1 MyBatis源码概述 1.1 为什么要看MyBatis框架的源码 1.2 如何深入学习MyBatis源码 1.3 源码分析的5大原则 2 ...
maven scala plugin 实现jvmArgs,执行过程原理解析笔记
scala-maven-plugin 构建scala,springboot工程我们在使用maven构建scala工程的时候, 要使用scala-maven-plugin插件,常规配置如下 : < ...
openstack nova 源码解析 — Nova API 执行过程从(novaclient到Action)
目录目录 Nova API Nova API 的执行过程 novaclient 将 Commands 转换为标准的HTTP请求 PasteDeploy 将 HTTP 请求路由到具体的 WSGI Ap ...

ElasticJob - 任务执行过程解析

作业执行流程图:

ElasticJob - 任务执行过程解析相关推荐

最新文章

热门文章