分布式系统中的定时任务全解

Posted 2020-07-27 不去天涯

概述

分布式系统中的定时任务全解（一）–java基础实现
分布式系统中的定时任务全解（二）–常见工程实现
分布式系统中的定时任务全解（四）–补充

前两篇从java语言中定时任务的基础实现，到第三方框架依赖下的常用实现方式都已经讲到了。

接下来的这一节会比较长，会从elastic-job使用、使用中会遇到的问题以及elastic-job的几个切面上的原理3个大部分。

集成elastic-job

1.首先引入maven仓库

<!-- 引入elastic-job核心模块 -->
<dependency>
    <groupId>com.dangdang</groupId>
    <artifactId>elastic-job-core</artifactId>
    <version>1.1.0</version>
</dependency>

<!-- 使用springframework自定义命名空间时引入 -->
<dependency>
    <groupId>com.dangdang</groupId>
    <artifactId>elastic-job-spring</artifactId>
    <version>1.1.0</version>
</dependency>

2.实现自己定义的作业

@Component
public class MyElasticJob extends AbstractSimpleElasticJob {

    @Override
    public void process(JobExecutionMultipleShardingContext context) {
        // do something by sharding items
    }
}

3.配置作业

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
    xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
    xmlns:context="http://www.springframework.org/schema/context" 
    xmlns:reg="http://www.dangdang.com/schema/ddframe/reg" 
    xmlns:job="http://www.dangdang.com/schema/ddframe/job" 
    xsi:schemaLocation="http://www.springframework.org/schema/beans 
                        http://www.springframework.org/schema/beans/spring-beans.xsd 
                        http://www.springframework.org/schema/context 
                        http://www.springframework.org/schema/context/spring-context.xsd 
                        http://www.dangdang.com/schema/ddframe/reg 
                        http://www.dangdang.com/schema/ddframe/reg/reg.xsd 
                        http://www.dangdang.com/schema/ddframe/job 
                        http://www.dangdang.com/schema/ddframe/job/job.xsd 
                        ">
    <context:component-scan base-package="com.dangdang.example.elasticjob" />
    <context:property-placeholder location="classpath:conf/*.properties" />

    <reg:zookeeper id="regCenter" server-lists="${serverLists}" namespace="${namespace}" base-sleep-time-milliseconds="${baseSleepTimeMilliseconds}" max-sleep-time-milliseconds="${maxSleepTimeMilliseconds}" max-retries="${maxRetries}" nested-port="${nestedPort}" nested-data-dir="${nestedDataDir}" />

    <job:simple id="simpleElasticJob" class="com.dangdang.example.elasticjob.spring.job.SimpleJobDemo" registry-center-ref="regCenter" sharding-total-count="${simpleJob.shardingTotalCount}" cron="${simpleJob.cron}" sharding-item-parameters="${simpleJob.shardingItemParameters}" monitor-execution="${simpleJob.monitorExecution}" monitor-port="${simpleJob.monitorPort}" failover="${simpleJob.failover}" description="${simpleJob.description}" disabled="${simpleJob.disabled}" overwrite="${simpleJob.overwrite}" />
    <job:dataflow id="throughputDataFlowJob" class="com.dangdang.example.elasticjob.spring.job.ThroughputDataFlowJobDemo" registry-center-ref="regCenter" sharding-total-count="${throughputDataFlowJob.shardingTotalCount}" cron="${throughputDataFlowJob.cron}" sharding-item-parameters="${throughputDataFlowJob.shardingItemParameters}" monitor-execution="${throughputDataFlowJob.monitorExecution}" failover="${throughputDataFlowJob.failover}" process-count-interval-seconds="${throughputDataFlowJob.processCountIntervalSeconds}" concurrent-data-process-thread-count="${throughputDataFlowJob.concurrentDataProcessThreadCount}" description="${throughputDataFlowJob.description}" disabled="${throughputDataFlowJob.disabled}" overwrite="${throughputDataFlowJob.overwrite}" streaming-process="${throughputDataFlowJob.streamingProcess}" />
    <job:dataflow id="sequenceDataFlowJob" class="com.dangdang.example.elasticjob.spring.job.SequenceDataFlowJobDemo" registry-center-ref="regCenter" sharding-total-count="${sequenceDataFlowJob.shardingTotalCount}" cron="${sequenceDataFlowJob.cron}" sharding-item-parameters="${sequenceDataFlowJob.shardingItemParameters}" monitor-execution="${sequenceDataFlowJob.monitorExecution}" failover="${sequenceDataFlowJob.failover}" process-count-interval-seconds="${sequenceDataFlowJob.processCountIntervalSeconds}" max-time-diff-seconds="${sequenceDataFlowJob.maxTimeDiffSeconds}" description="${sequenceDataFlowJob.description}" disabled="${sequenceDataFlowJob.disabled}" overwrite="${sequenceDataFlowJob.overwrite}" />
</beans>

属性文件定义:

#job.properties

