分布式定时任务

分布式定时任务

1，什么是分布式定时任务；

2，为什么要采用分布式定时任务；

3，怎么样设计实现一个分布式定时任务；

4，当前比较流行的分布式定时任务框架；

1，什么是分布式定时任务：

首先，我们要了解计划任务这个概念，计划任务是指由计划的定时运行或者周期性运行的程序。我们最常见的就是Linux的‘crontab’和Windows的‘计划任务’。

那么什么是分布式定时任务，个人总结为：把分散的，可靠性差的计划任务纳入统一的平台，并实现集群管理调度和分布式部署的一种定时任务的管理方式。叫做分布式定时任务。

2，为什么要采用分布式定时任务：

单点定时任务的缺点:

功能相对简单，交互性差，任务部署效率低，开发和维护成本比较高，不能很好的满足各系统定时任务的管理和控制，尤其在多系统的环境下更加明显；

许多任务都是单机部署，可用性差；

任务跟踪和告警难以实现。

分布式定时任务的优势：

通过集群的方式进行管理调度，大大降低了开发和维护成本；

分布式部署，保证了系统的高可用性，伸缩性，负载均衡，提高了容错；

可以通过控制台部署和管理定时任务，方便灵活高效；

任务都可以持久化到数据库，避免了宕机和数据丢失带来的隐患，同时有完善的任务失败重做机制和详细的任务跟踪及告警策略。

3，怎么样设计和实现一个分布式定时任务：

3.1 分时方案

严格划分时间片，交替运行计划任务，当主系统宕机后，备用系统仍然工作，但是处理初期被拉长了。

缺点：周期延长了。

untitled.png

3.2 HA高可用方案：

正常情况下主系统工作，备用系统守候，心跳检测发现主系统出现故障备用系统启动。

缺点：单一系统，不能做负载均衡，只能垂直扩展，也就是硬件层面的升级，无法做水平扩展。

untitled1.png

3.3 多路心跳方案：

采用多路心跳，做服务级，进程级的，IP和端口级别的心跳检测，正常情况是主系统工作，备用系统守候，心跳检测主系统出现故障，备用系统启动，当再次检测到主系统工作，则将执行权交回主系统。

缺点：开发比较复杂，程序健壮性要求高。

untitled2.png

3.4 任务抢占方案：

A，B两台服务器同时工作，启动需要存在一前一后，谁先启动谁率先加锁，其他服务器只能等待，他们同时对互斥锁进行监控，一旦发现锁被释放，其他服务那个先抢到，那个运行，运行前加排他锁。

优点：可以进一步实现多服务器横向扩展。

缺点：开发复杂，程序健壮性要求高，有时候会出现不释放锁的问题。

untitled4.png

3.5 任务轮询或任务轮询+抢占排队方案

每个服务器首次启动时加入队列；

每次任务运行首先判断自己是否是当前可运行任务，如果是便运行；

如果不是当前运行的任务，检查自己是否在队列中，如果在，便推出，如果不在队列中，便键入队列。

untitled5.png

通过以上这些方案，可以看出3.5的方案才是优先选择的，扩展性好，开发复杂度不是很高。那么这种方案需要的需要的技术原理是什么呢，那就是分布式互斥锁和队列。

3.6 原理：

分布式互斥锁：

互斥锁也叫排他锁，用于并发时管理多进程和多进程同一时刻只能有一个进程或者线程操作一个功能。我们将进程，线程中的锁延伸到互联网上，实现对一个节点运行的进程或线程加锁，解锁操作。这样便能控制节点上的进程或线程的并发。如下图：

untitled7.png

有两台服务器运行定时任务，其中serverA的T2做了加锁操作，其他程序必须等它释放锁了才能运行。那么如果serverA在加锁的过程中，出现宕机怎么办，是否会一直处于别锁状态。那么我们可以在每个锁都设置一个超时阈值，一旦超时便自动解锁。这样就不会因为宕机导致锁一直不被释放。另外我们还要考虑命名空间的问题，主要是防止出现同名锁，导致被覆盖。

队列：

在上面的基础上，排队运行任务。

untitled8.png

