调用dubbo接口返回timeout_Dubbo中的时间轮(TimeWheel)算法应用

  点击上方“JavaEdge”&＃xff0c;关注公众号

1 定时任务

Netty、Quartz、Kafka 以及 Linux 都有定时任务功能。

• JDK 自带的 java.util.Timer 和 DelayedQueue 可实现简单的定时任务&＃xff0c;底层用的是堆&＃xff0c;存取复杂度都是 O(nlog(n))&＃xff0c;但无法支撑海量定时任务。

• 在任务量大、性能要求高的场景&＃xff0c;为了将任务存取及取消操作时间复杂度降为 O(1)&＃xff0c;会采用时间轮算法。

2 时间轮模型及其应用

一种高效批量管理定时任务的调度模型。一般会实现成一个环形结构&＃xff0c;类似一个时钟&＃xff0c;分为很多槽&＃xff0c;一个槽代表一个时间间隔&＃xff0c;每个槽使用双向链表存储定时任务。

指针周期性跳动&＃xff0c;跳动到一个槽位&＃xff0c;就执行该槽位的定时任务。

• Hashed Timing Wheel 结构示意图

适用场景

• 故障恢复

• 流量控制

• 调度算法

• 控制网络中的数据包生命周期

计时器维护代价高&＃xff0c;如果

• 处理器在每个时钟滴答声中都会中断

• 使用精细粒度计时器

• 未完成的计时器很多

需要高效的定时器算法以减少总体中断的开销。

单层时间轮的容量和精度都是有限的&＃xff0c;对于精度要求特别高、时间跨度特别大或是海量定时任务需要调度的场景&＃xff0c;通常会使用多级时间轮以及持久化存储与时间轮结合的方案。

模型和性能指标

模型中的规则

• 客户端调用&＃xff1a;

  • START_TIMER(时间间隔&＃xff0c;Request_ID&＃xff0c;Expiry_Action)

  • STOP_TIMER(Request_ID)

• 计时器tick调用&＃xff1a;

  • PER_TICK_BOOKKEEPING

  • EXPIRY_PROCESSING

性能指标

• 空间
  数据结构使用的内存

• 延迟
  开始和结束上述任何例程所需的时间

3 Dubbo的时间轮结构

Dubbo 时间轮实现位于 dubbo-common 模块的 org.apache.dubbo.common.timer 包&＃xff0c;下面我们就来分析时间轮涉及的核心接口和实现。

核心接口

TimerTask

在 Dubbo 中&＃xff0c;所有定时任务都要实现 TimerTask 接口。只定义了一个 run() 方法&＃xff0c;入参是一个 Timeout 接口对象。

Timeout

Timeout 对象与 TimerTask 对象一一对应&＃xff0c;类似线程池返回的 Future 对象与提交到线程池中的任务对象之间的关系。
通过 Timeout 对象&＃xff0c;不仅可以查看定时任务的状态&＃xff0c;还可以操作定时任务(例如取消关联的定时任务)。

• Timeout 接口中的方法&＃xff1a;

Timer 接口定义了定时器的基本行为&＃xff0c;核心是 newTimeout() &＃xff1a;提交一个定时任务(TimerTask)并返回关联的 Timeout 对象&＃xff0c;类似于向线程池提交任务。

HashedWheelTimeout

HashedWheelTimeout 是 Timeout 接口的唯一实现&＃xff0c;是 HashedWheelTimer 的内部类。HashedWheelTimeout 扮演了两个角色&＃xff1a;

• 时间轮中双向链表的节点&＃xff0c;即定时任务 TimerTask 在 HashedWheelTimer 中的容器

• 定时任务 TimerTask 提交到 HashedWheelTimer 之后返回的句柄(Handle)&＃xff0c;用于在时间轮外部查看和控制定时任务

核心字段

• prev、next。通过双向链表被用来在HashedWheelTimerBucket链timeouts(定时任务)&＃xff0c;由于只在WorkerThread上行动&＃xff0c;没有必要进行同步/volatile。

• task&＃xff0c;实际被调度的任务

• deadline&＃xff0c;定时任务执行的时间。在创建 HashedWheelTimeout 时指定
  计算公式&＃xff1a;currentTime(创建 HashedWheelTimeout 的时间) &＃43; delay(任务延迟时间) - startTime(HashedWheelTimer 的启动时间)&＃xff0c;ns

• state&＃xff0c;定时任务当前所处状态

  • 可选状态&＃xff1a;

  • STATE_UPDATER 用于实现 state 状态变更的原子性。

• remainingRounds&＃xff0c;当前任务剩余的时钟周期数。时间轮所能表示的时间长度有限&＃xff0c;在任务到期时间与当前时刻的时间差&＃xff0c;超过时间轮单圈能表示时长&＃xff0c;就出现套圈&＃xff0c;需要该字段值表示剩余的时钟周期。

核心API

• isCancelled()

• isExpired()

• state()
  检查当前 HashedWheelTimeout 状态

• cancel() 方法

• expire() 方法

• remove()

HashedWheelBucket

时间轮中的一个槽。
时间轮中的槽实际上就是一个用于缓存和管理双向链表的容器&＃xff0c;双向链表中的每一个节点就是一个 HashedWheelTimeout 对象&＃xff0c;也就关联了一个 TimerTask 定时任务。

HashedWheelBucket 持有双向链表的首尾两个节点 - head 和 tail&＃xff0c;再加上每个 HashedWheelTimeout 节点均持有前驱和后继引用&＃xff0c;即可正、逆向遍历整个链表。

核心API

• addTimeout()

• pollTimeout()

• remove()
  从双向链表中移除指定的 HashedWheelTimeout 节点。

