一个零碎中存在着大量的调度工作，同时调度工作存在工夫的滞后性，而大量的调度工作如果每一个都应用本人的调度器来治理工作的生命周期的话，节约cpu的资源而且很低效。

本文来介绍 go-zero 中 提早操作，它可能让开发者调度多个工作时，只需关注具体的业务执行函数和执行工夫「立刻或者提早」。而 提早操作，通常能够采纳两个计划：

  1. Timer：定时器保护一个优先队列，到工夫点执行，而后把须要执行的 task 存储在 map 中
  2. collection 中的 timingWheel ，保护一个寄存工作组的数组，每一个槽都保护一个存储task的双向链表。开始执行时，计时器每隔指定工夫执行一个槽外面的tasks。

计划2把保护task从 优先队列 O(nlog(n)) 降到 双向链表 O(1)，而执行task也只有轮询一个工夫点的tasks O(N)，不须要像优先队列，放入和删除元素 O(nlog(n))。

咱们先看看 go-zero 中本人对 timingWheel 的应用 ：

cache 中的 timingWheel

首先咱们先来在 collection 的 cache 中对于 timingWheel 的应用：

timingWheel, err := NewTimingWheel(time.Second, slots, func(k, v interface{}) {  key, ok := k.(string)  if !ok {    return  }  cache.Del(key)})if err != nil {  return nil, err}cache.timingWheel = timingWheel

这是 cache 初始化中也同时初始化 timingWheel 做key的过期解决，参数顺次代表：

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

在 cache 中执行函数则是 删除过期key，而这个过期则由 timingWheel 来管制推动工夫。

接下来，就通过 cache 对 timingWheel 的应用来意识。

初始化

// 真正做初始化func newTimingWheelWithClock(interval time.Duration, numSlots int, execute Execute, ticker timex.Ticker) (    *TimingWheel, error) {    tw := &TimingWheel{        interval:      interval,                     // 单个工夫格工夫距离        ticker:        ticker,                       // 定时器，做工夫推动，以interval为单位推动        slots:         make([]*list.List, numSlots), // 工夫轮        timers:        NewSafeMap(),                 // 存储task{key, value}的map [执行execute所须要的参数]        tickedPos:     numSlots - 1,                 // at previous virtual circle        execute:       execute,                      // 执行函数        numSlots:      numSlots,                     // 初始化 slots num        setChannel:    make(chan timingEntry),       // 以下几个channel是做task传递的        moveChannel:   make(chan baseEntry),        removeChannel: make(chan interface{}),        drainChannel:  make(chan func(key, value interface{})),        stopChannel:   make(chan lang.PlaceholderType),    }    // 把 slot 中存储的 list 全副筹备好    tw.initSlots()    // 开启异步协程，应用 channel 来做task通信和传递    go tw.run()    return tw, nil}

以上比拟直观展现 timingWheel 的 “工夫轮”，前面会围绕这张图解释其中推动的细节。

go tw.run() 开一个协程做工夫推动：

