go-zero 如何應對海量定時/延遲任務？

一個系統中存在着大量的調度任務，同時調度任務存在時間的滯後性，而大量的調度任務若是每個都使用本身的調度器來管理任務的生命週期的話，浪費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 = 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

總結

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

好將來技術