原文地址

Java内置

使用 Timer

创建 java.util.TimerTask 任务,在 run 方法中实现业务逻辑。通过 java.util.Timer 进行调度,支持按照固定频率执行。所有的 TimerTask 是在同一个线程中串行执行,相互影响。也就是说,对于同一个 Timer 里的多个 TimerTask 任务,如果一个 TimerTask 任务在执行中,其它 TimerTask 即使到达执行的时间,也只能排队等待。如果有异常产生,线程将退出,整个定时任务就失败。

  1. import java.util.Timer;
  2. import java.util.TimerTask;
  3. public class TestTimerTask {
  4. public static void main(String[] args) {
  5. TimerTask timerTask = new TimerTask() {
  6. @Override
  7. public void run() {
  8. System.out.println("hell world");
  9. }
  10. };
  11. Timer timer = new Timer();
  12. timer.schedule(timerTask, 10, 3000);
  13. }
  14. }

使用 ScheduledExecutorService

基于线程池设计的定时任务解决方案,每个调度任务都会分配到线程池中的一个线程去执行,解决 Timer 定时器无法并发执行的问题,支持 fixedRate 和 fixedDelay。

  1. import java.util.concurrent.Executors;
  2. import java.util.concurrent.ScheduledExecutorService;
  3. import java.util.concurrent.TimeUnit;
  4. public class TestTimerTask {
  5. public static void main(String[] args) {
  6. ScheduledExecutorService ses = Executors.newScheduledThreadPool(5); //按照固定频率执行,每隔5秒跑一次
  7. ses.scheduleAtFixedRate(new Runnable() {
  8. @Override
  9. public void run() {
  10. System.out.println("hello fixedRate");
  11. }
  12. }, 0, 5, TimeUnit.SECONDS);
  13. //按照固定延时执行,上次执行完后隔3秒再跑
  14. ses.scheduleWithFixedDelay(new Runnable() {
  15. @Override
  16. public void run() {
  17. System.out.println("hello fixedDelay");
  18. }
  19. }, 0, 3, TimeUnit.SECONDS);
  20. }
  21. }

Spring方案

Springboot 中提供了一套轻量级的定时任务工具 Spring Task,通过注解可以很方便的配置,支持 cron 表达式、fixedRate、fixedDelay。
Spring Task 相对于上面提到的两种解决方案,最大的优势就是支持 cron 表达式,可以处理按照标准时间固定周期执行的业务,比如每天几点几分执行。

  1. import org.springframework.scheduling.annotation.EnableScheduling;
  2. import org.springframework.scheduling.annotation.Scheduled;
  3. import org.springframework.stereotype.Component;
  4. @Component
  5. @EnableScheduling
  6. public class MyTask {
  7. /** * 每分钟的第30秒跑一次 */
  8. @Scheduled(cron = "30 * * * * ?")
  9. public void task1() throws InterruptedException {
  10. System.out.println("hello cron");
  11. }
  12. /** * 每隔5秒跑一次 */
  13. @Scheduled(fixedRate = 5000)
  14. public void task2() throws InterruptedException {
  15. System.out.println("hello fixedRate");
  16. }
  17. /** * 上次跑完隔3秒再跑 */
  18. @Scheduled(fixedDelay = 3000)
  19. public void task3() throws InterruptedException {
  20. System.out.println("hello fixedDelay");
  21. }
  22. }

业务幂等解决方案

现在的应用基本都是分布式部署,所有机器的代码都是一样的,前面介绍的 Java 和 Spring 自带的解决方案,都是进程级别的,每台机器在同一时间点都会执行定时任务。这样会导致需要业务幂等的定时任务业务有问题,比如每月定时给用户推送消息,就会推送多次。
于是,很多应用很自然的就想到了使用分布式锁的解决方案。即每次定时任务执行之前,先去抢锁,抢到锁的执行任务,抢不到锁的不执行。怎么抢锁,又是五花八门,比如使用 DB、zookeeper、redis。
使用 DB 或者 Zookeeper 抢锁
使用 DB 或者 Zookeeper 抢锁的架构差不多,原理如下:
定时任务框架 - 图1

  1. 定时时间到了,在回调方法里,先去抢锁。
  2. 抢到锁,则继续执行方法,没抢到锁直接返回。
  3. 执行完方法后,释放锁。