从上图中可以看出，TaskQueue中排队情况，运行是自上而下的，当然这个顺序可以自己设置规则，只需要先进先出的远程即可。另外，Task Queue我们需要做至少两个节点，他们遵循主从结构的原则，主节点需要实时向从节点同步数据，保证在主节点不可用，从节点可以替代。当然，这里可以使用权重轮询的方式，加上数据异步同步，让所有节点都可以做主从的切换，根据运行状况来分配，可能会更好，但是这样开发难度也有所提高，但是大大增加了高可用性。

3.7 总结：

最后，我们要根据我们实际的情况，需要提供数据库和缓存方面的一些配套服务，这里就不做详解;

这样我们整体的一个分布式定时任务平台就可以实现了，就可以保证计划任务的分布式运行。

4，当前比较流行的分布式定时任务框架：

4.1 Quartz：

Quartz是Java领域最著名的开源任务调度工具。Quartz提供了极为广泛的特性如持久化任务，集群和分布式任务

特点：

完全由Java写成，同时可以很方便的和java的另外一个框架spring集成；

强大的调度功能：支持丰富多样的调度方法，可以满足各种常规及特殊需求；

灵活的应用方式：支持任务和调度的多种组合方式，支持调度数据的多种存储方式；

分布式和集群能力，负载均衡和高可用性；

4.2 Elastic-job:

Elastic-Job是ddframe中dd-job的作业模块中分离出来的分布式弹性作业框架。去掉了和dd-job中的监控和ddframe接入规范部分。该项目基于成熟的开源产品Quartz和 Zookeeper及其客户端Curator进行二次开发

项目开源地址：https://github.com/dangdangdotcom/elastic-job

特点：

定时任务：基于成熟的定时任务作业框架Quartz cron表达式执行定时任务；

作业注册中心：基于Zookeeper和其客户端Curator实现全局作业注册控制中心。用于注册，控制和协调分布式作业执行。

作业分片：将要给任务分片成多个小任务项到多服务器上同时执行；

弹性扩容缩容：运行中的作业服务器崩溃，或新增N台作业服务器，作业框架将在下次作业执行前重新分片，不影响当前作业执行；

支持多种作业执行模式：支持OneOff，Perpetual和SequenecePerpetual三种作业模式；

失效转移：运行中的作业服务器崩溃不会导致重新分片，只会在下次作业启动时分片。启用失效转移功能可以在本次作业执行过程中，监测其他作业服务器空闲，抓取未完成的孤儿分片项执行；

运行时状态收集：监控作业运行时状态，统计最近一段时间处理的数据成功和失败数量，记录作业上次运行开始时间，结束时间和下次运行时间；

作业停止，恢复和禁用：用于操作作业启动和停止，并可以禁止某作业运行，一般在上线时常用；

被错过执行的作业重触发：自动记录错过执行的作业，并在上次作业完成后自动触发。

多线程快速处理数据：使用多线程处理抓取到的数据，提升吞吐量；

幂等性：重复作业任务项判定，不重复执行已运行的作业任务项；

容错处理：作业服务器和Zookeeper服务器通信失败后则立即停止作业运行，防止作业注册中心将失效的分片分项配给其他作业服务器，而当前作业服务器任在执行任务，导致重复执行。

Spring支持：支持Spring容器，自定义命名空间，支持占位符；

运维平台：提供了运维平台，可以管理作业和注册中心。

从以上可以看出Elastic-job是在Quartz的基础上又做了一次全面的升级，做了配套的周边基础服务工作，完全成为了一个成熟的分布式定时任务框架。后面会分别介绍Quartz和 Elastic-job的详细原理和具体的使用方法。

///////////////////////

Quartz应用和集群原理分析：

使用的环境版本：spring4.x+quartz2.2.x

****1.1 如何在spring中集成quartz集群****

1.1.1 基于maven项目，需要在pom.xml引入的j依赖为：

dependency groupIdorg.quartz-scheduler/groupId artifactIdquartz/artifactId /dependency

1.1.2 Quartz集群的基本配置信息：命名为quartz.properties

