TimingWheel

    1. Timer：定时器维护一个优先队列，到时间点执行，然后把需要执行的 task 存储在 map 中
    2. collection 中的 timingWheel ，维护一个存放任务组的数组，每一个槽都维护一个存储task的双向链表。开始执行时，计时器每隔指定时间执行一个槽里面的tasks。

    方案2把维护task从 优先队列 O(nlog(n)) 降到 双向链表 O(1)，而执行task也只要轮询一个时间点的tasks O(N)，不需要像优先队列，放入和删除元素 O(nlog(n))。

    首先我们先来在 collection 的 cache 中关于 timingWheel 的使用：

    这是 cache 初始化中也同时初始化 timingWheel 做key的过期处理，参数依次代表：

    • interval：时间划分刻度
    • numSlots：时间槽
    • execute：时间点执行函数

    在 cache 中执行函数则是 删除过期key，而这个过期则由 timingWheel 来控制推进时间。

    接下来，就通过 cache 对 timingWheel 的使用来认识。

    1. // 真正做初始化
    2. func newTimingWheelWithClock(interval time.Duration, numSlots int, execute Execute, ticker timex.Ticker) (
    3.     *TimingWheel, error) {
    4.     tw := &TimingWheel{
    5.         interval:      interval,                     // 单个时间格时间间隔
    6.         ticker:        ticker,                       // 定时器，做时间推动，以interval为单位推进
    7.         slots:         make([]*list.List, numSlots), // 时间轮
    8.         timers:        NewSafeMap(),                 // 存储task{key, value}的map [执行execute所需要的参数]
    9.         tickedPos:     numSlots - 1,                 // at previous virtual circle
    10.         execute:       execute,                      // 执行函数
    11.         numSlots:      numSlots,                     // 初始化 slots num
    12.         setChannel:    make(chan timingEntry),       // 以下几个channel是做task传递的
    13.         moveChannel:   make(chan baseEntry),
    14.         removeChannel: make(chan interface{}),
    15.         drainChannel:  make(chan func(key, value interface{})),
    16.         stopChannel:   make(chan lang.PlaceholderType),
    17.     }
    18.     // 把 slot 中存储的 list 全部准备好
    19.     tw.initSlots()
    20.     // 开启异步协程，使用 channel 来做task通信和传递
    21.     go tw.run()
    23.     return tw, nil
    24. }

    以上比较直观展示 timingWheel 的 “时间轮”，后面会围绕这张图解释其中推进的细节。

    1. func (tw *TimingWheel) run() {
    2.     for {
    3.         select {
    4.         case <-tw.ticker.Chan():
    5.             tw.onTick()
    6.       // add task 会往 setChannel 输入task
    8.             tw.setTask(&task)
    9.         ...
    10.         }
    11.     }
    12. }

    可以看出，在初始化的时候就开始了 timer 执行，并以interval时间段转动，然后底层不停的获取来自 slot 中的 list 的task，交给 execute 执行。

    3bbddc1ebb79455da91dfcf3da6bc72f~tplv-k3u1fbpfcp-zoom-1.image.png

    紧接着就是设置 cache key ：

    1. 先看在 data map 中有没有存在这个key
    2. 存在，则更新 expire -> MoveTimer()
    3. 第一次设置key -> SetTimer()

    所以对于 timingWheel 的使用上就清晰了，开发者根据需求可以 add 或是 update。

    同时我们跟源码进去会发现：SetTimer() MoveTimer() 都是将task输送到channel，由 run() 中开启的协程不断取出 channel 的task操作。

    1. func (tw *TimingWheel) moveTask(task baseEntry) {
    2.     // timers: Map => 通过key获取 [positionEntry「pos, task」]
    3.     val, ok := tw.timers.Get(task.key)
    4.     if !ok {
    5.         return
    6.     }
    8.     timer := val.(*positionEntry)
    9.   // {delay < interval} => 延迟时间比一个时间格间隔还小，没有更小的刻度，说明任务应该立即执行
    10.     if task.delay < tw.interval {
    11.         threading.GoSafe(func() {
    12.             tw.execute(timer.item.key, timer.item.value)
    13.         })
    14.         return
    15.     }
    16.     // 如果 > interval，则通过 延迟时间delay 计算其出时间轮中的 new pos, circle
    17.     pos, circle := tw.getPositionAndCircle(task.delay)
    18.     if pos >= timer.pos {
    19.         timer.item.circle = circle
    20.     // 记录前后的移动offset。为了后面过程重新入队
    21.         timer.item.diff = pos - timer.pos
    22.     } else if circle > 0 {
    23.         // 转移到下一层，将 circle 转换为 diff 一部分
    24.         timer.item.circle = circle
    25.         // 因为是一个数组，要加上 numSlots [也就是相当于要走到下一层]
    27.     } else {
    28.         // 如果 offset 提前了，此时 task 也还在第一层
    29.         // 标记删除老的 task，并重新入队，等待被执行
    30.         timer.item.removed = true
    31.         newItem := &timingEntry{
    32.             baseEntry: task,
    33.             value:     timer.item.value,
    34.         }
    35.         tw.slots[pos].PushBack(newItem)
    36.         tw.setTimerPosition(pos, newItem)
    37.     }
    38. }

    以上过程有以下几种情况：

    • delay < interval：因为 < 单个时间精度，表示这个任务已经过期，需要马上执行
    • 针对改变的 delay：
      • new >= old：<newPos, newCircle, diff>
      • newCircle > 0：计算diff，并将 circle 转换为 下一层，故diff + numslots
      • 如果只是单纯延迟时间缩短，则将老的task标记删除，重新加入list，等待下一轮loop被execute

    之前在初始化中，run() 中定时器的不断推进，推进的过程主要就是把 list中的 task 传给执行的 execute func。我们从定时器的执行开始看：

    1. // 定时器 「每隔 interval 会执行一次」
    2. func (tw *TimingWheel) onTick() {
    3.   // 每次执行更新一下当前执行 tick 位置
    4.     tw.tickedPos = (tw.tickedPos + 1) % tw.numSlots
    5.   // 获取此时 tick位置 中的存储task的双向链表
    6.     l := tw.slots[tw.tickedPos]
    7.     tw.scanAndRunTasks(l)
    8. }

    紧接着是如何去执行 execute：

    具体的分支情况在注释中说明了，在看的时候可以和前面的 moveTask() 结合起来，其中 circle 下降，diff 的计算是关联两个函数的重点。

    至于 diff 计算就涉及到 pos, circle 的计算：

    1. // interval: 4min, d: 60min, numSlots: 16, tickedPos = 15
    2. // step = 15, pos = 14, circle = 0
    3. func (tw *TimingWheel) getPositionAndCircle(d time.Duration) (pos int, circle int) {
    4.     steps := int(d / tw.interval)
    5.     pos = (tw.tickedPos + steps) % tw.numSlots
    6.     circle = (steps - 1) / tw.numSlots
    7.     return
    8. }

    上面的过程可以简化成下面：

    1. steps = d / interval
    2. circle = (step - 1) / numSlots

    总结

    timingWheel 靠定时器推动，时间前进的同时会取出当前时间格中 list「双向链表」的task，传递到 execute 中执行。

    在 中还有很多实用的组件工具，用好工具对于提升服务性能和开发效率都有很大的帮助，希望本篇文章能给大家带来一些收获。