示例代码如下:

@Component 
@EnableScheduling public class MyTask {     
    /**      * 每分钟的第30秒跑一次      */     
    @Scheduled(cron = "30 * * * * ?")     
    public void task1() throws InterruptedException {         
        String lockName = "task1";         
        if (tryLock(lockName, 30)) {             
            System.out.println("hello cron");             
            releaseLock(lockName);         
        } else {             
            return;         
        }     
    }     
    private boolean tryLock(String lockName, long expiredTime) {         
        //TODO        
        return true;     
    }     
    private void releaseLock(String lockName) {         
        //TODO     
    } 
}

当前的这个设计,仔细一点的同学可以发现,其实还是有可能导致任务重复执行的。比如任务执行的非常快,A 这台机器抢到锁,执行完任务后很快就释放锁了。B 这台机器后抢锁,还是会抢到锁,再执行一遍任务。

使用 redis 抢锁

使用 redis 抢锁,其实架构上和 DB/zookeeper 差不多,不过 redis 抢锁支持过期时间,不用主动去释放锁,并且可以充分利用这个过期时间,解决任务执行过快释放锁导致任务重复执行的问题,架构如下:
定时任务框架 - 图2
示例代码如下:

@Component 
@EnableScheduling public class MyTask {     
    /**      * 每分钟的第30秒跑一次      */     
    @Scheduled(cron = "30 * * * * ?")     
    public void task1() throws InterruptedException {         
        String lockName = "task1";         
        if (tryLock(lockName, 30)) {             
            System.out.println("hello cron");             
            releaseLock(lockName);         
        } else {             
            return;         
        }     
    }     
    private boolean tryLock(String lockName, long expiredTime) {         
        //TODO        
        return true;     
    }     
    private void releaseLock(String lockName) {         
        //TODO     
    } 
}

看到这里,可能又会有同学有问题,加一个过期时间是不是还是不够严谨,还是有可能任务重复执行?
——的确是的,如果有一台机器突然长时间的 fullgc,或者之前的任务还没处理完(Spring Task 和 ScheduledExecutorService 本质还是通过线程池处理任务),还是有可能隔了 30 秒再去调度任务的。

轻量级框架

Quartz

Quartz[1] 是一套轻量级的任务调度框架,只需要定义了 Job(任务),Trigger(触发器)和 Scheduler(调度器),即可实现一个定时调度能力。支持基于数据库的集群模式,可以做到任务幂等执行。

Quartz架构

image.png Quartz实现过程

  1. pom中引入quart包

    <!--quartz-->
    <dependency>
     <groupId>org.quartz-scheduler</groupId>
     <artifactId>quartz</artifactId>
     <version>2.3.2</version>
    </dependency>
    
  2. 继承job类,包含excute()方法

    public class TestJob implements Job {
     @Override
     public void execute(JobExecutionContext jobExecutionContext) {
         String data = LocalDateTime.now()
             .format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));
         System.out.println("START DATA BACKUP, current time :" + data);
     }
    }
    
  3. 定时器

    public class TestScheduler {
     public static void main(String[] args) throws Exception {
         // 获取任务调度的实例
         Scheduler scheduler = StdSchedulerFactory.getDefaultScheduler();
         // 定义任务调度实例, 并与TestJob绑定
         JobDetail job = JobBuilder.newJob(TestJob.class)
             .withIdentity("testJob", "testJobGroup")
             .build();
         // 定义触发器, 会马上执行一次, 接着5秒执行一次
         Trigger trigger = TriggerBuilder.newTrigger()
             .withIdentity("testTrigger", "testTriggerGroup")
             .startNow()
             .withSchedule(SimpleScheduleBuilder.repeatSecondlyForever(5))
             .build();
         // 使用触发器调度任务的执行
         scheduler.scheduleJob(job, trigger);
         // 开启任务
         scheduler.start();
     }
    }
    /**
    OUTPUT:
    START DATA BACKUP, current time :2020-11-17 21:48:30
    START DATA BACKUP, current time :2020-11-17 21:48:35
    START DATA BACKUP, current time :2020-11-17 21:48:40
    START DATA BACKUP, current time :2020-11-17 21:48:45
    **/
    

    默认配置

    jar包中自带quartz.properties文件,若不想使用默认配置,需要重新覆盖
    image.png

    实现动态调度

    设置job参数任务表,读取表任务,维护任务信息
    定时任务框架 - 图5
    Quartz 支持任务幂等执行,其实理论上还是抢 DB 锁,我们看下 quartz 的表结构:
    其中,QRTZ_LOCKS 就是 Quartz 集群实现同步机制的行锁表,其表结构如下:

    --QRTZ_LOCKS表结构 
    CREATE TABLE `QRTZ_LOCKS` (   
    `LOCK_NAME` varchar(40) NOT NULL,  
    PRIMARY KEY (`LOCK_NAME`) 
    ) ENGINE=InnoDB DEFAULT CHARSET=utf8; 
    --QRTZ_LOCKS记录 +-----------------+  | LOCK_NAME       | +-----------------+  | CALENDAR_ACCESS | | JOB_ACCESS      | | MISFIRE_ACCESS  | | STATE_ACCESS    | | TRIGGER_ACCESS  | +-----------------+
    