org.quartz.scheduler.instanceName: DefaultQuartzScheduler org.quartz.scheduler.rmi.export: org.quartz.scheduler.rmi.proxy: org.quartz.scheduler.wrapJobExecutionInUserTransaction: ThreadPool 实现的类名 org.quartz.threadPool.class: org.quartz.simpl.SimpleThreadPool org.quartz.threadPool.threadCount: org.quartz.threadPool.threadPriority: org.quartz.threadPool.threadsInheritContextClassLoaderOfInitializingThread: org.quartz.jobStore.misfireThreshold: org.quartz.scheduler.instanceId: AUTO org.quartz.jobStore.class: org.quartz.impl.jdbcjobstore.JobStoreTX org.quartz.jobStore.isClustered: org.quartz.jobStore.clusterCheckinInterval:

1.1.3 在项目中加入Quartz的初始化信息：命名spring-quartz.xml

?xml ? beans xmlns:xsi xmlns:mvc xmlns:aop xmlns:context xsi:schemaLocation http://www.springframework.org/schema/beans/spring-beans.xsd http://www.springframework.org/schema/mvc http://www.springframework.org/schema/mvc/spring-mvc.xsd http://www.springframework.org/schema/aop http://www.springframework.org/schema/aop/spring-aop.xsd http://www.springframework.org/schema/context bean -- 自定义的bean注入类，解决job里面无法注入spring的service的问题 -- property bean / /property -- quartz的数据源 -- property / -- quartz的基本配置信息引入 -- property / -- 调度标识名 -- property / --必须的，QuartzScheduler 延时启动，应用启动完后 QuartzScheduler 再启动 -- property / -- 通过applicationContextSchedulerContextKey属性配置spring上下文 -- property / --可选，QuartzScheduler 启动时更新己存在的Job，这样就不用每次修改targetObject后删除qrtz_job_details表对应记录了 -- property / -- 设置自动启动 -- property / -- 注册触发器 -- property list ref / /list /property -- 注册jobDetail -- property list ref / /list /property /bean --配置调度具体执行的方法-- bean property / property / property / /bean --配置调度执行的触发的时间-- bean property / property -- 每天上午00:30点执行任务调度 -- value 00 * * ?/value /property /bean

1.1.4 在web.xml启动项中加入spring-quartz.xml文件*

servlet servlet-namedispatcherServlet/servlet-name servlet-classorg.springframework.web.servlet.DispatcherServlet/servlet-class init-param param-namecontextConfigLocation/param-name param-valueclasspath:/spring/spring-quartz.xml/param-value /init-param load-on-startup/load-on-startup /servlet

1.1.5 附上对应的解决无法注入的jobFactory的代码*:


 org.quartz.spi.TriggerFiredBundle
 org.springframework.beans.factory.annotation.Autowired
 org.springframework.beans.factory.config.AutowireCapableBeanFactory
 org.springframework.scheduling.quartz.AdaptableJobFactory
 org.springframework.stereotype.Component
/**
 * Created by lyndon on /9/13.
 */
@Component
public class jobFactory extends AdaptableJobFactory 
    //这个对象Spring会帮我们自动注入进来,也属于Spring技术范畴.
    @Autowired
    private AutowireCapableBeanFactory capableBeanFactory
    protected Object createJobInstanceTriggerFiredBundle bundle throws Exception 
        //调用父类的方法
        Object jobInstance  super.createJobInstancebundle
        //进行注入,这属于Spring的技术,不清楚的可以查看Spring的API.
        capableBeanFactory.autowireBeanjobInstance
         jobInstance

****1.2 quartz框架实现分布式定时任务的原理****；

Quartz集群中每个节点都是一个单独的Quartz应用，它又管理着其他的节点。这个集群需要每个节点单独的启动或停止；和我们的应用服务器集群不同，独立的Quratz节点之间是不需要通信的。不同节点之间是通过数据库表来感知另一个应用。只有使用持久的JobStore才能完成Quartz集群。

untitled21.jpg

1.2.1 既然Quartz分布式集群是利用数据库锁机制来实现集群环境下的并发控制，我们就需要了解Quratz的数据库表：可以去官方现在对于版本的sql文件导入。

untitled22.png

