一文詳解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特色,以及二分插入等常見算法都體現得淋漓盡致。