- 定义任务结构体 首先，我们需要定义任务结构体，用于存储任务相关的信息。任务结构体至少应该包含以下几个字段：
- ID：任务唯一标识符 *
- Topic：任务主题，用于区分不同类型的任务*
- Delay：任务延迟时间，即从任务加入队列到任务执行之间的时间间隔*
- TTR（Time To Run）：任务最长执行时间，即任务从开始执行到执行结束的时间间隔上限*
- Body：任务内容，即需要执行的具体操作*
- 定义延迟队列 其次，我们需要定义延迟队列结构体，用于存储所有的任务。延迟队列应该包括以下几个组件：
- 任务队列：一个优先队列，用于存储所有任务。按照任务的延迟时间进行排序，最先执行的任务排在队列前面。*
- 延迟桶：一个时间轮，用于管理任务的延迟时间。我们可以设置多个桶，每个桶表示一个时间段，比如1秒、10秒、1分钟等等。每个桶内部维护一个任务队列，用于存储所有延迟时间在该桶时间范围内的任务。每次时间轮tick时，我们需要将当前桶的所有任务添加到任务队列中，以便执行。*
- 定时器：一个定时器，用于定时执行任务队列中的任务。每次定时器tick时，我们从任务队列中取出最先执行的任务，并执行之。如果任务执行超时，我们可以将任务重新加入队列，并根据重试次数设置不同的延迟时间。*
添加任务 当有新任务需要添加到延迟队列中时，我们需要将任务加入到延迟桶中。首先，我们根据任务的延迟时间计算出应该加入到哪个桶中。然后，将任务加入到该桶的任务队列中。如果任务队列为空，则需要重新设置定时器的时间。
执行任务 当定时器tick时，我们从任务队列中取出最先执行的任务，并执行之。如果任务执行成功，则不需要进行任何操作；如果任务执行失败，则可以将任务重新加入队列，并根据重试次数设置不同的延迟时间。在任务执行完成后，我们需要重新设置定时器的时间。
删除任务 当某个任务需要从延迟队列中删除时，我们需要在任务队列和对应的延迟桶中删除该任务。
以上是一个完整的延迟队列系统的实现步骤。具体实现时，我们可以根据需求进行一些调整和优化，比如设置任务优先级、增加任务超时监控、支持任务重试、支持任务取消等等。
package main
import (
"container/heap"
"container/list"
"fmt"
"sync"
"time"
)
// 任务结构体
type Task struct {
ID uint64 // 任务ID
Delay time.Time // 执行时间
Handler func() // 任务处理函数
}
// 实现 heap.Interface 接口，用于构建优先队列
type TaskQueue []*Task
func (t TaskQueue) Len() int {
return len(t)
}
func (t TaskQueue) Less(i, j int) bool {
return t[i].Delay.Before(t[j].Delay)
}
func (t TaskQueue) Swap(i, j int) {
t[i], t[j] = t[j], t[i]
}
func (t *TaskQueue) Push(x interface{}) {
*t = append(*t, x.(*Task))
}
func (t *TaskQueue) Pop() interface{} {
old := *t
n := len(old)
x := old[n-1]
*t = old[:n-1]
return x
}
// 延迟队列结构体
type DelayQueue struct {
buckets map[time.Duration]*list.List // 延迟桶
taskQueue TaskQueue // 任务队列
mutex sync.Mutex // 互斥锁
}
// 创建延迟队列
func NewDelayQueue() *DelayQueue {
dq := &DelayQueue{
buckets: make(map[time.Duration]*list.List),
}
heap.Init(&dq.taskQueue)
go dq.loop()
return dq
}
// 循环执行任务
func (dq *DelayQueue) loop() {
for {
time.Sleep(time.Second)
dq.mutex.Lock()
for {
now := time.Now()
if dq.taskQueue.Len() == 0 {
break
}
if dq.taskQueue[0].Delay.After(now) {
break
}
task := heap.Pop(&dq.taskQueue).(*Task)
go task.Handler()
dq.removeFromBucket(task)
}
dq.mutex.Unlock()
}
}
// 添加任务到延迟队列中
func (dq *DelayQueue) AddTask(task *Task) {
task.Delay = time.Now().Add(task.Delay.Sub(time.Now()))
bucket := dq.getBucket(task.Delay.Sub(time.Now()))
bucket.PushBack(task)
heap.Push(&dq.taskQueue, task)
}
// 从延迟桶中移除任务
func (dq *DelayQueue) removeFromBucket(task *Task) {
delay := task.Delay.Sub(time.Now())
bucket := dq.getBucket(delay)
for e := bucket.Front(); e != nil; e = e.Next() {
if e.Value == task {
bucket.Remove(e)
break
}
}
}
// 获取延迟时间所对应的桶
func (dq *DelayQueue) getBucket(delay time.Duration) *list.List {
for d := range dq.buckets {
if d == delay {
return dq.buckets[d]
}
if d > delay {
break
}
}
bucket := list.New()
dq.buckets[delay] = bucket
return bucket
}
// 测试代码
func main() {
dq := NewDelayQueue()
task1 := &Task{
ID: 1,
Delay: time.Now().Add(time.Second * 3),
// 处理函数
Handler: func() {
fmt.Println("Task 1 is executed.")
},
}
task2 := &Task{
ID: 2,
Delay: time.Now().Add(time.Second * 5),
Handler: func() {
fmt.Println("Task 2 is executed.")
},
}
task3 := &Task{
ID: 3,
Delay: time.Now().Add(time.Second * 2),
Handler: func() {
fmt.Println("Task 3 is executed.")
},
}
dq.AddTask(task1)
dq.AddTask(task2)
dq.AddTask(task3)
time.Sleep(time.Second * 10)
}
