一文详解golang延时任务的实现

作者:游鱼的编程旅行 时间:2024-03-23 16:00:16 

前言

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

当然,我们有很多种思路,可以应对这个问题。例如,将待发布商品信息添加到db,然后通过定时任务轮询数据表的方式,查询当前时间点的发布商品;又比如,将商品信息全部添加到redis中,通过SortSet属性完成这个功能。最终的选择,取决于我们的业务场景和运行环境。

在这里,我想给大家分享一套,基于golang实现的延时任务方案。

你可以收获

  • golang管道的灵活运用

  • golang timer的应用

  • golang切片元素插入排序的实现思路

  • 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管道有元素输出的时候),执行任务。

一文详解golang延时任务的实现

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特色,以及二分插入等常见算法都体现得淋漓尽致。

来源:https://juejin.cn/post/7212210588826501176

标签:golang,延时任务,延时
0
投稿

猜你喜欢

  • 我的论坛源代码(九)

    2023-11-15 05:50:05
  • django manage.py扩展自定义命令方法

    2023-11-03 20:55:42
  • Vue下滚动到页面底部无限加载数据的示例代码

    2024-06-05 15:31:44
  • 深入理解ES6 Promise 扩展always方法

    2024-04-18 09:46:46
  • PyTorch之torch.randn()如何创建正态分布随机数

    2022-10-20 01:22:18
  • 利用Python实现自动工作汇报的脚本分享

    2022-05-31 10:55:37
  • mysql代码执行结构实例分析【顺序、分支、循环结构】

    2024-01-21 23:18:40
  • 手把手教你将Flask应用封装成Docker服务的实现

    2023-05-27 06:57:15
  • 介绍27款经典的CSS框架

    2011-03-04 16:24:00
  • 详解nodejs express下使用redis管理session

    2024-05-11 09:51:40
  • 神经网络训练采用gpu设置的方式

    2021-04-24 16:43:25
  • 浅谈MySQL 有哪些死锁场景

    2024-01-13 13:45:54
  • Python中可变和不可变对象的深入讲解

    2022-12-29 21:31:21
  • Python word实现读取及导出代码解析

    2021-05-11 07:56:47
  • Python入门教程(三十九)Python的NumPy安装与入门

    2023-08-25 10:38:30
  • 10分钟带你上手Vue3中新增的API

    2024-06-05 10:03:00
  • Python函数式编程指南(一):函数式编程概述

    2023-07-08 01:20:25
  • python3中apply函数和lambda函数的使用详解

    2022-03-12 00:08:22
  • Ubuntu下安装卸载python3.8的过程

    2021-09-06 10:43:28
  • Window环境下MySQL UDF提权

    2024-01-21 09:53:01
  • asp之家 网络编程 m.aspxhome.com