1.2.2 Quartz线程模型：

Quartz中有两类线程：Scheduler调度线程和任务执行线程。

任务执行线程: Quartz不会在主线程（QuartzSchedulerThread）中处理用户job。Quratz是将线程管理的职责委托给ThreadPool,一般的设置使用SimpleThreadPool，SimpleThreadPool创建一定数量的工作线程（WorkerThread），当然这样就意味所有的线程都是异步操作的，所以我们在工作线程的job里面实现业务的时候是没必要重新去创建一个新的线程的，在Quartz创建工作线程的时候已经完成了异步任务的创建。

Scheduler调度线程：QuartzScheduler被创建的时候会创建一个QuratzSchedulerThread实例。

1.2.3 Quartz源码分析：

QuartzSchedulerThreand包含有决定何时下一个Job将被触发的处理循环，主要的逻辑在其的run()方法中，如下图：

untitled23.png

由此可知，QuartzSchedulerThread不断的在获取trigger,触发trigger,释放trigger。

那么具体又是如何获取trigger的呢，可以从上面的源码中可以发现：qsRsrcs.getJobStore()返回对象是JobStore ,具体的集群配置参考1.1.2. org.quartz.jobStore.class:org.quartz.impl.jdbcjobstore.JobStoreTX

JobStoreTx继承自JobStoreSupport,而JobStoreSupport的acquireNextTrigger,triggerFired,releaseAcquiredTrigger方法负责具体trigger相关操作，都必须获得TRIGGER-ACCESS锁。核心逻辑在executeInNonManagedTxLock方法中。

untitled24.png

由上代码可知Quartz集群基于数据库锁的同步操作流程如下图所示：

untitled25.png

/////////////////////////////////

Quartz分布式定时任务的暂停和恢复等：

前两篇我们了解了quartz分布式定时任务的基本原理和实现方式，知道所有的定时任务都会被持久化到数据库。那么我们肯定可以通过操作数据库来做定时任务的暂停，恢复，立即启动，添加等操作。

事实上，quartz已经给我们提供来一些列的api接口来操作对应的定时任务，我们只需要在这个基础之上做进一步的扩展和封装就可以实现我们自己业务，下面，将围绕定时任务的控制，提供一个简单的实现方式。

使用的环境版本：spring4.x+quartz2.2.x

1,首先，我们需要创建一个我们自己job的实体类ScheduleJob：

/** 
* Created by lyndon on /9/13. * job的实体类
 */
public class ScheduleJob    
        private String jobNo //任务编号    
        private String jobName //任务名称    
        private String jobGroup //任务所属组    
        private String desc //任务描述          
        private String jobStatus //任务状态    
        private String cronExpression //任务对应的时间表达式
        private String triggerName //触发器名称
         //此处省略get和set方法

2, 创建我们自己的QuartzImplService服务层：


 org.quartz.*
 org.quartz.impl.matchers.GroupMatcher
 org.slf4j.Logger
 org.slf4j.LoggerFactory
 org.springframework.stereotype.Service
 javax.annotation.Resource
 java.util.*
/**
 * Created by lyndon on /9/13.
 * quartz_job的工具类
 */
