一文詳解golang延時任務的實現

前言

在實際業務場景中,我們有時候會碰到一些延時的需求:例如,在電商平臺,運營在管理後臺添加商品後,不需要立刻展示在前臺,而是在之後某個時間點才展現。

當然,我們有很多種思路,可以應對這個問題。例如,將待發佈商品信息添加到db,然後通過定時任務輪詢數據表的方式,查詢當前時間點的發佈商品;又比如,將商品信息全部添加到redis中,通過SortSet屬性完成這個功能。最終的選擇,取決於我們的業務場景和運行環境。

在這裡,我想給大傢分享一套,基於golang實現的延時任務方案。

你可以收獲

  • golang管道的靈活運用
  • golang timer的應用
  • golang切片元素插入排序的實現思路
  • golang延時任務的實現思路

正文

思維導圖

為瞭讓大傢有一個大致的印象,我將正文的大綱列在下面。

實現思路

我們都知道,任何一種隊列,實際上都是存在生產者和消費者兩部分的。隻不過,延時任務相對於普通隊列,多瞭一個延時的特性罷瞭。

1、生產者

從生產者的角度上講,當用戶推送一個任務過來的時候,會攜帶著延遲執行的時間數值。為瞭讓這個任務到預定時刻能執行,我們需要將這個任務放在內存裡儲存一段時間,並且時間是一維的,在不斷增長。那麼,我們用什麼數據結構存儲呢?

(1)選擇一:map。由於map具有無序性,無法按照執行時間排序,我們無法保證取出的任務是否是當前時間點需要執行的,所以排除這個選項。

(2)選擇二:channel。的確,channel有時候可以看作隊列,然而,它的輸出和輸入嚴格遵循著“先進先出”的原則,遺憾的是,先進的任務未必就是先執行的,因此,channel也並不合適。

(3)選擇三:slice。切片貌似可行,因為切片元素是具有有序性的,所以,如果我們能夠按照執行時間的順序排列好所有的切片元素,那麼,每次隻要讀取切片的頭元素(也可能是尾元素),就可以得到我們要的任務。

2、消費者

從消費者的角度來說,它最大的難點在於,如何讓每個任務,在特定的時間點被消費。那麼,針對每一個任務,我們如何實現,讓它等待一段時間後再執行呢?

沒錯,就是timer。

總結下來,“切片+timer”的組合,應該是可以達到目的的。

步步為營

1、數據流

(1)用戶調用InitDelayQueue() ,初始化延時任務對象。

(2)開啟協程,監聽任務操作管道(add/delete信號),以及執行時間管道(timer.C信號)。

(3)用戶發出add/delete信號。

(4)(2)中的協程捕捉到(3)中的信號,對任務列表進行變更。

(5)當任務執行的時間點到達的時候(timer.C管道有元素輸出的時候),執行任務。

2、數據結構

(1)延時任務對象

// 延時任務對象
type DelayQueue struct {
   tasks                 []*task             // 存儲任務列表的切片
   add                   chan *task          // 用戶添加任務的管道信號
   remove                chan string         // 用戶刪除任務的管道信號
   waitRemoveTaskMapping map[string]struct{} // 等待刪除的任務id列表
}

這裡需要註意,有一個waitRemoveTaskMapping字段。由於要刪除的任務,可能還在add管道中,沒有及時更新到tasks字段中,所以,需要臨時記錄下客戶要刪除的任務id。

(2)任務對象

// 任務對象
type task struct {
   id       string    // 任務id
   execTime time.Time // 執行時間
   f        func()    // 執行函數
}

3、初始化延時任務對象

// 初始化延時任務對象
func InitDelayQueue() *DelayQueue {
   q := &DelayQueue{
      add:                   make(chan *task, 10000),
      remove:                make(chan string, 100),
      waitRemoveTaskMapping: make(map[string]struct{}),
   }
   return q
}

