golang延时任务如何实现
短信预约 -IT技能 免费直播动态提醒
这篇文章主要介绍“golang延时任务如何实现”，在日常操作中，相信很多人在golang延时任务如何实现问题上存在疑惑，小编查阅了各式资料，整理出简单好用的操作方法，希望对大家解答”golang延时任务如何实现”的疑惑有所帮助！接下来，请跟着小编一起来学习吧！
实现思路
我们都知道，任何一种队列，实际上都是存在生产者和消费者两部分的。只不过，延时任务相对于普通队列，多了一个延时的特性罢了。
生产者
从生产者的角度上讲，当用户推送一个任务过来的时候，会携带着延迟执行的时间数值。为了让这个任务到预定时刻能执行，我们需要将这个任务放在内存里储存一段时间，并且时间是一维的，在不断增长。那么，我们用什么数据结构存储呢？
（1）选择一：map。由于map具有无序性，无法按照执行时间排序，我们无法保证取出的任务是否是当前时间点需要执行的，所以排除这个选项。
（2）选择二：channel。的确，channel有时候可以看作队列，然而，它的输出和输入严格遵循着“先进先出”的原则，遗憾的是，先进的任务未必就是先执行的，因此，channel也并不合适。
（3）选择三：slice。切片貌似可行，因为切片元素是具有有序性的，所以，如果我们能够按照执行时间的顺序排列好所有的切片元素，那么，每次只要读取切片的头元素（也可能是尾元素），就可以得到我们要的任务。
消费者
从消费者的角度来说，它最大的难点在于，如何让每个任务，在特定的时间点被消费。那么，针对每一个任务，我们如何实现，让它等待一段时间后再执行呢？
没错，就是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_queueimport ( "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_queueimport ( "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)}头脑风暴
上面的方案，的确实现了延时任务的效果，但是其中仍然有一些问题，仍然值得我们思考和优化。
按照上面的方案，如果大量延时任务的执行时间，集中在同一个时间点，会造成短时间内timer频繁地创建和销毁。
上述方案相比于time.AfterFunc()方法，我们需要在哪些场景下作出取舍。
如果服务崩溃或重启，如何去持久化队列中的任务。
到此，关于“golang延时任务如何实现”的学习就结束了，希望能够解决大家的疑惑。理论与实践的搭配能更好的帮助大家学习，快去试试吧！若想继续学习更多相关知识，请继续关注编程网网站，小编会继续努力为大家带来更多实用的文章！
免责声明：
① 本站未注明“稿件来源”的信息均来自网络整理。其文字、图片和音视频稿件的所属权归原作者所有。本站收集整理出于非商业性的教育和科研之目的，并不意味着本站赞同其观点或证实其内容的真实性。仅作为临时的测试数据，供内部测试之用。本站并未授权任何人以任何方式主动获取本站任何信息。
② 本站未注明“稿件来源”的临时测试数据将在测试完成后最终做删除处理。有问题或投稿请发送至: 邮箱/279061341@qq.com QQ/279061341