@Service
public class QuartzUtils 
    private final Logger logger  LoggerFactory.getLoggerQuartzUtils.class
    @Resource
    private Scheduler scheduler
    /**
     *
     * 获取计划任务列表
     * @return  ListScheduleJob
     */
    public ListScheduleJob getPlanJobList throws SchedulerException
        ListScheduleJob jobList  new ArrayList
        GroupMatcherJobKey matcher  GroupMatcher.anyJobGroup
        SetJobKey jobKeys  scheduler.getJobKeysmatcher
        jobKeys  scheduler.getJobKeysmatcher
         JobKey jobKey  jobKeys 
            List? extends Trigger triggers  scheduler.getTriggersOfJobjobKey
             Trigger trigger  triggers 
                ScheduleJob job  new ScheduleJob
                job.setJobNamejobKey.getName
                job.setJobGroupjobKey.getGroup
                // 此处是我自己业务需要，给每个定时任务配置类对应的编号和描述
                String value  PropertiesUtils.getStringCNjobKey.getName
                ifnull  value  .equalsvalue
                    job.setJobNovalue.split
                    job.setDescvalue.split
                else
                    job.setJobNo
                    job.setDesc
                job.setTriggerName + trigger.getKey
                Trigger.TriggerState triggerState  scheduler.getTriggerStatetrigger.getKey
                job.setJobStatustriggerState.name
                 trigger instanceof CronTrigger 
                    CronTrigger cronTrigger  CronTrigger trigger
                    String cronExpression  cronTrigger.getCronExpression
                    job.setCronExpressioncronExpression
                jobList.addjob
      // 对返回的定时任务安装编号做排序
        Collections.sortjobList,new ComparatorScheduleJob
            public int compareScheduleJob arg0, ScheduleJob arg1 
                 arg0.getJobNo.compareToarg1.getJobNo
         jobList
    /**
     * 获取正在运行的任务列表
     * @return ListScheduleJob
     */
    public ListScheduleJob getCurrentJobList throws SchedulerException
        ListJobExecutionContext executingJobs  scheduler.getCurrentlyExecutingJobs
        ListScheduleJob jobList  new ArrayListScheduleJobexecutingJobs.size
         JobExecutionContext executingJob  executingJobs 
            ScheduleJob job  new ScheduleJob
            JobDetail jobDetail  executingJob.getJobDetail
            JobKey jobKey  jobDetail.getKey
            Trigger trigger  executingJob.getTrigger
            job.setJobNamejobKey.getName
            job.setJobGroupjobKey.getGroup
            String value  PropertiesUtils.getStringCNjobKey.getName
            ifnull  value  .equalsvalue
                job.setJobNovalue.split
                job.setDescvalue.split
            else
                job.setJobNo
                job.setDesc
            job.setTriggerName + trigger.getKey
            Trigger.TriggerState triggerState  scheduler.getTriggerStatetrigger.getKey
            job.setJobStatustriggerState.name
             trigger instanceof CronTrigger 
                CronTrigger cronTrigger  CronTrigger trigger
                String cronExpression  cronTrigger.getCronExpression
                job.setCronExpressioncronExpression
            jobList.addjob
        Collections.sortjobList,new ComparatorScheduleJob
            public int compareScheduleJob arg0, ScheduleJob arg1 
                 arg0.getJobNo.compareToarg1.getJobNo
          jobList
    /**
     * 暂停当前任务
     * @param scheduleJob
     */
    public void pauseJobScheduleJob scheduleJob throws SchedulerException
        JobKey jobKey  JobKey.jobKeyscheduleJob.getJobName, scheduleJob.getJobGroup
        ifscheduler.checkExistsjobKey
            scheduler.pauseJobjobKey
    /**
     * 恢复当前任务
     * @param scheduleJob
     */
    public void resumeJobScheduleJob scheduleJob throws SchedulerException
        JobKey jobKey  JobKey.jobKeyscheduleJob.getJobName, scheduleJob.getJobGroup
        ifscheduler.checkExistsjobKey
            //并恢复
            scheduler.resumeJobjobKey
            //重置当前时间
            this.rescheduleJobscheduleJob
    /**
     * 删除任务
     * @param scheduleJob
     * @return boolean
     */
    public boolean deleteJobScheduleJob scheduleJob throws SchedulerException
        JobKey jobKey  JobKey.jobKeyscheduleJob.getJobName, scheduleJob.getJobGroup
        ifscheduler.checkExistsjobKey
             scheduler.deleteJobjobKey
    /**
     * 立即触发当前任务
     * @param scheduleJob
     */
    public void triggerJobScheduleJob scheduleJob throws SchedulerException
        JobKey jobKey  JobKey.jobKeyscheduleJob.getJobName, scheduleJob.getJobGroup
        ifscheduler.checkExistsjobKey
            scheduler.triggerJobjobKey
    /**
     * 更新任务的时间表达式
     * @param scheduleJob
     * @return Date
     */
    public Date rescheduleJobScheduleJob scheduleJob throws SchedulerException
        TriggerKey triggerKey  TriggerKey.triggerKeyscheduleJob.getJobName,
                scheduleJob.getJobGroup
        ifscheduler.checkExiststriggerKey
            CronTrigger trigger  CronTrigger scheduler.getTriggertriggerKey
            CronScheduleBuilder scheduleBuilder  CronScheduleBuilder.cronSchedulescheduleJob
                    .getCronExpression
            //按新的cronExpression表达式重新构建trigger
            trigger  trigger.getTriggerBuilder.withIdentitytriggerKey
                    .withSchedulescheduleBuilder.build
            //按新的trigger重新设置job执行
             scheduler.rescheduleJobtriggerKey, trigger
         null
    /**
     * 查询其中一个任务的状态
     * @param scheduleJob
     * @return
     * @throws SchedulerException
     */
    public String scheduleJobScheduleJob scheduleJob throws SchedulerException 
        String status  null
        TriggerKey triggerKey  TriggerKey.triggerKeyscheduleJob.getJobName, scheduleJob.getJobGroup
        CronTrigger trigger  CronTrigger scheduler.getTriggertriggerKey
         null  trigger 
            Trigger.TriggerState triggerState  scheduler.getTriggerStatetrigger.getKey
            status  triggerState.name
         status
    /**
     * 校验job是否已经加载
     * @param scheduleJob  基本信息参数
     * @return          是否已经加载
     */
    public boolean checkJobExistedScheduleJob scheduleJob throws SchedulerException 
         scheduler.checkExistsnew JobKeyscheduleJob.getJobName, scheduleJob.getJobGroup
    private String getStatuDescString status
        ifstatus.equalsIgnoreCase
        else ifstatus.equalsIgnoreCase
        else

