一個系統中存在著大量的調度任務，同時調度任務存在時間的滯后性，而大量的調度任務如果每一個都使用自己的調度器來管理任務的生命周期的話，浪費cpu的資源而且很低效，
本文來介紹 go-zero 中 延遲操作，它可能讓開發者調度多個任務時，只需關注具體的業務執行函式和執行時間「立即或者延遲」，而 延遲操作，通常可以采用兩個方案：
Timer：定時器維護一個優先佇列，到時間點執行，然后把需要執行的 task 存盤在 map 中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)
}
}
- 先看在
data map中有沒有存在這個key - 存在，則更新
expire->MoveTimer() - 第一次設定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 = 0
func (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 / interval pos = step % numSlots - 1 circle = (step - 1) / numSlots
總結
-
timingWheel靠定時器推動，時間前進的同時會取出當前時間格中list「雙向鏈表」的task，傳遞到execute中執行，因為是是靠internal固定時間刻度推進，可能就會出現：一個 60s 的task，internal = 1s，這樣就會空跑59次loop， -
而在擴展時間上，采取
circle分層，這樣就可以不斷復用原有的numSlots，因為定時器在不斷loop，而執行可以把上層的slot下降到下層，在不斷loop中就可以執行到上層的task，這樣的設計可以在不創造額外的資料結構，突破長時間的限制，
同時在
go-zero中還有很多實用的組件工具，用好工具對于提升服務性能和開發效率都有很大的幫助，希望本篇文章能給大家帶來一些識訓，
專案地址
https://github.com/tal-tech/go-zero
好未來技術
轉載請註明出處，本文鏈接：https://www.uj5u.com/houduan/173963.html
標籤：Go