分布式集群定时任务

发表于

背景

根据是否可重复执行，将定时任务分为两种，一种是清除缓存数据或清理数据库过期数据等重复执行无影响的，另一种是不可重复执行的，如定时创建订单，定时处理数据，定时发邮件。

单机单节点的服务升级为容器多节点，各节点通过负载均衡对外提供一致服务，升级后会有定时任务重复执行的问题。

解决多点同时执行定时任务的关键是，如何保证多点中仅某一个节点处理同一任务。思路如各节点的共识机制、主从方案(保证某个节点为主节点，仅主节点执行定时任务)、多节点执行定时任务加锁(通过单线程的redis存储状态)等。

集群各节点共用的部分则为问题解决的关键，方案如数据库(mysql、redis、mongodb等)的状态记录、配置中心通过ip标记执行节点、注册中心取最小索引为主节点。

方案举例

以下均为在SpringBoot下测试，本地测试时配置不同的服务端口保证端口资源无冲突。
备注：①任务需保留请求触发的接口，预防定时任务因某种原因失效。②多点部署时保证各节点服务无时间差。

标记处理

各节点定时任务生成uuid入库，一段时间后，查询最后入库的记录(即先标记的为无效记录)，与当前uuid比对，uuid相同的节点则执行定时任务。实现如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
@Component
@EnableScheduling
public class ScheduleTask {
    private static Logger logger = LoggerFactory.getLogger(ScheduleTask.class);
    private final ScheduleDataDao scheduleDataDao;
    @Autowired
    public ScheduleTask(ScheduleDataDao scheduleDataDao) {
        this.scheduleDataDao = scheduleDataDao;
    }
    @Bean
    public TaskScheduler taskScheduler() {
        return new ConcurrentTaskScheduler();
    }
    @Scheduled(fixedRate = 3000)
    public void testSchedule() throws InterruptedException {
        String testTaskFlag = "testTaskId";
        logger.info("start schedule, time: {}", System.currentTimeMillis());
        //生成节点标识
        String uuid = UUID.randomUUID().toString();
        //任务记录入库
        Map<String, Object> param = new HashMap<>();
        param.put("taskId", testTaskFlag);
        param.put("uuid", uuid);
        scheduleDataDao.addTaskRecord(param);
        //休眠
        Thread.sleep(1000);
        //有效节点判断
        String validUuid = scheduleDataDao.queryLatestTaskRecord(testTaskFlag);
        if (validUuid != null && uuid.equals(validUuid)) {
            //处理流程TODO、执行记录入库
            param.put("custom", "insert operation");
            scheduleDataDao.addActualTaskRecord(param);
            logger.info("schedule execute finished, time: {}", System.currentTimeMillis());
        }
        logger.info("end schedule, time: {}", System.currentTimeMillis());
    }
}

以上方式，两张数据表，一张用于存储所有节点定时任务标记，作为执行判断的依据，另一张仅用于存储实际执行任务记录，通过数据库表清晰了解执行情况。

生产环境如果不能保证集群节点有相同的时钟序列，可在插入标记记录前，查询最后的标记记录，不存在则插入，若存在，则判断时间间隔是否大于某个阈值，大于则插入。

整合Quartz

Quartz是功能完善的任务调度框架，支持集群环境下的任务调度，代价是将任务调度状态序列化到数据库。Quartz框架需要10多张表协同，配置繁多。
Quartz 集群和其他分布式集群不同的是，集群实例之间不需要互相通信，只需和DB交互，通过DB感知其他节点，实现Job调度。因此节点扩容只需启动新实例，不需要做额外配置。
Quartz组件：

  • Scheduler：任务调度控制器，管理 Trigger 和 Job，所有的调度都是由它控制。
  • Trigger：任务调度单元，定义触发的条件。CronTrigger 可以通过 Cron 表达式规定任务触发规则，SimpleTrigger 规定任务执行几次，每次的时间间隔，类似 SchedulerExecutor。
  • Job：调度任务，JobDetail定义业务执行的具体过程。一个 Job 可以对应多个 Trigger，一个 Trigger 只能对应一个 Job。

Scheduler线程：

  • 调度线程，负责任务调度 (QuartzSchedulerThread)
  • 工作线程池，负责执行任务 (QuartzSchedulerResources)

官网
个人示例项目
参考1、参考2
maven pom依赖

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
<!-- https://mvnrepository.com/artifact/org.quartz-scheduler/quartz -->
<dependency>
    <groupId>org.quartz-scheduler</groupId>
    <artifactId>quartz</artifactId>
    <version>2.3.0</version>
</dependency>
<!-- https://mvnrepository.com/artifact/org.quartz-scheduler/quartz-jobs -->
<dependency>
    <groupId>org.quartz-scheduler</groupId>
    <artifactId>quartz-jobs</artifactId>
    <version>2.3.0</version>
</dependency>
<!-- https://mvnrepository.com/artifact/org.springframework/spring-context-support -->
<dependency>
    <groupId>org.springframework</groupId>
    <artifactId>spring-context-support</artifactId>
</dependency>

Quartz相关表

1
2
3
4
5
6
7
8
9
10
11
QRTZ_BLOB_TRIGGERS
QRTZ_CALENDARS
QRTZ_CRON_TRIGGERS 
QRTZ_FIRED_TRIGGERS
QRTZ_JOB_DETAILS
QRTZ_LOCKS
QRTZ_PAUSED_TRIGGER_GRPS
QRTZ_SCHEDULER_STATE
QRTZ_SIMPLE_TRIGGERS
QRTZ_SIMPROP_TRIGGERS
QRTZ_TRIGGERS

外部请求

服务本身留有请求触发的接口，接收请求后执行原有定时任务的业务逻辑。如脚本(py、shell或其他外部程序)定时请求到容器组，容器组负载均衡到某一pod处理。若使用kubernetes管理容器服务，定制任务使用kubernetes cronjob。