1 定时任务
Netty、Quartz、Kafka 以及 Linux 都有定时任务功能。
- 常规的JDK 的 java.util.Timer 和 DelayedQueue 等工具类,可实现简单的定时任务,底层用的是堆数据结构,存取复杂度都是 O(nlog(n)),无法支撑海量定时任务。
- 而在定时任务量大、性能要求高的场景,为将任务存取及取消操作时间复杂度降为 O(1),会使用时间轮方案。
2 时间轮模型及其应用
一种高效批量管理定时任务的调度模型。时间轮一般会实现成一个环形结构,类似一个时钟,分为很多槽,一个槽代表一个时间间隔,每个槽使用双向链表存储定时任务。指针周期性地跳动,跳动到一个槽位,就执行该槽位的定时任务。
- Hashed Timing Wheel 结构示意图
适用场景
- 故障恢复
- 流量控制
- 调度算法
- 控制网络中的数据包生命周期
计时器维护代价高,如果
- 处理器在每个时钟滴答声中都会中断
- 使用精细粒度计时器
- 未完成的计时器很多
需要高效的定时器算法以减少总体中断的开销。
单层时间轮的容量和精度都是有限的,对于精度要求特别高、时间跨度特别大或是海量定时任务需要调度的场景,通常会使用多级时间轮以及持久化存储与时间轮结合的方案。
模型和性能指标
模型中的规则
- 客户端调用:
- START_TIMER(时间间隔,Request_ID,Expiry_Action)
- STOP_TIMER(Request_ID)
- 计时器tick调用:
- PER_TICK_BOOKKEEPING
- EXPIRY_PROCESSING
性能指标
- 空间 数据结构使用的内存
- 延迟 开始和结束上述任何例程所需的时间
3 Dubbo的时间轮结构
Dubbo 的时间轮实现位于 dubbo-common 模块的 org.apache.dubbo.common.timer 包中,下面我们就来分析时间轮涉及的核心接口和实现。
核心接口
TimerTask
在 Dubbo 中,所有定时任务都要实现 TimerTask 接口。只定义了一个 run() 方法,入参是一个 Timeout 接口对象。
Timeout
Timeout 对象与 TimerTask 对象一一对应,类似线程池返回的 Future 对象与提交到线程池中的任务对象之间的关系。 通过 Timeout 对象,不仅可以查看定时任务的状态,还可以操作定时任务(例如取消关联的定时任务)。
- Timeout 接口中的方法:
Timer 接口定义了定时器的基本行为,核心是 newTimeout()
:提交一个定时任务(TimerTask)并返回关联的 Timeout 对象,类似于向线程池提交任务。
HashedWheelTimeout
HashedWheelTimeout 是 Timeout 接口的唯一实现,是 HashedWheelTimer 的内部类。HashedWheelTimeout 扮演了两个角色:
- 时间轮中双向链表的节点,即定时任务 TimerTask 在 HashedWheelTimer 中的容器
- 定时任务 TimerTask 提交到 HashedWheelTimer 之后返回的句柄(Handle),用于在时间轮外部查看和控制定时任务
核心字段
- prev、next。通过双向链表被用来在HashedWheelTimerBucket链timeouts(定时任务),由于只在WorkerThread上行动,没有必要进行同步/volatile。
- task,实际被调度的任务
- deadline,定时任务执行的时间。在创建 HashedWheelTimeout 时指定
计算公式:
currentTime(创建 HashedWheelTimeout 的时间) delay(任务延迟时间) - startTime(HashedWheelTimer 的启动时间)
,ns
- state,定时任务当前所处状态
- remainingRounds,当前任务剩余的时钟周期数。时间轮所能表示的时间长度有限,在任务到期时间与当前时刻的时间差,超过时间轮单圈能表示时长,就出现套圈,需要该字段值表示剩余的时钟周期。
核心API
- isCancelled()
- isExpired()
- state() 检查当前 HashedWheelTimeout 状态
- cancel() 方法
- expire() 方法
- remove()
HashedWheelBucket
时间轮中的一个槽。
时间轮中的槽实际上就是一个用于缓存和管理双向链表的容器,双向链表中的每一个节点就是一个 HashedWheelTimeout
对象,也就关联了一个 TimerTask
定时任务。
HashedWheelBucket 持有双向链表的首尾两个节点 - head 和 tail,再加上每个 HashedWheelTimeout 节点均持有前驱和后继引用,即可正、逆向遍历整个链表。
核心API
- addTimeout()
- pollTimeout()
- remove() 从双向链表中移除指定的 HashedWheelTimeout 节点。
- clearTimeouts() 循环调用 pollTimeout() 方法处理整个双向链表,并返回所有未超时或者未被取消的任务。
- expireTimeouts() 遍历双向链表中的全部 HashedWheelTimeout 节点。 在处理到期的定时任务时,会通过 remove() 方法取出,并调用其 expire() 方法执行;对于已取消的任务,通过 remove() 方法取出后直接丢弃;对于未到期的任务,会将 remainingRounds 字段(剩余时钟周期数)减一。
HashedWheelTimer
Timer
接口的实现,通过时间轮算法实现了一个定时器。
职能
根据当前时间轮指针选定对应 HashedWheelBucket
槽,从链表头部开始迭代,计算每个 HashedWheelTimeout
定时任务:
- 属于当前时钟周期则取出运行
- 不属于则将其剩余的时钟周期数减一
核心域
- workerState
时间轮当前所处状态,三个可选值,由
AtomicIntegerFieldUpdater
实现其原子地修改。
- startTime 当前时间轮的启动时间,提交到该时间轮的定时任务的 deadline 字段值均以该时间戳为起点进行计算。
- wheel 时间轮环形队列,每个元素都是一个槽。当指定时间轮槽数为 n 时,会向上取最靠近 n 的 2 次幂值
- timeouts、cancelledTimeouts
HashedWheelTimer 会在处理 HashedWheelBucket 的双向链表前,先处理这俩队列的数据:
- timeouts 队列 缓冲外部提交时间轮中的定时任务
- cancelledTimeouts 队列 暂存取消的定时任务
- tick
位于
HashedWheelTimer$Worker
,时间轮的指针,步长为 1 的单调递增计数器
- mask
掩码,
mask = wheel.length - 1
,执行ticks & mask
便能定位到对应的时钟槽
- ticksDuration 时间指针每次加 1 所代表的实际时间,单位为纳秒。
- pendingTimeouts 当前时间轮剩余的定时任务总数。
- workerThread 时间轮内部真正执行定时任务的线程。
- worker 真正执行定时任务的逻辑封装这个 Runnable 对象中。
newTimeout()
提交定时任务,在定时任务进入到 timeouts 队列之前会先调用 start() 方法启动时间轮,其中会完成下面两个关键步骤:
- 确定时间轮的 startTime 字段
- 启动 workerThread 线程,开始执行 worker 任务。
之后根据 startTime 计算该定时任务的 deadline,最后才能将定时任务封装成 HashedWheelTimeout
并添加到 timeouts
队列。
4 时间轮指针一次转动的执行流程
HashedWheelTimer$Worker.run()
:
- 时间轮指针转动,时间轮周期开始
- 清理用户主动取消的定时任务,这些定时任务在用户取消时,记录到
cancelledTimeouts
队列中。在每次指针转动的时候,时间轮都会清理该队列 - 将缓存在 timeouts 队列中的定时任务转移到时间轮中对应的槽中
- 根据当前指针定位对应槽,处理该槽位的双向链表中的定时任务
- 检测时间轮的状态。如果时间轮处于运行状态,则循环执行上述步骤,不断执行定时任务。如果时间轮处于停止状态,则执行下面的步骤获取到未被执行的定时任务并加入
unprocessedTimeouts
队列:遍历时间轮中每个槽位,并调用clearTimeouts
() 方法;对 timeouts 队列中未被加入槽中循环调用 poll() - 最后再次清理 cancelledTimeouts 队列中用户主动取消的定时任务。
5 定时任务应用
并不直接用于周期性操作,而是只向时间轮提交执行单次的定时任务,在上一次任务执行完成的时候,调用 newTimeout() 方法再次提交当前任务,这样就会在下个周期执行该任务。即使在任务执行过程中出现了 GC、I/O 阻塞等情况,导致任务延迟或卡住,也不会有同样的任务源源不断地提交进来,导致任务堆积。
Dubbo 时间轮应用主要在如下方面:
- 失败重试, 例如,Provider 向注册中心进行注册失败时的重试操作,或是 Consumer 向注册中心订阅时的失败重试等
- 周期性定时任务, 例如,定期发送心跳请求,请求超时的处理,或是网络连接断开后的重连机制
参考
- https://zhuanlan.zhihu.com/p/32906730