3，提供对应的Controller


 com.innmall.hotelmanager.common.Result
 com.innmall.hotelmanager.service.quartz.QuartzUtils
 com.innmall.hotelmanager.service.quartz.ScheduleJob
 org.quartz.SchedulerException
 org.slf4j.Logger
 org.slf4j.LoggerFactory
 org.springframework.web.bind.annotation.RequestMapping
 org.springframework.web.bind.annotation.RestController
 javax.annotation.Resource
 java.util.List
/**
 * Created by lyndon on /9/13.
 */
@RestController
@RequestMappingvalue  
public class QuartzController 
    private Logger logger  LoggerFactory.getLoggerthis.getClass
    @Resource
    private QuartzUtils quartzUtils
    //获取定时任务的列表
    @RequestMappingvalue  
    public Result getPlanJobListString openId
        //QuartzUtils quartzUtils  new QuartzUtils
        ListScheduleJob list  null
        try 
            list  quartzUtils.getPlanJobList
         catch SchedulerException e 
            e.printStackTrace
         Result.successlist
    //暂停任务
    @RequestMappingvalue  
    public Result pauseJobString openId
        //QuartzUtils quartzUtils  new QuartzUtils
        ScheduleJob job  new ScheduleJob
        job.setJobGroup
        job.setJobName
        try 
            quartzUtils.pauseJobjob
         catch SchedulerException e 
            e.printStackTrace
         Result.success
    //恢复任务
    @RequestMappingvalue  
    public Result resumeJobString openId
        //QuartzUtils quartzUtils  new QuartzUtils
        ScheduleJob job  new ScheduleJob
        job.setJobGroup
        job.setJobName
        try 
            quartzUtils.resumeJobjob
         catch SchedulerException e 
            e.printStackTrace
         Result.success
    //立即触发任务
    @RequestMappingvalue  
    public Result triggerJobString openId
        //QuartzUtils quartzUtils  new QuartzUtils
        ScheduleJob job  new ScheduleJob
        job.setJobGroup
        job.setJobName
        try 
            quartzUtils.triggerJobjob
         catch SchedulerException e 
            e.printStackTrace
         Result.success
    //删除任务
    @RequestMappingvalue  
    public Result deleteJobString openId
        //QuartzUtils quartzUtils  new QuartzUtils
        ScheduleJob job  new ScheduleJob
        job.setJobGroup
        job.setJobName
        try 
            quartzUtils.deleteJobjob
         catch SchedulerException e 
            e.printStackTrace
         Result.success