延迟队列实现（2）

1. 定义任务结构体 首先，我们需要定义任务结构体，用于存储任务相关的信息。任务结构体至少应该包含以下几个字段：

• ID：任务唯一标识符 *
• Topic：任务主题，用于区分不同类型的任务*
• Delay：任务延迟时间，即从任务加入队列到任务执行之间的时间间隔*
• TTR（Time To Run）：任务最长执行时间，即任务从开始执行到执行结束的时间间隔上限*
• Body：任务内容，即需要执行的具体操作*

2. 定义延迟队列 其次，我们需要定义延迟队列结构体，用于存储所有的任务。延迟队列应该包括以下几个组件：

• 任务队列：一个优先队列，用于存储所有任务。按照任务的延迟时间进行排序，最先执行的任务排在队列前面。*
• 延迟桶：一个时间轮，用于管理任务的延迟时间。我们可以设置多个桶，每个桶表示一个时间段，比如1秒、10秒、1分钟等等。每个桶内部维护一个任务队列，用于存储所有延迟时间在该桶时间范围内的任务。每次时间轮tick时，我们需要将当前桶的所有任务添加到任务队列中，以便执行。*
• 定时器：一个定时器，用于定时执行任务队列中的任务。每次定时器tick时，我们从任务队列中取出最先执行的任务，并执行之。如果任务执行超时，我们可以将任务重新加入队列，并根据重试次数设置不同的延迟时间。*

3. 添加任务 当有新任务需要添加到延迟队列中时，我们需要将任务加入到延迟桶中。首先，我们根据任务的延迟时间计算出应该加入到哪个桶中。然后，将任务加入到该桶的任务队列中。如果任务队列为空，则需要重新设置定时器的时间。

4. 执行任务 当定时器tick时，我们从任务队列中取出最先执行的任务，并执行之。如果任务执行成功，则不需要进行任何操作；如果任务执行失败，则可以将任务重新加入队列，并根据重试次数设置不同的延迟时间。在任务执行完成后，我们需要重新设置定时器的时间。

5. 删除任务 当某个任务需要从延迟队列中删除时，我们需要在任务队列和对应的延迟桶中删除该任务。

以上是一个完整的延迟队列系统的实现步骤。具体实现时，我们可以根据需求进行一些调整和优化，比如设置任务优先级、增加任务超时监控、支持任务重试、支持任务取消等等。

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)
}