可以看出 QRTZ_LOCKS 中有 5 条记录,代表 5 把锁,分别用于实现多个 Quartz Node 对 Job、Trigger、Calendar 访问的同步控制。

开源中间件

上面提到的解决方案,在架构上都有一个问题,那就是每次调度都需要抢锁,特别是使用 DB 和 Zookeeper 抢锁,性能会比较差,一旦任务量增加到一定的量,就会有比较明显的调度延时。还有一个痛点,就是业务想要修改调度配置,或者增加一个任务,得修改代码重新发布应用。
于是开源社区涌现了一堆任务调度中间件,通过任务调度系统进行任务的创建、修改和调度,这其中国内最火的就是 XXL-JOB 和 ElasticJob。

ElasticJob

ElasticJob[2] 是一款基于 Quartz 开发,依赖 Zookeeper 作为注册中心、轻量级、无中心化的分布式任务调度框架,目前已经通过 Apache 开源。
ElasticJob 相对于 Quartz 来说,从功能上最大的区别就是支持分片,可以将一个任务分片参数分发给不同的机器执行。架构上最大的区别就是使用 Zookeeper 作为注册中心,不同的任务分配给不同的节点调度,不需要抢锁触发,性能上比 Quartz 上强大很多,架构图如下:
定时任务框架 - 图6
开发上也比较简单,和 springboot 结合比较好,可以在配置文件定义任务如下:

elasticjob:   
  regCenter:     
    serverLists: 
      localhost: 2181     
      namespace: elasticjob-lite-springboot   
  jobs:     
    simpleJob:       
    elasticJobClass: org.apache.shardingsphere.elasticjob.lite.example.job.SpringBootSimpleJob       
    cron: 0/5 * * * * ?       
    timeZone: GMT+08:00       
    shardingTotalCount: 3       
    shardingItemParameters: 0=Beijing,1=Shanghai,2=Guangzhou     
  scriptJob:       
    elasticJobType: SCRIPT       
    cron: 0/10 * * * * ?       
    shardingTotalCount: 3       
  props:         
    script.command.line: "echo SCRIPT Job: "     
    manualScriptJob:       
    elasticJobType: SCRIPT       
    jobBootstrapBeanName: manualScriptJobBean       
    shardingTotalCount: 9       
  props:         
    script.command.line: "echo Manual SCRIPT Job: "

实现任务接口如下:

@Component 
public class SpringBootShardingJob implements SimpleJob {     
    @Override     
    public void execute(ShardingContext context) {         
        System.out.println("分片总数="+context.getShardingTotalCount() + ", 分片号="+context.getShardingItem()+ ", 分片参数="+context.getShardingParameter());     
    }

运行结果如下:
分片总数=3, 分片号=0, 分片参数=Beijing 分片总数=3, 分片号=1, 分片参数=Shanghai 分片总数=3, 分片号=2, 分片参数=Guangzhou
同时,ElasticJob 还提供了一个简单的 UI,可以查看任务的列表,同时支持修改、触发、停止、生效、失效操作。
定时任务框架 - 图7
遗憾的是,ElasticJob 暂不支持动态创建任务。

XXL-JOB

XXL-JOB[3] 是一个开箱即用的轻量级分布式任务调度系统,其核心设计目标是开发迅速、学习简单、轻量级、易扩展,在开源社区广泛流行。
XXL-JOB 是 Master-Slave 架构,Master 负责任务的调度,Slave 负责任务的执行,架构图如下:
定时任务框架 - 图8
XXL-JOB 接入也很方便,不同于 ElasticJob 定义任务实现类,是通过@XxlJob 注解定义 JobHandler。

@Component 
public class SampleXxlJob {     
    private static Logger logger = LoggerFactory.getLogger(SampleXxlJob.class);   
    /**      * 1、简单任务示例(Bean模式)      */     
    @XxlJob("demoJobHandler")     
    public ReturnT<String> demoJobHandler(String param) throws Exception {       
        XxlJobLogger.log("XXL-JOB, Hello World.");         
        for (int i = 0; i < 5; i++) {             
            XxlJobLogger.log("beat at:" + i);             
            TimeUnit.SECONDS.sleep(2);         
        }         return ReturnT.SUCCESS;     
    }     
    /**      * 2、分片广播任务      */    
    @XxlJob("shardingJobHandler")    
    public ReturnT<String> shardingJobHandler(String param) throws Exception {
        // 分片参数         
        ShardingUtil.ShardingVO shardingVO = ShardingUtil.getShardingVo();
        XxlJobLogger.log("分片参数:当前分片序号 = {}, 总分片数 = {}", shardingVO.getIndex(), shardingVO.getTotal());         
        // 业务逻辑         
        for (int i = 0; i < shardingVO.getTotal(); i++) {             
            if (i == shardingVO.getIndex()) {
                XxlJobLogger.log("第 {} 片, 命中分片开始处理", i);             
            } else {                 
                XxlJobLogger.log("第 {} 片, 忽略", i);             
            }         
        }         
        return ReturnT.SUCCESS;     
    } 
}

XXL-JOB 相较于 ElasticJob,最大的特点就是功能比较丰富,可运维能力比较强,不但支持控制台动态创建任务,还有调度日志、运行报表等功能。
定时任务框架 - 图9
定时任务框架 - 图10
XXL-JOB 的历史记录、运行报表和调度日志,都是基于数据库实现的:
定时任务框架 - 图11
由此可以看出,XXL-JOB 所有功能都依赖数据库,且调度中心不支持分布式架构,在任务量和调度量比较大的情况下,会有性能瓶颈。不过如果对任务量级、高可用、监控报警、可视化等没有过高要求的话,XXL-JOB 基本可以满足定时任务的需求。

企业级方案

开源软件只能提供基础的调度能力,在监管控上的能力一般都比较弱。比如日志服务,业界往往使用 ELK 解决方案;短信报警,需要有短信平台;监控大盘,现在主流的解决方案是 Prometheus;等等。企业想要有这些能力,不但需要额外的开发成本,还需要昂贵的资源成本。
另外使用开源软件也伴随着稳定性的风险,就是出了问题没人能处理,想要反馈到社区等社区处理,这个链路太长了,早就产生故障了。

SchedulerX

阿里云任务调度 SchedulerX[4] 是阿里巴巴自研的基于 Akka 架构的一站式任务调度平台,兼容开源 XXL-JOB、ElasticJob、Quartz(规划中),支持 Cron 定时、一次性任务、任务编排、分布式跑批,具有高可用、可视化、可运维、低延时等能力,自带企业级监控大盘、日志服务、短信报警等服务。
优势
安全防护

  • 多层次安全防护:支持 HTTPS 和 VPC 访问,同时还有阿里云的多层安全防护,防止恶意攻击。
  • 多租户隔离机制:支持多地域、命名空间和应用级别的隔离。
  • 权限管控:支持控制台读写的权限管理,客户端接入的鉴权。

企业级高可用
SchedulerX2.0 采用高可用架构,任务多备份机制,经历阿里集团多年双十一、容灾演练,可以做到任意一个机房挂了,任务调度都不会收到影响。
商业级报警运维