在這個過程中,我們需要對用戶對任務的操作信號,以及任務的執行時間信號進行監聽。

func (q *DelayQueue) start() {
   for {
      // to do something...
      select {
      case now := <-timer.C:
         // 任務執行時間信號
         // to do something...
      case t := <-q.add:
         // 任務推送信號
         // to do something...
      case id := <-q.remove:
         // 任務刪除信號
         // to do something...
      }
   }
}

完善我們的初始化方法:

// 初始化延時任務對象
func InitDelayQueue() *DelayQueue {
   q := &DelayQueue{
      add:                   make(chan *task, 10000),
      remove:                make(chan string, 100),
      waitRemoveTaskMapping: make(map[string]struct{}),
   }
   // 開啟協程,監聽任務相關信號
   go q.start()
   return q
}

4、生產者推送任務

生產者推送任務的時候,隻需要將任務加到add管道中即可,在這裡,我們生成一個任務id,並返回給用戶。

// 用戶推送任務
func (q *DelayQueue) Push(timeInterval time.Duration, f func()) string {
   // 生成一個任務id,方便刪除使用
   id := genTaskId()
   t := &task{
      id:       id,
      execTime: time.Now().Add(timeInterval),
      f:        f,
   }
   // 將任務推到add管道中
   q.add <- t
   return id
}

5、任務推送信號的處理

在這裡,我們要將用戶推送的任務放到延時任務的tasks字段中。由於,我們需要將任務按照執行時間順序排序,所以,我們需要找到新增任務在切片中的插入位置。又因為,插入之前的任務列表已經是有序的,所以,我們可以采用二分法處理。

// 使用二分法判斷新增任務的插入位置
func (q *DelayQueue) getTaskInsertIndex(t *task, leftIndex, rightIndex int) (index int) {
   if len(q.tasks) == 0 {
      return
   }
   length := rightIndex - leftIndex
   if q.tasks[leftIndex].execTime.Sub(t.execTime) >= 0 {
      // 如果當前切片中最小的元素都超過瞭插入的優先級,則插入位置應該是最左邊
      return leftIndex
   }
   if q.tasks[rightIndex].execTime.Sub(t.execTime) <= 0 {
      // 如果當前切片中最大的元素都沒超過插入的優先級,則插入位置應該是最右邊
      return rightIndex + 1
   }
   if length == 1 && q.tasks[leftIndex].execTime.Before(t.execTime) && q.tasks[rightIndex].execTime.Sub(t.execTime) >= 0 {
      // 如果插入的優先級剛好在僅有的兩個優先級之間,則中間的位置就是插入位置
      return leftIndex + 1
   }
   middleVal := q.tasks[leftIndex+length/2].execTime
   // 這裡用二分法遞歸的方式,一直尋找正確的插入位置
   if t.execTime.Sub(middleVal) <= 0 {
      return q.getTaskInsertIndex(t, leftIndex, leftIndex+length/2)
   } else {
      return q.getTaskInsertIndex(t, leftIndex+length/2, rightIndex)
   }
}

找到正確的插入位置後,我們才能將任務準確插入:

// 將任務添加到任務切片列表中
func (q *DelayQueue) addTask(t *task) {
   // 尋找新增任務的插入位置
   insertIndex := q.getTaskInsertIndex(t, 0, len(q.tasks)-1)
   // 找到瞭插入位置,更新任務列表
   q.tasks = append(q.tasks, &task{})
   copy(q.tasks[insertIndex+1:], q.tasks[insertIndex:])
   q.tasks[insertIndex] = t
}

那麼,在監聽add管道的時候,我們直接調用上述addTask() 即可。

func (q *DelayQueue) start() {
   for {
      // to do something...
      select {
      case now := <-timer.C:
         // 任務執行時間信號
         // to do something...
      case t := <-q.add:
         // 任務推送信號
         q.addTask(t)
      case id := <-q.remove:
         // 任務刪除信號
         // to do something...
      }
   }
}

