经验首页 前端设计 程序设计 Java相关 移动开发 数据库/运维 软件/图像 大数据/云计算 其他经验
当前位置：技术经验 » 程序设计 » Go语言 » 查看文章
一文详解golang延时任务的实现
来源：jb51　　时间：2023/3/22 9:23:29　　对本文有异议

前言

在实际业务场景中，我们有时候会碰到一些延时的需求：例如，在电商平台，运营在管理后台添加商品后，不需要立刻展示在前台，而是在之后某个时间点才展现。

当然，我们有很多种思路，可以应对这个问题。例如，将待发布商品信息添加到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）延时任务对象

  1. // 延时任务对象
  2. type DelayQueue struct {
  3.    tasks                 []*task             // 存储任务列表的切片
  4.    add                   chan *task          // 用户添加任务的管道信号
  5.    remove                chan string         // 用户删除任务的管道信号
  6.    waitRemoveTaskMapping map[string]struct{} // 等待删除的任务id列表
  7. }

这里需要注意，有一个waitRemoveTaskMapping字段。由于要删除的任务，可能还在add管道中，没有及时更新到tasks字段中，所以，需要临时记录下客户要删除的任务id。

（2）任务对象

  1. // 任务对象
  2. type task struct {
  3.    id       string    // 任务id
  4.    execTime time.Time // 执行时间
  5.    f        func()    // 执行函数
  6. }

3、初始化延时任务对象

  1. // 初始化延时任务对象
  2. func InitDelayQueue() *DelayQueue {
  3.    q := &DelayQueue{
  4.       add:                   make(chan *task, 10000),
  5.       remove:                make(chan string, 100),
  6.       waitRemoveTaskMapping: make(map[string]struct{}),
  7.    }
  8.    return q
  9. }

在这个过程中，我们需要对用户对任务的操作信号，以及任务的执行时间信号进行监听。

  1. func (q *DelayQueue) start() {
  2.    for {
  3.       // to do something...
  4.       select {
  5.       case now := <-timer.C:
  6.          // 任务执行时间信号
  7.          // to do something...
  8.       case t := <-q.add:
  9.          // 任务推送信号
  10.          // to do something...
  11.       case id := <-q.remove:
  12.          // 任务删除信号
  13.          // to do something...
  14.       }
  15.    }
  16. }

完善我们的初始化方法：

  1. // 初始化延时任务对象
  2. func InitDelayQueue() *DelayQueue {
  3.    q := &DelayQueue{
  4.       add:                   make(chan *task, 10000),
  5.       remove:                make(chan string, 100),
  6.       waitRemoveTaskMapping: make(map[string]struct{}),
  7.    }
  8.    // 开启协程，监听任务相关信号
  9.    go q.start()
  10.    return q
  11. }

4、生产者推送任务

生产者推送任务的时候，只需要将任务加到add管道中即可，在这里，我们生成一个任务id，并返回给用户。

  1. // 用户推送任务
  2. func (q *DelayQueue) Push(timeInterval time.Duration, f func()) string {
  3.    // 生成一个任务id，方便删除使用
  4.    id := genTaskId()
  5.    t := &task{
  6.       id:       id,
  7.       execTime: time.Now().Add(timeInterval),
  8.       f:        f,
  9.    }
  10.    // 将任务推到add管道中
  11.    q.add <- t
  12.    return id
  13. }

5、任务推送信号的处理

在这里，我们要将用户推送的任务放到延时任务的tasks字段中。由于，我们需要将任务按照执行时间顺序排序，所以，我们需要找到新增任务在切片中的插入位置。又因为，插入之前的任务列表已经是有序的，所以，我们可以采用二分法处理。

  1. // 使用二分法判断新增任务的插入位置
  2. func (q *DelayQueue) getTaskInsertIndex(t *task, leftIndex, rightIndex int) (index int) {
  3.    if len(q.tasks) == 0 {
  4.       return
  5.    }
  6.    length := rightIndex - leftIndex
  7.    if q.tasks[leftIndex].execTime.Sub(t.execTime) >= 0 {
  8.       // 如果当前切片中最小的元素都超过了插入的优先级，则插入位置应该是最左边
  9.       return leftIndex
  10.    }
  11.    if q.tasks[rightIndex].execTime.Sub(t.execTime) <= 0 {
  12.       // 如果当前切片中最大的元素都没超过插入的优先级，则插入位置应该是最右边
  13.       return rightIndex + 1
  14.    }
  15.    if length == 1 && q.tasks[leftIndex].execTime.Before(t.execTime) && q.tasks[rightIndex].execTime.Sub(t.execTime) >= 0 {
  16.       // 如果插入的优先级刚好在仅有的两个优先级之间，则中间的位置就是插入位置
  17.       return leftIndex + 1
  18.    }
  19.    middleVal := q.tasks[leftIndex+length/2].execTime
  20.    // 这里用二分法递归的方式，一直寻找正确的插入位置
  21.    if t.execTime.Sub(middleVal) <= 0 {
  22.       return q.getTaskInsertIndex(t, leftIndex, leftIndex+length/2)
  23.    } else {
  24.       return q.getTaskInsertIndex(t, leftIndex+length/2, rightIndex)
  25.    }
  26. }

