调度系统功能架构图（延迟任务调度系统）
调度系统功能架构图（延迟任务调度系统）原理如下:通过死信和死信路由实现下面所讨论技术方案的前提是精确触发，所以我们不讨论目前业界的一些分布式调度系统如：elastic-job xxl-job tbschedule等 这些系统解决不了延迟任务精确触发问题。可以实现的方案有？RabbitMQ实现
延迟任务调度系统（技术选型与设计）- 延迟任务调度系统（技术选型与设计）延迟任务的场景是？
- 现有的解决方案是？
- 存在的问题是什么？
- 希望达到的目标是？
- 可以实现的方案有？RabbitMQ实现通过死信和死信路由实现
- 通过延迟消息插件来实现
- Redis实现
- DelayQueue实现
- 时间轮实现单表时间轮
- 分层时间轮
- 之前的设计（DB/DelayQueue/ZooKeeper）
- 另一种方案（DB/DelayQueue/ZooKeeper/MQ
延迟任务的场景是？
- 习题考试截止前3天，给未提交用户发送消息
- 学习项目开课前2小时，给参与用户发送通知
- 问卷开始收集时，才对用户可见
- 问卷结束收集时，触发一些操作
- 指定时间发布课件
- 课程结束时，开始计算用户结业信息
- 直播时间到了，给用户发送消息
- 用户下单后，30分钟内未付款，关闭订单
- 用户付款后，24小时内未发货，提示发货
- 用户打车后，48小时后自动评价为5星
- 这类业务的特点是：延迟执行。一种比较简单的方法是使用后台线程扫描符合条件的业务数据，逐一处理。 这种方法扫描间隔时间不好设置，间隔时间过大影响精确度，过小则影响效率和性能。
现有的解决方案是？
- 通过linux的crontab触发定时任务
- 扫描业务表，筛选出符合条件的数据对其进行操作
存在的问题是什么？
- 由于每种类型的任务都设有扫描间隔，任务不能精确处理
- 扫描业务库，影响业务正常操作
- 任务的执行过于密集，容易导致服务器间隔性压力
- 存在系统单点，触发定时调度的服务挂了，所有任务都不会执行
- 系统不具容错能力，一旦错过了，任务就不会再被执行
- 没有统一的视图来查看任务的执行情况
- 没有告警来提示失败的任务
希望达到的目标是？
- 精确性(可在指定时间触发任务处理)
- 通用性
- 高性能(集群能力不少于1000TPS)
- 高可用(支持多实例部署)
- 可伸缩(增加和减少服务时，任务会重新分配)
- 可重试(任务失败可重试)
- 多协议(支持http\dubbo调用)
- 可管理(业务使用方可修改、删除任务)
- 能告警(失败次数达到阈值可触发告警)
- 统一视图(方便查看任务执行情况 可手动干预任务执行)
下面所讨论技术方案的前提是精确触发，所以我们不讨论目前业界的一些分布式调度系统如：elastic-job xxl-job tbschedule等 这些系统解决不了延迟任务精确触发问题。
可以实现的方案有？
RabbitMQ实现
通过死信和死信路由实现
原理如下:
何为死信:
- 消息被拒绝
- 消息已过期
- 队列达到最大长度
RabbitMQ可以对队列和消息设置x-message-tt、expiration来控制消息的存活时间，如果超时，消息变为死信。
何为死信路由:
RabbitMQ可以对队列设置x-dead-letter-exchange和x-dead-letter-routing-key两个参数。
当消息在一个队列中变成死信后会按这两个参数路由，消息就可以重新被消费。
实例操作:
- 创建延迟队列(设置死信路由)
- 创建就绪队列
- 创建死信路由
- 绑定死信路由与就绪队列
- 发送延迟消息
- 消息过期后进入就绪队列
优点：
- 高效，可以利用RabbitMQ的分布式特性轻易进行横向扩展，且支持持久化
缺点：
- 不支持对已发送的消息进行管理
- 一个消息比在同一队列中的其他消息提前过期，提前过期的消息也不会优先进入死信队列。
所以需要确保业务上每个任务的延迟时间是一致的。如果有不同延时的任务，需要为每种不同延迟的任务单独创建消息队列，缺乏灵活性。
通过延迟消息插件来实现
原理如下：
核心代码流程：
其原理是延迟消息会被保存到Mnesia表，在Exchange中根据每个message头设置的延迟时间x-delay，消息过期后才路由到对应队列。
实例操作:
- 下载插件
- https://bintray.com/rabbitmq/community-plugins/download_file?file_path=rabbitmq_delayed_message_exchange-0.0.1.ez
- 安装插件
docker-compose.xml(将插件安装到容器中) version: '2' services: rabbitmq: hostname: rabbitmq image: rabbitmq:3.6.8-management mem_limit: 200m ports: - "5672:5672" - "15672:15672" volumes: - ~/dockermapping/rabbitmq/lib:/var/lib/rabbitmq/ - /Users/oldlu/workspace/document/docker-compose/rabbitmq/rabbitmq_delayed_message_exchange-0.0.1.ez:/usr/lib/rabbitmq/lib/rabbitmq_server-3.6.8/plugins/rabbitmq_delayed_message_exchange-0.0.1.ez 启用插件 rabbitmq-plugins enable rabbitmq_delayed_message_exchange
- 创建类型为x-delayed-message的路由
- 创建就绪队列
- 绑定队列和路由
- 发布延迟消息(设置x-delay=延迟的毫秒数)
源码分析
https://github.com/rabbitmq/rabbitmq-delayed-message-exchange/blob/master/src/rabbit_delayed_message.erl
核心函数 消息入队：internal_delay_message 启动Timer：maybe_delay_first 消息处理：handle_info
优点：
- 一个消息比其他消息提前过期，提前过期的消息会被提前路由到队列，不需要为不同延迟的消息创建单独的消息队列。
缺点：
- 不支持对已发送的消息进行管理
- 集群中只有一个数据副本(保存在当前节点下的Mnesia表中)，如果节点不可用或关闭插件会丢失消息。
- 目前该插件只支持disk节点，不支持RAM节点
- 性能比原生差一点(普通的Exchange收到消息后直接路由到队列，而延迟队列需要判断消息是否过期，未过期的需要保存在表中，时间到了再捞出来路由)
Redis实现
有序集合(Sorted Set)是Redis提供的一种数据结构，具有set和hash的特点。
其中每个元素都关联一个score，并以这个score来排序。
其内部实现用到了两个数据结构:hash table和 skip list(跳跃表)
skip list的特点
- 由很多层结构组成，level是通过一定的概率随机产生的
- 每一层都是一个有序的链表，默认是升序
- 最底层的链表包含所有元素
- 如果一个元素出现在Level i的链表中，则它在Level i之下的链表也都会出现
- 每个节点包含两个指针，一个指向同一链表中的下一个元素，一个指向下面一层的元素
- 插入和删除的时间复杂度是O(logn)，当达到了一定的数据规模之后，它的效率与红黑树差不多
主要命令
- zadd:向Sorted Set中添加元素
- zrem:删除Sorted Set中的指定元素
- zrange:按照从小到大的顺序返回指定区间内的元素
实现延迟队列
- 将延迟任务加到Sorted Set，将延迟时间设为score
- 启动一个线程不断判断Sorted Set中第一个元素的score是否大于当前时间
- 如果大于，从Sorted Set中移除任务并添加到执行队列中
- 如果小于，进行短暂休眠后重试
实例操作
root@redis:/usr/local/bin# redis-cli 127.0.0.1:6379> zadd delayqueue 1 task1 (integer) 1 127.0.0.1:6379> zadd delayqueue 2 task2 (integer) 1 127.0.0.1:6379> zadd delayqueue 4 task4 (integer) 1 127.0.0.1:6379> zadd delayqueue 3 task3 (integer) 1 127.0.0.1:6379> 127.0.0.1:6379> zrange delayqueue 0 0 withscores 1) "task1"
优点：
- 实现简单
- 任务可管理（可删除、修改任务）
缺点：
- 需要有短轮询线程不断判断第一个元素是否过期，造成CPU空耗
- 分布式场景中，容易引起多个节点读取到相同任务
DelayQueue实现
DelayQueue是一个使用优先队列实现的BlockingQueue，优先队列比较的是时间，内部存储的是实现Delayed接口的对象。 只有在对象过期后才能从队列中获取对象。
内部结构
- 可重入锁
- 用于根据delay时间排序的优先级队列
- 用于优化阻塞通知的线程leader
- 用于实现阻塞和通知的Condition对象
Leader/Followers
Leader/Followers是多个工作线程轮流进行事件监听、分发、处理的一种模式。 该模式最大的优点在于，它是自己监听事件并处理客户请求，从接收到处理都是在同一线程中完成， 所以不需要在线程之间传递数据，解决线程频繁切换带来的开销。
该模式工作的任何时间点，只有一个线程成为Leader ，负责事件监听，而其他线程都是Follower，在休眠中等待成为Leader。 该模式的工作线程存在三种状态，工作线程同一时间只能处于一种状态，这三种状态为：
- Leading:线程处于领导者状态，负责事件监听。Leader监听到事件后，有两种处理方式:
- 可以转移至Processing状态，自己处理该事件，并调用方法推选新领导者。
- 也可以指定其他Follower来处理事件，此时Leader状态不变。
- Processing:线程正在处理事件，处理完事件如果当前线程集中没有领导者，它将成为新领导者，否则转为追随者。
- Following:线程处于追随者状态，等待成为新的领导者也可能被领导者指定来处理新的事件。
核心源码分析：
- 入队
public boolean offer(E e) { final ReentrantLock lock = this.lock; lock.lock(); try { q.offer(e); if (q.peek() == e) {//入队对象延迟时间是队列中最短的 leader = null;//重置leader available.signal();//唤醒一个线程去监听新加入的对象 } return true; } finally { lock.unlock(); } }
- 出队
public E take() throws InterruptedException { final ReentrantLock lock = this.lock; lock.lockInterruptibly(); try { for (;;) { E first = q.peek(); if (first == null) available.await();//队列为空，无限等待 else { long delay = first.getDelay(TimeUnit.NANOSECONDS); if (delay <= 0)//延迟时间已过，直接返回 return q.poll(); else if (leader != null)//已有leader在监听了，无限等待 available.await(); else { Thread thisThread = Thread.currentThread(); leader = thisThread;//当前线程成为leader try { available.awaitNanos(delay);//在delay纳秒后唤醒 } finally { if (leader == thisThread)// 入队一个最小延迟时间的对象时leader会被清空 leader = null; } } } } } finally { if (leader == null && q.peek() != null)//leader不存在且队列不为空，唤醒一个follower去成为leader去监听 available.signal(); lock.unlock(); } }
优点：
- 效率高，任务触发时间延迟低
缺点：
- 数据是保存在内存，需要自己实现持久化
- 不具备分布式能力，需要自己实现高可用
时间轮实现
时间轮是一种环形的数据结构，分成多个格。
每个格代表一段时间，时间越短，精度越高。
每个格上用一个链表保存在该格的过期任务。
指针随着时间一格一格转动，并执行相应格子中的到期任务。
名词解释：
- 时间格：环形结构中用于存放延迟任务的区块
- 指针：指向当前操作的时间格，代表当前时间
- 格数：时间轮中时间格的个数
- 间隔：每个时间格之间的间隔，代表时间轮能达到的精度
- 总间隔：当前时间轮总间隔，等于格数*间隔，代表时间轮能表达的时间范围
单表时间轮
以上图为例，假设一个格子是1秒，则整个时间轮能表示的时间段为8s， 如果当前指针指向2，此时需要调度一个3s后执行的任务，需要放到第5个格子(2 3)中，指针再转3次就可以执行了。
单表时间轮存在的问题是：
格子的数量有限 所能代表的时间有限 当要存放一个10s后到期的任务怎么办？这会引起时间轮溢出。
有个办法是把轮次信息也保存到时间格链表的任务上。
如果任务要在10s后执行，算出轮次10/8 round等1，格子10%8等于2，所以放入第二格。
检查过期任务时应当只执行round为0的任务，链表中其他任务的round减1。
带轮次单表时间轮存在的问题是：
如果任务的时间跨度很大，数量很大，单层时间轮会造成任务的round很大，单个格子的链表很长，每次检查的量很大，会做很多无效的检查。怎么办?
分层时间轮
过期任务一定是在底层轮中被执行的，其他时间轮中的任务在接近过期时会不断的降级进入低一层的时间轮中。
分层时间轮中每个轮都有自己的格数和间隔设置，当最低层的时间轮转一轮时，高一层的时间轮就转一个格子。
分层时间轮大大增加了可表示的时间范围，同时减少了空间占用。
举个例子：
上图的分层时间轮可表达8 8 8=512s的时间范围，如果用单表时间轮可能需要512个格子， 而分层时间轮只要8 8 8=24个格子，如果要设计一个时间范围是1天的分层时间轮，三个轮的格子分别用24、60、60即可。
工作原理：
时间轮指针转动有两种方式：
- 根据自己的间隔转动（秒钟轮1秒转1格；分钟轮1分钟转1格；时钟轮1小时转1格）
- 通过下层时间轮推动（秒钟轮转1圈，分钟轮转1格；分钟轮转1圈，时钟轮转1格）
指针转到特定格子时有两种处理方式：
- 如果是底层轮，指针指向格子中链表上的元素均表示过期
- 如果是其他轮，将格子上的任务移动到精度细一级的时间轮上，比如时钟轮的任务移动到分钟轮上
举个例子：
- 添加1个5s后执行的任务
- 算出任务应该放在秒钟轮的第5个格子
- 在秒钟轮指针进行5次转动后任务会被执行
- 添加一个50s后执行的任务
- 算出该任务的延迟时间已经溢出秒钟轮
- 50/8=6 所以该任务会被保存在分钟轮的第6个格子
- 在秒钟轮走了6圈(6*8s=48s)之后，分钟轮的指针指向第6个格子
- 此时该格子中的任务会被降级到秒钟轮，并根据50%8=2，任务会被移动到秒钟轮的第2个格子
- 在秒钟轮指针又进行2次转动后(50s)任务会被执行
- 添加一个250s后执行的任务
- 算出该任务的延迟时间已经溢出分钟轮
- 250/8/8=3 所以该任务会被保存在时钟轮的第3个格子
- 在分钟轮走了3圈(3*64s=192s)之后，时钟轮的指针指向第3个格子
- 此时该格子中的任务会被降级到分钟轮，并根据(250-192)/8=7，任务会被移动到分钟轮的第7个格子
- 在秒钟轮走了7圈(7*8s=56s)之后，分钟轮的指针指向第7个格子
- 此时该格子中的任务会被降级到秒钟轮，并根据(250-192-56)=2，任务会被移动到秒钟轮的第2个格子
- 在秒钟轮指针又进行2次转动后任务会被执行
优点：
- 高性能（插入任务、删除任务的时间复杂度均为O(1)，DelayQueue由于涉及到排序，插入和移除的复杂度是O(logn)）
缺点：
- 数据是保存在内存，需要自己实现持久化
- 不具备分布式能力，需要自己实现高可用
- 延迟任务过期时间受时间轮总间隔限制
对于超出范围的任务可放在一个缓冲区中(可用队列、redis或数据库实现)，等最高时间轮转到下一格子就从缓冲中取出符合范围的任务落到时间轮中。
比如：
- 添加一个600s后执行的任务A
- 算出该任务的延迟时间已经溢出时间轮
- 所以任务被保存到缓冲队列中
- 在时钟轮走了1格之后，会从缓冲队列中取满足范围的任务落到时间轮中
- 缓冲队列中的所有任务延迟时间均需减去64s，任务A减去64s后是536s，依然大于时间轮范围，所以不会被移出队列
- 在时钟轮又走了1格之后，任务A减去64s是536-64=472s，在时间轮范围内，会被落入时钟轮
之前的设计（DB/DelayQueue/ZooKeeper）
调度系统提供任务操作接口供业务系统提交任务、取消任务、反馈执行结果等。
针对dubbo调用 将任务抽象成JobCallbackService接口，由业务系统实现并注册成服务。
整体架构
数据库：
- 负责保存所有的任务数据
内存队列：
- 实际为DelayQueue，延迟任务精确触发的机制由它保证
- 只存储未来N分钟内过期且最多1000个任务
ZooKeeper：
- 管理整个调度集群
- 存储调度节点信息
- 存储节点分片信息
主节点：
- 有新的节点上下线时对数据重新分片
调度节点：
- 提供dubbo、http接口供业务系统调用，用于提交任务、取消任务、反馈执行结果等
- 从ZK注册中心获取当前节点的分片信息，再从数据库拉取即将过期的数据放到DelayQueue
- 调用业务系统注册的回调服务接口，发起调度请求
- 接收业务系统的反馈结果，更新执行结果，移除任务或发起重试
业务系统：
- 作为被调度的服务需要实现回调接口JobCallbackService，并注册为dubbo服务提供者
- 在需要延迟任务的场景调用调度系统接口操作任务
数据库设计
表说明
- job_callback_service:服务配置表，配置业务回调服务，包括服务协议、回调服务、重试次数
- job_delay_task:延迟任务表，用于存储延迟任务，包括任务分片号、回调服务、调用总次数、失败数、任务状态、回调参数等
- job_delay_task_execlog:延迟任务执行表，记录调度系统发起的每一次回调
- job_delay_task_backlog:延迟任务调度结果表，记录任务最终状态等信息
主从切换
利用ZooKeeper临时序列节点特性，序号最小的节点为主节点，其他节点为从节点。
主节点监听集群状态，集群状态发生变化时重新分片。
从节点监听序号比它小的兄弟节点，兄弟节点发生变化重新寻找和建立监听关系。
数据分片
任务状态
- delay：延迟任务提交后的初始状态
- ready：过期时间已到，消息推入就绪队列的状态
- running：业务订阅消息，收到消息开始处理的状态
- finished：业务处理成功
- failed：业务处理失败
主要流程
服务加载
- 从DB读取服务配置
- 根据配置动态构造Consumer对象并添加到Spring容器中
提交任务
- 业务系统通过dubbo或http接口提交任务
- 判断任务过期时间是否在一个扫描周期内
- 如果是，
- 设置分片号(从当前节点所负责的分片随机获取)
- 添加到内存队列
- 任务保存到job_delay_task表
- 如果否
- 设置分片号(根据分片总数和随机算法算出分片号)
- 任务保存到delay_task表
定时器
- 由一个线程管理
- 根据配置的扫描间隔设置定时器的执行周期
- 根据当前时间和扫描间隔算出该时段的过期时间X-Delay
- 从DB获取过期时间在X-Delay之前的所有任务，并放到DelayQueue
调度任务
- 由一个线程池管理
- 所有线程都阻塞在DelayQueue的方法take
- take到任务，从DB中获取任务，判断是否存在
- 如果不在，什么也不做(任务已执行成功或已被删除)
- 如果存在，判断调用次数是否超过设置
- 如果不超
- 调用业务回调服务
- 从任务中取出调用的服务配置
- 从容器中获取对应的Consumer对象
- 异步调用业务回调服务
- 设置下次重试时间，记录调用日志job_delay_task_execlog
- 如果超过，将任务转移到job_delay_task_backlog
任务反馈
- 更新任务调用结果
优点
- 功能全面，高可用、易伸缩、可重试
缺点
- 略微复杂
- 需要将服务配置动态生成为Consumer对象
- 增加新的服务需要通知所有调度节点刷新
- 存在一定的耦合性(直接调用业务服务，协议耦合)，如果接入系统是thrift协议呢？
- 需要处理任务的重试
- 调度系统直接回调业务服务，如果业务服务不可用可能会造成盲目重试，不能很好的控制流量(调度系统不知道业务服务的处理能力)
如果引入MQ，使用MQ来解耦服务调用的协议，保证任务的重试，并由消费方根据自己的处理能力控制流量会不会更好呢？
另一种方案（DB/DelayQueue/ZooKeeper/MQ）
整体架构
数据库设计
主要流程
调度任务
- 由一个线程池管理
- 所有线程都阻塞在DelayQueue的take方法
- take到任务，从DB中获取任务，判断是否存在
- 如果不在，什么也不做(任务已执行成功或已被删除)
- 如果存在，将任务转移到job_delay_task_execlog；往消息队列投递消息
缺点
需要业务系统依赖于MQ