simpleJob.cron=0/5 * * * * ?
simpleJob.shardingTotalCount=10
simpleJob.shardingItemParameters=0=A,1=B,2=C,3=D,4=E,5=F,6=G,7=H,8=I,9=J
simpleJob.monitorExecution=false
simpleJob.failover=true
simpleJob.description=\\u53EA\\u8FD0\\u884C\\u4E00\\u6B21\\u7684\\u4F5C\\u4E1A\\u793A\\u4F8B
simpleJob.disabled=false
simpleJob.overwrite=true
simpleJob.monitorPort=9888

throughputDataFlowJob.cron=0/5 * * * * ?
throughputDataFlowJob.shardingTotalCount=10
throughputDataFlowJob.shardingItemParameters=0=A,1=B,2=C,3=D,4=E,5=F,6=G,7=H,8=I,9=J
throughputDataFlowJob.monitorExecution=true
throughputDataFlowJob.failover=true
throughputDataFlowJob.processCountIntervalSeconds=10
throughputDataFlowJob.concurrentDataProcessThreadCount=3
throughputDataFlowJob.description=\\u4E0D\\u505C\\u6B62\\u8FD0\\u884C\\u7684\\u4F5C\\u4E1A\\u793A\\u4F8B
throughputDataFlowJob.disabled=false
throughputDataFlowJob.overwrite=true
throughputDataFlowJob.streamingProcess=true

sequenceDataFlowJob.cron=0/5 * * * * ?
sequenceDataFlowJob.shardingTotalCount=10
sequenceDataFlowJob.shardingItemParameters=0=A,1=B,2=C,3=D,4=E,5=F,6=G,7=H,8=I,9=J
sequenceDataFlowJob.maxTimeDiffSeconds=-1
sequenceDataFlowJob.monitorExecution=true
sequenceDataFlowJob.failover=true
sequenceDataFlowJob.processCountIntervalSeconds=10
sequenceDataFlowJob.description=\\u6309\\u987A\\u5E8F\\u4E0D\\u505C\\u6B62\\u8FD0\\u884C\\u7684\\u4F5C\\u4E1A\\u793A\\u4F8B
sequenceDataFlowJob.disabled=false
sequenceDataFlowJob.overwrite=true

#reg.properties

serverLists=localhost:4181
namespace=elasticjob-example
baseSleepTimeMilliseconds=1000
maxSleepTimeMilliseconds=3000
maxRetries=3

nestedPort=4181
nestedDataDir=target/test_zk_data/

集成中遇到的问题

1.cron表达式总是和第一次运行时配置的一样，不变

因为，不论是elastic-job的github给出的示例，还是网上给的示例，配置项都是没有添加overwrite选项的，这个选项默认是false，也就是任务的配置信息，如果已经设置过，那么就会一直不变，就算后续你修改了自己的配置文件中的cron。

解决方法就是在你的job配置中加上overwrite选项：

<job:simple id="yourTaskId" class="yourTaskClass" registry-center-ref="regCenter" cron="0 0/30 * * * ?"   sharding-total-count="1" sharding-item-parameters="0=A" overwrite="true"/>

2.我在网上找到的例子，给的serverLists=” yourhost:2181”，为什么编译器告诉我serverLists配置项不被支持

网上很多示例都是针对1.1.0版本之前的示例，1.1.0版本elastic-job进行了很大的改动，包括一些属性。

解决方法就是，使用1.1.0版本以后的elastic-job，按照github官网给出的示例去做（https://github.com/dangdangdotcom/elastic-job）

3.我的job里面的autowired或者resource变量没有注入

解决方法，首先看你的变量是不是静态的，如果是静态的请换成非静态的，这时spring的问题。接下来看下你的job有没有添加@component注解。

4.我是springmvc的web工程，并且已经在其他xml文件有placeholder了，但是reg:zookeeper初始化时显示的连接仍然是“${xxx}”的样式

原理作者也有弄清楚（后续可以研究一下配置文件的加载过程）。如果elastic-job是在单独的xml中配置，那么需要在这个xml中添加placeholder，但是，你一定知道spring默认仅加载一个placeholder，那么只需要在placeholder属性中添加ignore-unresolvable=”true”即可。

5.如果我需要在job中重新设定下次触发的时间怎么办

在你的job中结束位置，添加如下代码：

JobRegistry.getInstance().getJobScheduleController（jobname）.rescheduleJob(cron)；

如果，你打印了日志，那么应该会发现，上面的语句调用后，job会被立即触发，看起来像是同时执行了两次。这是触发时间点计算的缘故，cron表达式是以s为单位的，计算机的执行是以毫秒为单位的，很可能当前的时间点仍然是你给定新cron表达式相符合的时间点。

给了具体的例子：

原有cron=”0/10 * * * * * ? ”
触发时间点是：9点18分10s
函数执行时间是：100ms
执行的最后调用：reschedule
新的cron=”0/5 * * * * * ?”
那么此时仍然符合新cron的触发时间点，所以reschedule会立即触发

这是无法避免的，请保证你的job的幂等性。

elastic-job不同侧面解析