找到正确的插入位置后，我们才能将任务准确插入：

  1. // 将任务添加到任务切片列表中
  2. func (q *DelayQueue) addTask(t *task) {
  3.    // 寻找新增任务的插入位置
  4.    insertIndex := q.getTaskInsertIndex(t, 0, len(q.tasks)-1)
  5.    // 找到了插入位置，更新任务列表
  6.    q.tasks = append(q.tasks, &task{})
  7.    copy(q.tasks[insertIndex+1:], q.tasks[insertIndex:])
  8.    q.tasks[insertIndex] = t
  9. }

那么，在监听add管道的时候，我们直接调用上述addTask() 即可。

  1. func (q *DelayQueue) start() {
  2.    for {
  3.       // to do something...
  4.       select {
  5.       case now := <-timer.C:
  6.          // 任务执行时间信号
  7.          // to do something...
  8.       case t := <-q.add:
  9.          // 任务推送信号
  10.          q.addTask(t)
  11.       case id := <-q.remove:
  12.          // 任务删除信号
  13.          // to do something...
  14.       }
  15.    }
  16. }

6、生产者删除任务

  1. // 用户删除任务
  2. func (q *DelayQueue) Delete(id string) {
  3.    q.remove <- id
  4. }

7、任务删除信号的处理

在这里，我们可以遍历任务列表，根据删除任务的id找到其在切片中的对应index。

  1. // 删除指定任务
  2. func (q *DelayQueue) deleteTask(id string) {
  3.    deleteIndex := -1
  4.    for index, t := range q.tasks {
  5.       if t.id == id {
  6.          // 找到了在切片中需要删除的所以呢
  7.          deleteIndex = index
  8.          break
  9.       }
  10.    }
  11.    if deleteIndex == -1 {
  12.       // 如果没有找到删除的任务，说明任务还在add管道中，来不及更新到tasks中，这里我们就将这个删除id临时记录下来
  13.       // 注意，这里暂时不考虑，任务id非法的特殊情况
  14.       q.waitRemoveTaskMapping[id] = struct{}{}
  15.       return
  16.    }
  17.    if len(q.tasks) == 1 {
  18.       // 删除后，任务列表就没有任务了
  19.       q.tasks = []*task{}
  20.       return
  21.    }
  22.    if deleteIndex == len(q.tasks)-1 {
  23.       // 如果删除的是，任务列表的最后一个元素，则执行下列代码
  24.       q.tasks = q.tasks[:len(q.tasks)-1]
  25.       return
  26.    }
  27.    // 如果删除的是，任务列表的其他元素，则需要将deleteIndex之后的元素，全部向前挪动一位
  28.    copy(q.tasks[deleteIndex:len(q.tasks)-1], q.tasks[deleteIndex+1:len(q.tasks)-1])
  29.    q.tasks = q.tasks[:len(q.tasks)-1]
  30.    return
  31. }

然后，我们可以完善start()方法了。

  1. func (q *DelayQueue) start() {
  2.    for {
  3.       // to do something...
  4.       select {
  5.       case now := <-timer.C:
  6.          // 任务执行时间信号
  7.          // to do something...
  8.       case t := <-q.add:
  9.          // 任务推送信号
  10.          q.addTask(t)
  11.       case id := <-q.remove:
  12.          // 任务删除信号
  13.          q.deleteTask(id)
  14.       }
  15.    }
  16. }

8、任务执行信号的处理