6、生產者刪除任務

// 用戶刪除任務
func (q *DelayQueue) Delete(id string) {
   q.remove <- id
}

7、任務刪除信號的處理

在這裡,我們可以遍歷任務列表,根據刪除任務的id找到其在切片中的對應index。

// 刪除指定任務
func (q *DelayQueue) deleteTask(id string) {
   deleteIndex := -1
   for index, t := range q.tasks {
      if t.id == id {
         // 找到瞭在切片中需要刪除的所以呢
         deleteIndex = index
         break
      }
   }
   if deleteIndex == -1 {
      // 如果沒有找到刪除的任務,說明任務還在add管道中,來不及更新到tasks中,這裡我們就將這個刪除id臨時記錄下來
      // 註意,這裡暫時不考慮,任務id非法的特殊情況
      q.waitRemoveTaskMapping[id] = struct{}{}
      return
   }
   if len(q.tasks) == 1 {
      // 刪除後,任務列表就沒有任務瞭
      q.tasks = []*task{}
      return
   }
   if deleteIndex == len(q.tasks)-1 {
      // 如果刪除的是,任務列表的最後一個元素,則執行下列代碼
      q.tasks = q.tasks[:len(q.tasks)-1]
      return
   }
   // 如果刪除的是,任務列表的其他元素,則需要將deleteIndex之後的元素,全部向前挪動一位
   copy(q.tasks[deleteIndex:len(q.tasks)-1], q.tasks[deleteIndex+1:len(q.tasks)-1])
   q.tasks = q.tasks[:len(q.tasks)-1]
   return
}

然後,我們可以完善start()方法瞭。

func (q *DelayQueue) start() {
   for {
      // to do something...
      select {
      case now := <-timer.C:
         // 任務執行時間信號
         // to do something...
      case t := <-q.add:
         // 任務推送信號
         q.addTask(t)
      case id := <-q.remove:
         // 任務刪除信號
         q.deleteTask(id)
      }
   }
}

8、任務執行信號的處理

start()執行的時候,分成兩種情況:任務列表為空,隻需要監聽add管道即可;任務列表不為空的時候,需要監聽所有管道。任務執行信號,主要是依靠timer來實現,屬於第二種情況。

func (q *DelayQueue) start() {
   for {
      if len(q.tasks) == 0 {
           // 任務列表為空的時候,隻需要監聽add管道
           select {
           case t := <-q.add:
              //添加任務
              q.addTask(t)
           }
           continue
      }
      // 任務列表不為空的時候,需要監聽所有管道
      // 任務的等待時間=任務的執行時間-當前的時間
      currentTask := q.tasks[0]
      timer := time.NewTimer(currentTask.execTime.Sub(time.Now()))
      select {
      case now := <-timer.C:
         // 任務執行信號
         timer.Stop()
        if _, isRemove := q.waitRemoveTaskMapping[currentTask.id]; isRemove {
           // 之前客戶已經發出過該任務的刪除信號,因此需要結束任務,刷新任務列表
           q.endTask()
           delete(q.waitRemoveTaskMapping, currentTask.id)
           continue
        }
        // 開啟協程,異步執行任務
        go q.execTask(currentTask, now)
        // 任務結束,刷新任務列表
        q.endTask()
      case t := <-q.add:
         // 任務推送信號
         timer.Stop()
         q.addTask(t)
      case id := <-q.remove:
         // 任務刪除信號
         timer.Stop()
         q.deleteTask(id)
      }
   }
}

執行任務:

// 執行任務
func (q *DelayQueue) execTask(task *task, currentTime time.Time) {
   if task.execTime.After(currentTime) {
      // 如果當前任務的執行時間落後於當前時間,則不執行
      return
   }
   // 執行任務
   task.f()
   return
}

結束任務,刷新任務列表:

// 一個任務去執行瞭,刷新任務列表
func (q *DelayQueue) endTask() {
   if len(q.tasks) == 1 {
      q.tasks = []*task{}
      return
   }
   q.tasks = q.tasks[1:]
}

9、完整代碼

delay_queue.go

package delay_queue
import (
   "go.mongodb.org/mongo-driver/bson/primitive"
   "time"
)
// 延時任務對象
type DelayQueue struct {
   tasks                 []*task             // 存儲任務列表的切片
   add                   chan *task          // 用戶添加任務的管道信號
   remove                chan string         // 用戶刪除任務的管道信號
   waitRemoveTaskMapping map[string]struct{} // 等待刪除的任務id列表
}
// 任務對象
type task struct {
   id       string    // 任務id
   execTime time.Time // 執行時間
   f        func()    // 執行函數
}
// 初始化延時任務對象
func InitDelayQueue() *DelayQueue {
   q := &DelayQueue{
      add:                   make(chan *task, 10000),
      remove:                make(chan string, 100),
      waitRemoveTaskMapping: make(map[string]struct{}),
   }
   // 開啟協程,監聽任務相關信號
   go q.start()
   return q
}
// 用戶刪除任務
func (q *DelayQueue) Delete(id string) {
   q.remove <- id
}
// 用戶推送任務
func (q *DelayQueue) Push(timeInterval time.Duration, f func()) string {
   // 生成一個任務id,方便刪除使用
   id := genTaskId()
   t := &task{
      id:       id,
      execTime: time.Now().Add(timeInterval),
      f:        f,
   }
   // 將任務推到add管道中
   q.add <- t
   return id
}
// 監聽各種任務相關信號
func (q *DelayQueue) start() {
   for {
      if len(q.tasks) == 0 {
         // 任務列表為空的時候,隻需要監聽add管道
         select {
         case t := <-q.add:
            //添加任務
            q.addTask(t)
         }
         continue
      }
      // 任務列表不為空的時候,需要監聽所有管道
      // 任務的等待時間=任務的執行時間-當前的時間
      currentTask := q.tasks[0]
      timer := time.NewTimer(currentTask.execTime.Sub(time.Now()))
      select {
      case now := <-timer.C:
         timer.Stop()
         if _, isRemove := q.waitRemoveTaskMapping[currentTask.id]; isRemove {
            // 之前客戶已經發出過該任務的刪除信號,因此需要結束任務,刷新任務列表
            q.endTask()
            delete(q.waitRemoveTaskMapping, currentTask.id)
            continue
         }
         // 開啟協程,異步執行任務
         go q.execTask(currentTask, now)
         // 任務結束,刷新任務列表
         q.endTask()
      case t := <-q.add:
         // 添加任務
         timer.Stop()
         q.addTask(t)
      case id := <-q.remove:
         // 刪除任務
         timer.Stop()
         q.deleteTask(id)
      }
   }
}
// 執行任務
func (q *DelayQueue) execTask(task *task, currentTime time.Time) {
   if task.execTime.After(currentTime) {
      // 如果當前任務的執行時間落後於當前時間,則不執行
      return
   }
   // 執行任務
   task.f()
   return
}
// 一個任務去執行瞭,刷新任務列表
func (q *DelayQueue) endTask() {
   if len(q.tasks) == 1 {
      q.tasks = []*task{}
      return
   }
   q.tasks = q.tasks[1:]
}
// 將任務添加到任務切片列表中
func (q *DelayQueue) addTask(t *task) {
   // 尋找新增任務的插入位置
   insertIndex := q.getTaskInsertIndex(t, 0, len(q.tasks)-1)
   // 找到瞭插入位置,更新任務列表
   q.tasks = append(q.tasks, &task{})
   copy(q.tasks[insertIndex+1:], q.tasks[insertIndex:])
   q.tasks[insertIndex] = t
}
// 刪除指定任務
func (q *DelayQueue) deleteTask(id string) {
   deleteIndex := -1
   for index, t := range q.tasks {
      if t.id == id {
         // 找到瞭在切片中需要刪除的所以呢
         deleteIndex = index
         break
      }
   }
   if deleteIndex == -1 {
      // 如果沒有找到刪除的任務,說明任務還在add管道中,來不及更新到tasks中,這裡我們就將這個刪除id臨時記錄下來
      // 註意,這裡暫時不考慮,任務id非法的特殊情況
      q.waitRemoveTaskMapping[id] = struct{}{}
      return
   }
   if len(q.tasks) == 1 {
      // 刪除後,任務列表就沒有任務瞭
      q.tasks = []*task{}
      return
   }
   if deleteIndex == len(q.tasks)-1 {
      // 如果刪除的是,任務列表的最後一個元素,則執行下列代碼
      q.tasks = q.tasks[:len(q.tasks)-1]
      return
   }
   // 如果刪除的是,任務列表的其他元素,則需要將deleteIndex之後的元素,全部向前挪動一位
   copy(q.tasks[deleteIndex:len(q.tasks)-1], q.tasks[deleteIndex+1:len(q.tasks)-1])
   q.tasks = q.tasks[:len(q.tasks)-1]
   return
}
// 尋找任務的插入位置
func (q *DelayQueue) getTaskInsertIndex(t *task, leftIndex, rightIndex int) (index int) {
   // 使用二分法判斷新增任務的插入位置
   if len(q.tasks) == 0 {
      return
   }
   length := rightIndex - leftIndex
   if q.tasks[leftIndex].execTime.Sub(t.execTime) >= 0 {
      // 如果當前切片中最小的元素都超過瞭插入的優先級,則插入位置應該是最左邊
      return leftIndex
   }
   if q.tasks[rightIndex].execTime.Sub(t.execTime) <= 0 {
      // 如果當前切片中最大的元素都沒超過插入的優先級,則插入位置應該是最右邊
      return rightIndex + 1
   }
   if length == 1 && q.tasks[leftIndex].execTime.Before(t.execTime) && q.tasks[rightIndex].execTime.Sub(t.execTime) >= 0 {
      // 如果插入的優先級剛好在僅有的兩個優先級之間,則中間的位置就是插入位置
      return leftIndex + 1
   }
   middleVal := q.tasks[leftIndex+length/2].execTime
   // 這裡用二分法遞歸的方式,一直尋找正確的插入位置
   if t.execTime.Sub(middleVal) <= 0 {
      return q.getTaskInsertIndex(t, leftIndex, leftIndex+length/2)
   } else {
      return q.getTaskInsertIndex(t, leftIndex+length/2, rightIndex)
   }
}
func genTaskId() string {
   return primitive.NewObjectID().Hex()
}

測試代碼:delay_queue_test.go

package delay_queue
import (
   "fmt"
   "testing"
   "time"
)
func TestDelayQueue(t *testing.T) {
   q := InitDelayQueue()
   for i := 0; i < 100; i++ {
      go func(i int) {
         id := q.Push(time.Duration(i)*time.Second, func() {
            fmt.Printf("%d秒後執行...\n", i)
            return
         })
         if i%7 == 0 {
            q.Delete(id)
         }
      }(i)
   }
   time.Sleep(time.Hour)
}

頭腦風暴

上面的方案,的確實現瞭延時任務的效果,但是其中仍然有一些問題,仍然值得我們思考和優化。

1、按照上面的方案,如果大量延時任務的執行時間,集中在同一個時間點,會造成短時間內timer頻繁地創建和銷毀。

2、上述方案相比於time.AfterFunc()方法,我們需要在哪些場景下作出取舍。

3、如果服務崩潰或重啟,如何去持久化隊列中的任務。

小結

本文和大傢討論瞭延時任務在golang中的一種實現方案,在這個過程中,一次性定時器timer、切片、管道等golang特色,以及二分插入等常見算法都體現得淋漓盡致。

推薦閱讀: