前言
在实际业务场景中，我们有时候会碰到一些延时的需求：例如，在电商平台，运营在管理后台添加商品后，不需要立刻展示在前台，而是在之后某个时间点才展现。
当然，我们有很多种思路，可以应对这个问题。例如，将待发布商品信息添加到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
3
4
5
6
7
|
// 延时任务对象type DelayQueue struct { tasks []*task // 存储任务列表的切片 add chan *task // 用户添加任务的管道信号 remove chan string // 用户删除任务的管道信号 waitRemoveTaskMapping map[string]struct{} // 等待删除的任务id列表} |
这里需要注意，有一个waitRemoveTaskMapping字段。由于要删除的任务，可能还在add管道中，没有及时更新到tasks字段中，所以，需要临时记录下客户要删除的任务id。
（2）任务对象
|
1
2
3
4
5
6
|
// 任务对象type task struct { id string // 任务id execTime time.Time // 执行时间 f func() // 执行函数} |
3、初始化延时任务对象
|
1
2
3
4
5
6
7
8
9
|
// 初始化延时任务对象func InitDelayQueue() *DelayQueue { q := &DelayQueue{ add: make(chan *task, 10000), remove: make(chan string, 100), waitRemoveTaskMapping: make(map[string]struct{}), } return q} |
在这个过程中，我们需要对用户对任务的操作信号，以及任务的执行时间信号进行监听。
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
|
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... } }} |
完善我们的初始化方法：
|
1
2
3
4
5
6
7
8
9
10
11
|
// 初始化延时任务对象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，并返回给用户。
|
1
2
3
4
5
6
7
8
9
10
11
12
13
|
// 用户推送任务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字段中。由于，我们需要将任务按照执行时间顺序排序，所以，我们需要找到新增任务在切片中的插入位置。又因为，插入之前的任务列表已经是有序的，所以，我们可以采用二分法处理。
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
|
// 使用二分法判断新增任务的插入位置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) }} |
找到正确的插入位置后，我们才能将任务准确插入：
|
1
2
3
4
5
6
7
8
9
|
// 将任务添加到任务切片列表中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() 即可。
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
|
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、生产者删除任务
|
1
2
3
4
|
// 用户删除任务func (q *DelayQueue) Delete(id string) { q.remove <- id} |
7、任务删除信号的处理
在这里，我们可以遍历任务列表，根据删除任务的id找到其在切片中的对应index。
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
|
// 删除指定任务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()方法了。
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
|
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来实现，属于第二种情况。
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
|
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) } }} |
执行任务：
|
1
2
3
4
5
6
7
8
9
10
|
// 执行任务func (q *DelayQueue) execTask(task *task, currentTime time.Time) { if task.execTime.After(currentTime) { // 如果当前任务的执行时间落后于当前时间，则不执行 return } // 执行任务 task.f() return} |
结束任务，刷新任务列表：
|
1
2
3
4
5
6
7
8
|
// 一个任务去执行了，刷新任务列表func (q *DelayQueue) endTask() { if len(q.tasks) == 1 { q.tasks = []*task{} return } q.tasks = q.tasks[1:]} |
9、完整代码
delay_queue.go
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
|
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
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
|
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)} |
头脑风暴
上面的方案，的确实现了延时任务的效果，但是其中仍然有一些问题，仍然值得我们思考和优化。
1、按照上面的方案，如果大量延时任务的执行时间，集中在同一个时间点，会造成短时间内timer频繁地创建和销毁。
2、上述方案相比于time.AfterFunc()方法，我们需要在哪些场景下作出取舍。
3、如果服务崩溃或重启，如何去持久化队列中的任务。
小结
本文和大家讨论了延时任务在golang中的一种实现方案，在这个过程中，一次性定时器timer、切片、管道等golang特色，以及二分插入等常见算法都体现得淋漓尽致。
原文链接：https://juejin.cn/post/7212210588826501176