start()执行的时候，分成两种情况：任务列表为空，只需要监听add管道即可；任务列表不为空的时候，需要监听所有管道。任务执行信号，主要是依靠timer来实现，属于第二种情况。

  1. func (q *DelayQueue) start() {
  2.    for {
  3.       if len(q.tasks) == 0 {
  4.            // 任务列表为空的时候，只需要监听add管道
  5.            select {
  6.            case t := <-q.add:
  7.               //添加任务
  8.               q.addTask(t)
  9.            }
  10.            continue
  11.       }
  12.       // 任务列表不为空的时候，需要监听所有管道
  13.       // 任务的等待时间=任务的执行时间-当前的时间
  14.       currentTask := q.tasks[0]
  15.       timer := time.NewTimer(currentTask.execTime.Sub(time.Now()))
  16.       select {
  17.       case now := <-timer.C:
  18.          // 任务执行信号
  19.          timer.Stop()
  20.         if _, isRemove := q.waitRemoveTaskMapping[currentTask.id]; isRemove {
  21.            // 之前客户已经发出过该任务的删除信号，因此需要结束任务，刷新任务列表
  22.            q.endTask()
  23.            delete(q.waitRemoveTaskMapping, currentTask.id)
  24.            continue
  25.         }
  26.         // 开启协程，异步执行任务
  27.         go q.execTask(currentTask, now)
  28.         // 任务结束，刷新任务列表
  29.         q.endTask()
  30.       case t := <-q.add:
  31.          // 任务推送信号
  32.          timer.Stop()
  33.          q.addTask(t)
  34.       case id := <-q.remove:
  35.          // 任务删除信号
  36.          timer.Stop()
  37.          q.deleteTask(id)
  38.       }
  39.    }
  40. }

执行任务：

  1. // 执行任务
  2. func (q *DelayQueue) execTask(task *task, currentTime time.Time) {
  3.    if task.execTime.After(currentTime) {
  4.       // 如果当前任务的执行时间落后于当前时间，则不执行
  5.       return
  6.    }
  7.    // 执行任务
  8.    task.f()
  9.    return
  10. }

结束任务，刷新任务列表：

  1. // 一个任务去执行了，刷新任务列表
  2. func (q *DelayQueue) endTask() {
  3.    if len(q.tasks) == 1 {
  4.       q.tasks = []*task{}
  5.       return
  6.    }
  7.    q.tasks = q.tasks[1:]
  8. }

9、完整代码