• clearTimeouts()
  循环调用 pollTimeout() 方法处理整个双向链表&＃xff0c;并返回所有未超时或者未被取消的任务。

• expireTimeouts()
  遍历双向链表中的全部 HashedWheelTimeout 节点。在处理到期的定时任务时&＃xff0c;会通过 remove() 方法取出&＃xff0c;并调用其 expire() 方法执行&＃xff1b;对于已取消的任务&＃xff0c;通过 remove() 方法取出后直接丢弃&＃xff1b;对于未到期的任务&＃xff0c;会将 remainingRounds 字段(剩余时钟周期数)减一。

HashedWheelTimer

Timer 接口的实现&＃xff0c;通过时间轮算法实现了一个定时器。

职能

根据当前时间轮指针选定对应 HashedWheelBucket 槽&＃xff0c;从链表头部开始迭代&＃xff0c;计算每个 HashedWheelTimeout 定时任务&＃xff1a;

• 属于当前时钟周期则取出运行

• 不属于则将其剩余的时钟周期数减一

核心域

• workerState
  时间轮当前所处状态&＃xff0c;三个可选值&＃xff0c;由 AtomicIntegerFieldUpdater 实现其原子地修改。

• startTime
  当前时间轮的启动时间&＃xff0c;提交到该时间轮的定时任务的 deadline 字段值均以该时间戳为起点进行计算。

• wheel
  时间轮环形队列&＃xff0c;每个元素都是一个槽。当指定时间轮槽数为 n 时&＃xff0c;会向上取最靠近 n 的 2 次幂值

• timeouts、cancelledTimeouts
  HashedWheelTimer 会在处理 HashedWheelBucket 的双向链表前&＃xff0c;先处理这俩队列的数据&＃xff1a;

  • timeouts 队列
    缓冲外部提交时间轮中的定时任务

  • cancelledTimeouts 队列
    暂存取消的定时任务

• tick
  位于 HashedWheelTimer$Worker &＃xff0c;时间轮的指针&＃xff0c;步长为 1 的单调递增计数器

• mask
  掩码&＃xff0c; mask &＃61; wheel.length - 1&＃xff0c;执行 ticks & mask 便能定位到对应的时钟槽

• ticksDuration
  时间指针每次加 1 所代表的实际时间&＃xff0c;单位为纳秒。

• pendingTimeouts
  当前时间轮剩余的定时任务总数。

• workerThread
  时间轮内部真正执行定时任务的线程。

• worker
  真正执行定时任务的逻辑封装这个 Runnable 对象中。

newTimeout()

提交定时任务&＃xff0c;在定时任务进入到 timeouts 队列之前会先调用 start() 方法启动时间轮&＃xff0c;其中会完成下面两个关键步骤&＃xff1a;

1. 确定时间轮的 startTime 字段

2. 启动 workerThread 线程&＃xff0c;开始执行 worker 任务。

之后根据 startTime 计算该定时任务的 deadline&＃xff0c;最后才能将定时任务封装成 HashedWheelTimeout 并添加到 timeouts 队列。

4 时间轮指针一次转动的执行流程

HashedWheelTimer$Worker.run()&＃xff1a;

1. 时间轮指针转动&＃xff0c;时间轮周期开始

2. 清理用户主动取消的定时任务&＃xff0c;这些定时任务在用户取消时&＃xff0c;记录到 cancelledTimeouts 队列中。在每次指针转动的时候&＃xff0c;时间轮都会清理该队列

3. 将缓存在 timeouts 队列中的定时任务转移到时间轮中对应的槽中

4. 根据当前指针定位对应槽&＃xff0c;处理该槽位的双向链表中的定时任务

5. 检测时间轮的状态。如果时间轮处于运行状态&＃xff0c;则循环执行上述步骤&＃xff0c;不断执行定时任务。如果时间轮处于停止状态&＃xff0c;则执行下面的步骤获取到未被执行的定时任务并加入 unprocessedTimeouts 队列&＃xff1a;遍历时间轮中每个槽位&＃xff0c;并调用 clearTimeouts() 方法&＃xff1b;对 timeouts 队列中未被加入槽中循环调用 poll()

6. 最后再次清理 cancelledTimeouts 队列中用户主动取消的定时任务。

5 定时任务应用

并不直接用于周期性操作&＃xff0c;而是只向时间轮提交执行单次的定时任务&＃xff0c;在上一次任务执行完成的时候&＃xff0c;调用 newTimeout() 方法再次提交当前任务&＃xff0c;这样就会在下个周期执行该任务。即使在任务执行过程中出现了 GC、I/O 阻塞等情况&＃xff0c;导致任务延迟或卡住&＃xff0c;也不会有同样的任务源源不断地提交进来&＃xff0c;导致任务堆积。

Dubbo 时间轮应用主要在如下方面&＃xff1a;

1. 失败重试&＃xff0c; 例如&＃xff0c;Provider 向注册中心进行注册失败时的重试操作&＃xff0c;或是 Consumer 向注册中心订阅时的失败重试等

2. 周期性定时任务&＃xff0c; 例如&＃xff0c;定期发送心跳请求&＃xff0c;请求超时的处理&＃xff0c;或是网络连接断开后的重连机制

参考

• https://zhuanlan.zhihu.com/p/32906730

往期推荐

DDD领域驱动实战-子域/核心域等核心概念

实战DDD领域驱动设计之限界上下文

DDD领域驱动设计实战(三)- 理解实体

DDD领域驱动设计实战(四)- 理解值对象

架构设计之高可扩展性

目前交流群已有 800&＃43;人&＃xff0c;旨在促进技术交流&＃xff0c;可关注公众号添加笔者微信邀请进群

喜欢文章&＃xff0c;点个“在看、点赞、分享”素质三连支持一下~