  • 报警:支持邮件、钉钉、短信、电话,(其他报警方式在规划中)。支持任务失败、超时、无可用机器报警。报警内容可以直接看出任务失败的原因,以钉钉机器人为例。

定时任务框架 - 图12

  • 运维操作:原地重跑、重刷数据、标记成功、查看堆栈、停止任务、指定机器等。

定时任务框架 - 图13
丰富的可视化
schedulerx 拥有丰富的可视化能力,比如:

  • 用户大盘

定时任务框架 - 图14

  • 查看任务历史执行记录

定时任务框架 - 图15

  • 查看任务运行日志

定时任务框架 - 图16

  • 查看任务运行堆栈

定时任务框架 - 图17

  • 查看任务操作记录

定时任务框架 - 图18
兼容开源
Schedulerx 兼容开源 XXL-JOB、ElasticJob、Quartz(规划中),业务不需要改一行代码,即可以将任务托管在 SchedulerX 调度平台,享有企业级可视化和报警的能力。
定时任务框架 - 图19
SchedulerX 支持通过控制台和 API 动态创建任务,也支持 Spring 声明式任务定义,一份任务配置可以拿到任何环境一键启动,配置如下:

spring:    
  schedulerx2:
    endpoint: acm.aliyun.com
    #请填写不同regin的endpoint
    namespace: 433d8b23-06e9-xxxx-xxxx-90d4d1b9a4af
    #region内全局唯一,建议使用UUID生成
    namespaceName: 学仁测试
    appName: myTest
    groupId: myTest.group
    #同一个命名空间下需要唯一
    appKey: myTest123@alibaba
    #应用的key,不要太简单,注意保管好
    regionId: public
    #填写对应的regionId
    aliyunAccessKey: xxxxxxx
    #阿里云账号的ak
    aliyunSecretKey: xxxxxxx
    #阿里云账号的sk
    alarmChannel: sms,ding
    #报警通道:短信和钉钉
  jobs:
    simpleJob:
      jobModel: standalone
      className: com.aliyun.schedulerx.example.processor.SimpleJob
      cron: 0/30 * * * * ?
      # cron表达式
      jobParameter: hello
      overwrite: true
  shardingJob:
     jobModel: sharding
    className: ccom.aliyun.schedulerx.example.processor.ShardingJob
    oneTime: 2022-06-02 12:00:00   
    # 一次性任务表达式
    jobParameter: 0=Beijing,1=Shanghai,2=Guangzhou
    overwrite: true
  broadcastJob:
    # 不填写cron和oneTime,表示api任务
    jobModel: broadcast
    className: com.aliyun.schedulerx.example.processor.BroadcastJob
    jobParameter: hello
    overwrite: true
  mapReduceJob:
    jobModel: mapreduce
    className: com.aliyun.schedulerx.example.processor.MapReduceJob 
    cron: 0 * * * * ?
    jobParameter: 100
    overwrite: true
  alarmUsers: 
    #报警联系人
    user1:
      userName: 张三 
      userPhone: 12345678900 
    user2:
      userName: 李四
      ding: https://oapi.dingtalk.com/robot/send?access_token=xxxxx

分布式跑批
SchedulerX 提供了丰富的分布式模型,可以处理各种各样的分布式业务场景。包括单机、广播、分片、MapReduce[5] 等,架构如下:
定时任务框架 - 图20
SchedulerX 的 MapReduce 模型,简单几行代码,就可以将海量任务分布式到多台机器跑批,相对于大数据跑批来说,具有速度快、数据安全、成本低、简单易学等特点。
任务编排
SchedulerX 通过工作流进行任务编排,并且提供了一个可视化的界面,操作简单,拖拖拽拽即可配置一个工作流。详细的任务状态图能一目了然看到下游任务为什么没跑,方便定位问题。
定时任务框架 - 图21
可抢占的任务优先级队列
常见场景是夜间离线报表业务,比如很多报表任务是晚上 1、2 点开始跑,要控制应用最大并发的任务数量(否则业务扛不住),达到并发上限的任务会在队列中等待。同时要求早上 9 点前必须把 KPI 报表跑出来,可以设置 KPI 任务高优先级,会抢占低优先级任务优先调度。
SchedulerX 支持可抢占的任务优先级队列,可以在控制台动态配置:
定时任务框架 - 图22