delay_queue.go

  1. package delay_queue
  2. import (
  3.    "go.mongodb.org/mongo-driver/bson/primitive"
  4.    "time"
  5. )
  6. // 延时任务对象
  7. type DelayQueue struct {
  8.    tasks                 []*task             // 存储任务列表的切片
  9.    add                   chan *task          // 用户添加任务的管道信号
  10.    remove                chan string         // 用户删除任务的管道信号
  11.    waitRemoveTaskMapping map[string]struct{} // 等待删除的任务id列表
  12. }
  13. // 任务对象
  14. type task struct {
  15.    id       string    // 任务id
  16.    execTime time.Time // 执行时间
  17.    f        func()    // 执行函数
  18. }
  19. // 初始化延时任务对象
  20. func InitDelayQueue() *DelayQueue {
  21.    q := &DelayQueue{
  22.       add:                   make(chan *task, 10000),
  23.       remove:                make(chan string, 100),
  24.       waitRemoveTaskMapping: make(map[string]struct{}),
  25.    }
  26.    // 开启协程，监听任务相关信号
  27.    go q.start()
  28.    return q
  29. }
  30. // 用户删除任务
  31. func (q *DelayQueue) Delete(id string) {
  32.    q.remove <- id
  33. }
  34. // 用户推送任务
  35. func (q *DelayQueue) Push(timeInterval time.Duration, f func()) string {
  36.    // 生成一个任务id，方便删除使用
  37.    id := genTaskId()
  38.    t := &task{
  39.       id:       id,
  40.       execTime: time.Now().Add(timeInterval),
  41.       f:        f,
  42.    }
  43.    // 将任务推到add管道中
  44.    q.add <- t
  45.    return id
  46. }
  47. // 监听各种任务相关信号
  48. func (q *DelayQueue) start() {
  49.    for {
  50.       if len(q.tasks) == 0 {
  51.          // 任务列表为空的时候，只需要监听add管道
  52.          select {
  53.          case t := <-q.add:
  54.             //添加任务
  55.             q.addTask(t)
  56.          }
  57.          continue
  58.       }
  59.       // 任务列表不为空的时候，需要监听所有管道
  60.       // 任务的等待时间=任务的执行时间-当前的时间
  61.       currentTask := q.tasks[0]
  62.       timer := time.NewTimer(currentTask.execTime.Sub(time.Now()))
  63.       select {
  64.       case now := <-timer.C:
  65.          timer.Stop()
  66.          if _, isRemove := q.waitRemoveTaskMapping[currentTask.id]; isRemove {
  67.             // 之前客户已经发出过该任务的删除信号，因此需要结束任务，刷新任务列表
  68.             q.endTask()
  69.             delete(q.waitRemoveTaskMapping, currentTask.id)
  70.             continue
  71.          }
  72.          // 开启协程，异步执行任务
  73.          go q.execTask(currentTask, now)
  74.          // 任务结束，刷新任务列表
  75.          q.endTask()
  76.       case t := <-q.add:
  77.          // 添加任务
  78.          timer.Stop()
  79.          q.addTask(t)
  80.       case id := <-q.remove:
  81.          // 删除任务
  82.          timer.Stop()
  83.          q.deleteTask(id)
  84.       }
  85.    }
  86. }
  87. // 执行任务
  88. func (q *DelayQueue) execTask(task *task, currentTime time.Time) {
  89.    if task.execTime.After(currentTime) {
  90.       // 如果当前任务的执行时间落后于当前时间，则不执行
  91.       return
  92.    }
  93.    // 执行任务
  94.    task.f()
  95.    return
  96. }
  97. // 一个任务去执行了，刷新任务列表
  98. func (q *DelayQueue) endTask() {
  99.    if len(q.tasks) == 1 {
  100.       q.tasks = []*task{}
  101.       return
  102.    }
  103.    q.tasks = q.tasks[1:]
  104. }
  105. // 将任务添加到任务切片列表中
  106. func (q *DelayQueue) addTask(t *task) {
  107.    // 寻找新增任务的插入位置
  108.    insertIndex := q.getTaskInsertIndex(t, 0, len(q.tasks)-1)
  109.    // 找到了插入位置，更新任务列表
  110.    q.tasks = append(q.tasks, &task{})
  111.    copy(q.tasks[insertIndex+1:], q.tasks[insertIndex:])
  112.    q.tasks[insertIndex] = t
  113. }
  114. // 删除指定任务
  115. func (q *DelayQueue) deleteTask(id string) {
  116.    deleteIndex := -1
  117.    for index, t := range q.tasks {
  118.       if t.id == id {
  119.          // 找到了在切片中需要删除的所以呢
  120.          deleteIndex = index
  121.          break
  122.       }
  123.    }
  124.    if deleteIndex == -1 {
  125.       // 如果没有找到删除的任务，说明任务还在add管道中，来不及更新到tasks中，这里我们就将这个删除id临时记录下来
  126.       // 注意，这里暂时不考虑，任务id非法的特殊情况
  127.       q.waitRemoveTaskMapping[id] = struct{}{}
  128.       return
  129.    }
  130.    if len(q.tasks) == 1 {
  131.       // 删除后，任务列表就没有任务了
  132.       q.tasks = []*task{}
  133.       return
  134.    }
  135.    if deleteIndex == len(q.tasks)-1 {
  136.       // 如果删除的是，任务列表的最后一个元素，则执行下列代码
  137.       q.tasks = q.tasks[:len(q.tasks)-1]
  138.       return
  139.    }
  140.    // 如果删除的是，任务列表的其他元素，则需要将deleteIndex之后的元素，全部向前挪动一位
  141.    copy(q.tasks[deleteIndex:len(q.tasks)-1], q.tasks[deleteIndex+1:len(q.tasks)-1])
  142.    q.tasks = q.tasks[:len(q.tasks)-1]
  143.    return
  144. }
  145. // 寻找任务的插入位置
  146. func (q *DelayQueue) getTaskInsertIndex(t *task, leftIndex, rightIndex int) (index int) {
  147.    // 使用二分法判断新增任务的插入位置
  148.    if len(q.tasks) == 0 {
  149.       return
  150.    }
  151.    length := rightIndex - leftIndex
  152.    if q.tasks[leftIndex].execTime.Sub(t.execTime) >= 0 {
  153.       // 如果当前切片中最小的元素都超过了插入的优先级，则插入位置应该是最左边
  154.       return leftIndex
  155.    }
  156.    if q.tasks[rightIndex].execTime.Sub(t.execTime) <= 0 {
  157.       // 如果当前切片中最大的元素都没超过插入的优先级，则插入位置应该是最右边
  158.       return rightIndex + 1
  159.    }
  160.    if length == 1 && q.tasks[leftIndex].execTime.Before(t.execTime) && q.tasks[rightIndex].execTime.Sub(t.execTime) >= 0 {
  161.       // 如果插入的优先级刚好在仅有的两个优先级之间，则中间的位置就是插入位置
  162.       return leftIndex + 1
  163.    }
  164.    middleVal := q.tasks[leftIndex+length/2].execTime
  165.    // 这里用二分法递归的方式，一直寻找正确的插入位置
  166.    if t.execTime.Sub(middleVal) <= 0 {
  167.       return q.getTaskInsertIndex(t, leftIndex, leftIndex+length/2)
  168.    } else {
  169.       return q.getTaskInsertIndex(t, leftIndex+length/2, rightIndex)
  170.    }
  171. }
  172. func genTaskId() string {
  173.    return primitive.NewObjectID().Hex()
  174. }

测试代码：delay_queue_test.go

  1. package delay_queue
  2. import (
  3.    "fmt"
  4.    "testing"
  5.    "time"
  6. )
  7. func TestDelayQueue(t *testing.T) {
  8.    q := InitDelayQueue()
  9.    for i := 0; i < 100; i++ {
  10.       go func(i int) {
  11.          id := q.Push(time.Duration(i)*time.Second, func() {
  12.             fmt.Printf("%d秒后执行...\n", i)
  13.             return
  14.          })
  15.          if i%7 == 0 {
  16.             q.Delete(id)
  17.          }
  18.       }(i)
  19.    }
  20.    time.Sleep(time.Hour)
  21. }

头脑风暴

上面的方案，的确实现了延时任务的效果，但是其中仍然有一些问题，仍然值得我们思考和优化。

1、按照上面的方案，如果大量延时任务的执行时间，集中在同一个时间点，会造成短时间内timer频繁地创建和销毁。

2、上述方案相比于time.AfterFunc()方法，我们需要在哪些场景下作出取舍。

3、如果服务崩溃或重启，如何去持久化队列中的任务。

小结

本文和大家讨论了延时任务在golang中的一种实现方案，在这个过程中，一次性定时器timer、切片、管道等golang特色，以及二分插入等常见算法都体现得淋漓尽致。

 友情链接：直通硅谷 　点职佳 　北美留学生论坛

Go语言热门文章

  • Golang从文件服务器获取图片显示到客户端
  • 谈谈Go语言中的内存重排
  • Golang 模块引入及表格读写业务快速实现示例
  • 创建并使用https证书
  • golang的reflection（转）
  • TICK技术栈（一）TICK技术栈介绍
  • beego解读Cache之MemoryCache
  • go语言reflect包最佳实践之struct操作（遍历、赋值与方法调用）
  • golang中gin框架接入jwt使用token验证身份
  • Golang解析、验证、修改URL之Host、Port、Path

    • Go语言推荐文章

  • Go 互斥锁 Mutex 源码分析(二)
  • golang对遍历目录操作的优化
  • 美团VS饿了么，到底谁更胜一筹？
  • 解码 xsync 的 map 实现
  • [golang]在Gin框架中使用JWT鉴权
  • Golang 依赖注入设计哲学｜12.6K 🌟 的依赖注入库 wire
  • 这么简单的问题都不会，那还面试什么！？
  • golang select 和外层的 for 搭配
  • Go语言的100个错误使用场景（61-68）｜并发实践
  • 实战：如何优雅的从 Skywalking 切换到 OpenTelemetry

    • 本站QQ群：前端 618073944 | Java 606181507 | Python 626812652 | C/C++ 612253063 | 微信 634508462 | 苹果 692586424 | C#/.net 182808419 | PHP 305140648 | 运维 608723728

    W3xue 的所有内容仅供测试，对任何法律问题及风险不承担任何责任。通过使用本站内容随之而来的风险与本站无关。
    关于我们  |  意见建议  |  捐助我们  |  报错有奖  |  广告合作、友情链接（目前9元/月）请联系QQ：27243702 沸活量
    皖ICP备17017327号-2　皖公网安备34020702000426号