golang怎麼實現延時任務？以下這篇文章跟大家分享一套基於golang實作的縮時任務方案，希望對大家有幫助！
在實際業務場景中，我們有時會碰到一些延時的需求：例如，在電商平台，運作在管理後台添加商品後，不需要立刻展示在前台，而是在之後某個時間點才展現。
當然，我們有很多種思路，可以處理這個問題。例如，將待發佈商品資訊新增至db，然後透過定時任務輪詢資料表的方式，查詢目前時間點的發佈商品；又例如，將商品資訊全部新增到redis中，透過SortSet屬性完成這個功能。最終的選擇，取決於我們的業務場景和運行環境。
在這裡，我想跟大家分享一套，基於golang實作的延時任務方案。
#思考導圖
為了讓大家有一個大致的印象，我將正文的大綱列在下面。
實作想法
#我們都知道，任何一種佇列，實際上都是存在生產者和消費者兩部分的。只不過，延時任務相對於一般佇列，多了一個延時的特性罷了。1、生產者
從生產者的角度上講，當使用者推送一個任務過來的時候，會帶著延遲執行的時間數值。為了讓這個任務到預定時刻能執行，我們需要將這個任務放在記憶體中儲存一段時間，並且時間是一維的，並且不斷增長。那麼，我們用什麼資料結構來儲存呢？ （1）選擇一：map。由於map具有無序性，無法按照執行時間排序，我們無法保證取出的任務是否是當前時間點需要執行的，所以排除這個選項。 （2）選擇二：channel。的確，channel有時候可以看作佇列，然而，它的輸出和輸入嚴格遵循著「先進先出」的原則，遺憾的是，先進的任務未必就是先執行的，因此，channel也並不合適。 （3）選擇三：slice。切片貌似可行，因為切片元素是有序性的，所以，如果我們能夠按照執行時間的順序排列好所有的切片元素，那麼，每次只要讀取切片的頭元素（也可能是尾元素），就可以得到我們要的任務。
2、消費者
從消費者的角度來說，它最大的困難在於，如何讓每個任務，在特定的時間點被消費。那麼，針對每一個任務，我們要如何實現，讓它等待一段時間後再執行呢？ 沒錯，就是timer。 總結下來，「切片 timer」的組合，應該是可以達到目的的。
步步為營
InitDelayQueue () ，初始化延時任務物件。
（2）開啟協程，監聽任務操作管道（add/delete訊號），以及執行時間管道（timer.C訊號）。 （3）用戶發出add/delete訊號。 （4）（2）中的協程捕捉（3）中的訊號，對任務清單進行變更。 （5）當任務執行的時間點到達的時候（timer.C管道有元素輸出的時候），執行任務。
// 延时任务对象
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() // 执行函数
}// 初始化延时任务对象
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
}// 用户推送任务
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
}在这里，我们要将用户推送的任务放到延时任务的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...
}
}
}// 用户删除任务
func (q *DelayQueue) Delete(id string) {
q.remove <- id
}在这里，我们可以遍历任务列表，根据删除任务的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)
}
}
}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:]
}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特色，以及二分插入等常见算法都体现得淋漓尽致。
以上是淺析golang怎麼實現延時任務的詳細內容。更多資訊請關注PHP中文網其他相關文章！
