如何自己设计一个定时任务分布式调度器

你的名字 2023-09-26 10:21 4阅读 0赞

## 为什么要使用分布式调度器 ##

分布式调度器主要应用于系统中一些任务定时调度处理。通常我们设计一个定时任务，最简单的就是直接使用`@scheduled`注解配置好定时任务，这样开发工作也简单。但是也许会有一种情况，如果发生在生产环境上，需要不重启就去变更定时任务时间，或者可能由于某些原因我们需要关闭某个定时任务，那么这时候就无法做到动态化。分布式调度器就能很好的解决这些疑难杂症。

有的人可能会问：现在开源的调度器也有一些很流行的，比如`xxl-job`，为什么还要自己设计一套。其实我们也不能说开源的设计不好，原因是它的功能太完善，如果要用好还要有人专门运维处理，功能过于强大，大部分功能都是鸡肋，所以自研一套简单的调度服务有些时候还是很有必要的。

## 分布式调度流程 ##

首先分布式调度器需要依赖数据库配置，主要配置调度服务接口和调度时间。通过调度服务集群获取数据库配置，解析完需要进行调度的任务，由于是job服务的一个集群（也可以单机部署）所以也要考虑到加锁，防止多个`job`服务同时对一个任务多次调度。最终`job`服务将解析完的服务接口，检测到触发时间点就对应用服务接口发起任务调度。

![format_png][]

## 分布式调度细节设计分析 ##

### 数据库设计 ###

> **job\_info表设计**：主要记录一些job任务的配置，下面分析一下主要字段：
> 
>  *  job\_cron：定时任务触发时间配置
>  *  config\_id：关联调度任务服务接口配置主键
>  *  execute\_timeout：任务调度超时时间配置，防止调度时间过长无结果
>  *  execute\_fail\_retry\_count：如果任务调度失败重试次数
>  *  job\_status：调度任务状态开关配置
>  *  trigger\_last\_time：最后一次调度时间
>  *  trigger\_next\_time：下一次执行时间

![format_png 1][]

> **job\_config表设计**：主要配置一些任务对应需要调度的服务接口信息。
> 
>  *  execute\_servier：所需调度的应用服务
>  *  execute\_method：调度应用服务接口
>  *  execute\_param：调度参数配置
>  *  service\_type：服务类型（GET/POST）

![format_png 2][]

### job服务调度流程设计 ###

*  读取配置：首先`job`服务需要不断的读取数据库配置，从而得知有哪一些任务需要进行调度。可以通过一个`while`循环加上休眠一段时间不断读取配置，下面就用简短的伪代码做个思路分析：

while(true) {
        // PRE_READ_TIME每次刷新时间间隔
        TimeUnit.MILLISECONDS.sleep(PRE_READ_TIME - System.currentTimeMillis() % 1000);
        // 读取配置，给定一个时间，获取这段时间内要执行调度的任务以及初次配置的任务trigger_next_time=0
        List<JobInfo> jobInfos = jobInfoMapper.select(time);
        // 循环对任务一一进行解析
        // 1.job应用对获取到的任务进行加锁，防止job集群其他服务同时调用，如果确定只会有单机部署可不加锁
        int resultCount = jobInfoMapper.updateByOptimisticLock(jobInfo);
        // 2.加锁成功继续执行下一步，对首次配置的任务(trigger_next_time=0)需要获取job_cron进行解析，计算出真实的下次执行时间trigger_next_time
         refreshNextValidTime(jobInfo, new Date(nowTime));
        // 3.即将要执行的任务加入队列
       checkHighFrequency(jobInfo, nowTime);
    }
    
    private void checkHighFrequency(JobInfo jobInfo, Long nowTime) throws ParseException {
        // PRE_READ_TIME = 5000，即提前预留5秒，将任务加入队列
        if (jobInfo.getTriggerNextTime() < (nowTime + PRE_READ_TIME)) {
            // 将任务放入待执行队列
            triggerPoolHelper.triggerJob(jobInfo, jobInfo.getTriggerNextTime() - nowTime);
            // 任务加入队列后，再次更新计算下次调度job的时间
            refreshNextValidTime(jobInfo, new Date(jobInfo.getTriggerNextTime()));
            // 判断是否是超高频繁任务，即调度周期小于5s一次
            checkHighFrequency(jobInfo, nowTime);
        }
    }
    // 计算下次执行时间
    private void refreshNextValidTime(JobInfo jobInfo, Date fromTime) throws ParseException {
        // 时间表达式转换计算下次触发时间
        Date nextValidTime = new CronExpression(jobInfo.getJobCron()).getNextValidTimeAfter(fromTime);
        if (nextValidTime != null) {
            jobInfo.setTriggerLastTime(jobInfo.getTriggerNextTime());
            jobInfo.setTriggerNextTime(nextValidTime.getTime());
        }
    }
    复制代码

*  线程池队列执行任务调度

public void triggerJob(JobInfo jobInfo, long delay) {
        JobInfo copyOf = new JobInfo();
        BeanUtils.copyProperties(jobInfo, copyOf);
        JobTriggerThread triggerThread = new JobTriggerThread(copyOf, tinyJobExecutor.get(jobInfo.getJobType()));
        // 小于0说明是延期的任务，立即执行
        if (delay <= 0) {
            // 加入线程池
            triggerPool.execute(triggerThread);
        }
        // 大于0说明还未到调度时间,延迟调度
        else {
            triggerPool.schedule(triggerThread, delay, TimeUnit.MILLISECONDS);
        }
    }
    复制代码

*  任务调度流程：`SpringCloud`服务使用`DiscoveryClient`，根据`job_config`表配置的服务名获取集群服务列表，再根据随机（或自定义算法）计算获取一个服务实例，用该实例创建请求并发起服务接口调用，最终再根据调用结果进行日志记录，以及失败后续是否进行重试处理。

List<ServiceInstance> serviceInstanceList = discoveryClient.getInstances(jobConfig.getExecuteService());
    // 随机获取服务列表（可自定义算法）
    ServiceInstance serviceInstance = getRandomInstrance(serviceInstanceList);
    CloseableHttpClient httpClient = HttpClients.createDefault();
    // 创建请求
    HttpPost httpPost = new HttpPost(serviceInstance.getUri() + "/" + jobConfig.getExecuteMethod() + "?" + jobConfig.getExecuteParam());
    // http发起调用
    CloseableHttpResponse response = httpClient.execute(httpPost);
    复制代码

## 案例配置说明 ##

*  添加两个定时任务配置，此配置如有需要也可开发个简单的页面方便配置添加与更改。
    
    ![format_png 3][]
 *  定时任务对应的调度服务接口配置
    
    ![format_png 4][]

根据以上的配置，定时刷新获取任务列表，任务首次配置`trigger_next_time=0`，需解析成具体执行时间点，任务调度判断该时间点是否达到可执行时间，在达到指定时间点job服务将对该接口发起调用并记录调度日志。

## 总结 ##

使用分布式调度器能够很好的管理我们的定时任务接口，开发人员也只需专注开发业务接口，让业务与配置完全分离。定时配置还可以根据业务场景统一进行时间协调管理，以免在有些时间点多任务同时处理，可以将时间配置的分散点以减轻`CPU`的压力。如果系统业务量少，定时任务也不多的情况也没必要多浪费时间开发一个调度系统。

[format_png]: https://img-blog.csdnimg.cn/img_convert/56ed720e31f7e26162e9e2141a466817.webp?x-oss-process=image/format,png
[format_png 1]: https://img-blog.csdnimg.cn/img_convert/d902f00d78b0a5f1f8f5ee224740cb44.webp?x-oss-process=image/format,png
[format_png 2]: https://img-blog.csdnimg.cn/img_convert/c8eba5eecad52257c74881af9fef882e.webp?x-oss-process=image/format,png
[format_png 3]: https://img-blog.csdnimg.cn/img_convert/c84159cb18b5ba43f043ab7b831a5ceb.webp?x-oss-process=image/format,png
[format_png 4]: https://img-blog.csdnimg.cn/img_convert/823f9280d950932561c650463953ee70.webp?x-oss-process=image/format,png