这里首先给出elastic-job主要设计师之一张亮的一篇博文地址，这里给出了很多elastic-job的机理层面的解析。（http://my.oschina.net/u/719192/blog/506062）

实现思想对比

1.先说下map/reduce的思想，这个看起来和定时任务没有关系，列在这里也正是因为两者之间算是完全不同的两种思想。

这里要特别指出的是，所有的计算节点，都是被动的接受任务，头结点给你什么任务，你就执行什么任务。

2.分布式定时任务（quartz/elastic-job）
接下来看一下定时任务的集群方案，是完全的一个翻转：

所有的任务执行节点上的调度器都在运行，他们执行不执行某个任务，是根据从协调中心获取到的数据判断的。quartz是看数据库记录，elastic-job是看zookeeper中的sharding信息。

3.另外一种思路？
可能你会想，为什么定时任务不能是单独的一个集群，然后可以通过管理端随时上传定时任务的jar上去，然后又头结点调度呢？这样的话不是所有的定时任务都能集中、统一管理起来了么？那该多好，和部署的服务器独立。我只能说，市面上是有这种类型的服务的，比如阿里提供的E-MapReduce服务，前端支持手动触发和定时触发作业，后端的作业处理使用的是Hadoop。

最后，说明一下，既然知道了分布式定时任务的思想，那么也就很容易理解他的部署方式了：

也就是直接和你的web服务在一起，每个服务器实例都是一个计算节点，连接到协调中心（数据库/zookeeper），定时任务触发时从协调中心查询自己是应当执行job任务，还是直接返回，跳过job的执行。

接下来着重从源代码层面的不同侧面做一些对elastic-job的粗范的解析，以利于使用elastic-job时遇到问题的快速解决。(2016年10月19日补充：以下分析均针对ej的1.1.0版本，1.1.1之后，已经拆分成了ej-lite和ej-cloud，其中ej-cloud就是博主提到的第三种思想)

任务的初始化过程

初始化的入口在：

new JobScheduler(regCenter, simpleJobConfig, new SimpleDistributeOnceElasticJobListener()).init();

接下来看下init方法到底做了什么事情：

其中最主要的两个步骤是：

1.registerStartUpInfo，这一步，这里面添加了对zookeeper的监听（后续会讲到监听和监听后做了什么），和zookeeper上相关节点的创建。

其中persistJobConfiguration方法中用到了前面问题中说到的overwrite，如果overwrite为false，那么shechule的触发cron表达式是直接从zookeeper中获取的，而不是本地xml配置的。

其中的setReshrdingFlag用来创建一个标记，所有服务器上的同名schedule运行的job都会检查是否存在这个标记（在后续的任务执行中会说到），如果存在，那么执行任务重新分片（后续会说到什么是分片，分片用来干什么，这是elastic-job优于quartz集群的大亮点之一）。

2.sheduleJob
这就是创建quartz的schedule，启动定时任务了。

任务执行

elastic-job任务分片，多种类型任务的封装都在这里，先看下elastic-job和做为elastic-job基础的quartz之间的关系：

看上图，应该能够很清楚，elastic-job所有的关键都在abstractElasticJob.execute方法中，接下来一块看一下：

看到里面的shardingIfNecessary和getShardingItems.isEmpty-return了么？

这就是分片和控制任务仅被应当执行的服务器实例执行的关键了。

分片

看了上边那么多，估计很可能你还是对分片是什么，怎么用，什么时候会触发分片存在很多很多的疑问，接下来一个一个的看。

说到分片这可以说是elastic-job在quartz集群调度上的一个创新。在quartz集群环境下，仅有一个服务器实例可以运行某一个特定的schedule。但是在elastic-job下，你可以指定有几个服务器实例执行这一个任务，可以是1个，2个，3个都可以。

那么几个服务器实例运行同一个任务不是冲突了么，至少是浪费了么？NONONO…..

举个例子，在一个更广的环境下，比如你有一个定时计算用户积分的任务，你的用户表是分在10个数据库里的。那么你可以用一个服务器实例去运行，也可以用5个服务器实例去运行，因为你有5台服务器可用，每个服务器实例分给2个数据库的计算任务。

这种场景用quartz是做不到的，用elastic-job，你只需要指定总分片10，那么每一个分片指定一个标记量，那么每台服务器会得到2个任务去执行。

当然如果你有5台服务器可用来计算，但是数据库仅有2个，那么你只能分两个片，那么这5台中有两台得到运行的权利，其他的服务器将在job.execute中看到自己的getShardingItems.isEmpty然后直接return。

分片在什么场景下会触发呢？

以下场景都会触发重新分片，以下场景是在zookeeper中添加了一个reshard的标记量，任务下次执行的时候就会触发分片。

监听器的作用

前边说过了，在初始化的时候elastic-job会注册一系列的zookeeper监听器，监听节点的变化，那么他具体监听了哪些地方呢？

汇总起来是两个方面：一个是响应console对定时任务的控制，一个是响应服务器的崩溃。当执行的节点崩溃时，会触发重新分片，由其他服务器接起定时任务的执行。