func (tw *TimingWheel) run() {    for {        select {      // 定时器做工夫推动 -> scanAndRunTasks()        case <-tw.ticker.Chan():            tw.onTick()      // add task 会往 setChannel 输出task        case task := <-tw.setChannel:            tw.setTask(&task)        ...        }    }}

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

Task Operation

紧接着就是设置 cache key ：

func (c *Cache) Set(key string, value interface{}) {    c.lock.Lock()    _, ok := c.data[key]    c.data[key] = value    c.lruCache.add(key)    c.lock.Unlock()    expiry := c.unstableExpiry.AroundDuration(c.expire)    if ok {        c.timingWheel.MoveTimer(key, expiry)    } else {        c.timingWheel.SetTimer(key, value, expiry)    }}
  1. 先看在 data map 中有没有存在这个key
  2. 存在，则更新 expire -> MoveTimer()
  3. 第一次设置key -> SetTimer()

所以对于 timingWheel 的应用上就清晰了，开发者依据需要能够 add 或是 update。

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

SetTimer() -> setTask()：

  • not exist task：getPostion -> pushBack to list -> setPosition
  • exist task：get from timers -> moveTask()

MoveTimer() -> moveTask()

由下面的调用链，有一个都会调用的函数：moveTask()

func (tw *TimingWheel) moveTask(task baseEntry) {    // timers: Map => 通过key获取 [positionEntry「pos, task」]    val, ok := tw.timers.Get(task.key)    if !ok {        return    }    timer := val.(*positionEntry)      // {delay < interval} => 延迟时间比一个工夫格距离还小，没有更小的刻度，阐明工作应该立刻执行    if task.delay < tw.interval {        threading.GoSafe(func() {            tw.execute(timer.item.key, timer.item.value)        })        return    }    // 如果 > interval，则通过 延迟时间delay 计算其出工夫轮中的 new pos, circle    pos, circle := tw.getPositionAndCircle(task.delay)    if pos >= timer.pos {        timer.item.circle = circle                // 记录前后的挪动offset。为了前面过程从新入队        timer.item.diff = pos - timer.pos    } else if circle > 0 {        // 转移到下一层，将 circle 转换为 diff 一部分        circle--        timer.item.circle = circle        // 因为是一个数组，要加上 numSlots [也就是相当于要走到下一层]        timer.item.diff = tw.numSlots + pos - timer.pos    } else {        // 如果 offset 提前了，此时 task 也还在第一层        // 标记删除老的 task，并从新入队，期待被执行        timer.item.removed = true        newItem := &timingEntry{            baseEntry: task,            value:     timer.item.value,        }        tw.slots[pos].PushBack(newItem)        tw.setTimerPosition(pos, newItem)    }}

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

  • delay < internal：因为 < 单个工夫精度，示意这个工作曾经过期，须要马上执行

  • 针对扭转的 delay：

    • new >= old：<newPos, newCircle, diff>
    • newCircle > 0：计算diff，并将 circle 转换为 下一层，故diff + numslots
    • 如果只是单纯延迟时间缩短，则将老的task标记删除，重新加入list，期待下一轮loop被execute

Execute

之前在初始化中，run() 中定时器的一直推动，推动的过程次要就是把 list中的 task 传给执行的 execute func。咱们从定时器的执行开始看：

// 定时器 「每隔 internal 会执行一次」func (tw *TimingWheel) onTick() {        // 每次执行更新一下以后执行 tick 地位    tw.tickedPos = (tw.tickedPos + 1) % tw.numSlots        // 获取此时 tick地位 中的存储task的双向链表    l := tw.slots[tw.tickedPos]    tw.scanAndRunTasks(l)}

紧接着是如何去执行 execute：

func (tw *TimingWheel) scanAndRunTasks(l *list.List) {    // 存储目前须要执行的task{key, value}  [execute所须要的参数，顺次传递给execute执行]    var tasks []timingTask    for e := l.Front(); e != nil; {        task := e.Value.(*timingEntry)                // 标记删除，在 scan 中做真正的删除 「删除map的data」        if task.removed {            next := e.Next()            l.Remove(e)            tw.timers.Del(task.key)            e = next            continue        } else if task.circle > 0 {            // 以后执行点曾经过期，然而同时不在第一层，所以以后层即然曾经实现了，就会降到下一层                        // 然而并没有批改 pos            task.circle--            e = e.Next()            continue        } else if task.diff > 0 {            // 因为之前曾经标注了diff，须要再进入队列            next := e.Next()            l.Remove(e)            pos := (tw.tickedPos + task.diff) % tw.numSlots            tw.slots[pos].PushBack(task)            tw.setTimerPosition(pos, task)            task.diff = 0            e = next            continue        }        // 以上的状况都是不能执行的状况，可能执行的会被退出tasks中        tasks = append(tasks, timingTask{            key:   task.key,            value: task.value,        })        next := e.Next()        l.Remove(e)        tw.timers.Del(task.key)        e = next    }    // for range tasks，而后把每个 task->execute 执行即可    tw.runTasks(tasks)}

具体的分支状况在正文中阐明了，在看的时候能够和后面的 moveTask() 联合起来，其中 circle 降落，diff 的计算是关联两个函数的重点。

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

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

下面的过程能够简化成上面：

steps = d / intervalpos = step % numSlots - 1circle = (step - 1) / numSlots

总结

  1. timingWheel 靠定时器推动，工夫后退的同时会取出以后工夫格中 list「双向链表」的task，传递到 execute 中执行。因为是是靠 internal 固定工夫刻度推动，可能就会呈现：一个 60s 的task，internal = 1s，这样就会空跑59次loop。
  2. 而在扩大工夫上，采取 circle 分层，这样就能够一直复用原有的 numSlots ，因为定时器在一直 loop，而执行能够把下层的 slot 降落到上层，在一直 loop 中就能够执行到下层的task。这样的设计能够在不发明额定的数据结构，冲破长时间的限度。
同时在 go-zero 中还有很多实用的组件工具，用好工具对于晋升服务性能和开发效率都有很大的帮忙，心愿本篇文章能给大家带来一些播种。

我的项目地址

https://github.com/tal-tech/go-zero

好将来技术