delayqueue的Go简易实现

一、缘起

很多时候，业务有“在一段时间之后，完成一个工作任务”的需求。
例如：滴滴打车订单完成后，如果用户一直不评价，48小时后会将自动评价为5星。
一般来说怎么实现这类“48小时后自动评价为5星”需求呢？
常见方案：启动一个cron定时任务，每小时跑一次，将完成时间超过48小时的订单取出，置为5星，并把评价状态置为已评价。
假设订单表的结构为：t_order(oid, finish_time, stars, status, …)，更具体的，定时任务每隔一个小时会这么做一次：
select oid from t_order where finish_time > 48hours and status=0;
update t_order set stars=5 and status=1 where oid in[…];
如果数据量很大，需要分页查询，分页update，这将会是一个for循环。
方案的不足：
（1）轮询效率比较低
（2）每次扫库，已经被执行过记录，仍然会被扫描（只是不会出现在结果集中），有重复计算的嫌疑
（3）时效性不够好，如果每小时轮询一次，最差的情况下，时间误差会达到1小时
（4）如果通过增加cron轮询频率来减少（3）中的时间误差，（1）中轮询低效和（2）中重复计算的问题会进一步凸显

二、高效延时消息设计与实现

高效延时消息，包含两个重要的数据结构：
（1）环形队列，例如可以创建一个包含3600个slot的环形队列（本质是个数组）
（2）任务集合，环上每一个slot是一个Set
同时，启动一个timer，这个timer每隔1s，在上述环形队列中移动一格，有一个Current Index指针来标识正在检测的slot。
Task结构中有两个很重要的属性：
（1）Cycle-Num：当Current Index第几圈扫描到这个Slot时，执行任务
（2）Task-Function：需要执行的任务指针

使用了“延时消息”方案之后，“订单48小时后关闭评价”的需求，只需将在订单关闭时，触发一个48小时之后的延时消息即可：
（1）无需再轮询全部订单，效率高
（2）一个订单，任务只执行一次
（3）时效性好，精确到秒（控制timer移动频率可以控制精度）

三、Go代码实现

package main

import (
    "errors"
    "fmt"
    "time"
)

const cicleSectionNum = 10

type TaskFunc func(args ...any)

type Task struct {
    runTime  time.Time //初次运行时间
    cycleNum int       //需要第几圈
    curIndex int       //当前运行到第几格
    //执行的函数
    exec   TaskFunc
    params []any
}

type DelayMessage struct {
    cycleNum  int //当前运行到第几圈了
    curIndex  int //当前运行到第几格
    slots     [cicleSectionNum]map[string]*Task
    closed    chan bool
    taskClose chan bool
    timeClose chan bool
    startTime time.Time
}

func NewDelayMessage() *DelayMessage {
    dm := &DelayMessage{
        cycleNum:  0,
        curIndex:  0,
        closed:    make(chan bool),
        taskClose: make(chan bool),
        timeClose: make(chan bool),
        startTime: time.Now(),
    }
    for i := 0; i < cicleSectionNum; i++ {
        dm.slots[i] = make(map[string]*Task)
    }
    return dm
}

func (dm *DelayMessage) Start() {
    go dm.taskLoop()
    go dm.timeLoop()
    select {
        case <-dm.closed:
        dm.taskClose <- true
        dm.timeClose <- true
        break
    }
}

func (dm *DelayMessage) Stop() {
    dm.closed <- true
}

func (dm *DelayMessage) taskLoop() {
    defer func() {
        fmt.Println("任务遍历结束！")
    }()
    for {
        select {
            case <-dm.taskClose:
            return
            default:
            {
                tasks := dm.slots[dm.curIndex]
                if len(tasks) > 0 {
                    for k, v := range tasks {
                        if v.cycleNum == dm.cycleNum {
                            go v.exec(v.params...)
                            delete(tasks, k)
                        }
                    }
                }
            }
        }

    }
}

func (dm *DelayMessage) timeLoop() {
    defer func() {
        fmt.Println("时间遍历结束!")
    }()
    tick := time.NewTicker(time.Second)
    for {
        select {
            case <-dm.timeClose:
            return
            case <-tick.C:
            fmt.Println(time.Now().Format(time.DateTime))
            dm.curIndex = (dm.curIndex + 1) % cicleSectionNum
            if dm.curIndex == 0 {
                dm.cycleNum += 1
            }
            fmt.Println("当前循环时间", dm.cycleNum, dm.curIndex)
        }
    }

}

func (dm *DelayMessage) AddTask(t time.Time, key string, exec TaskFunc, params []any) error {
    if dm.startTime.After(t) {
        return errors.New("时间错误")
    }
    //当前时间与指定时间相差秒数
    subSecond := t.Unix() - dm.startTime.Unix()
    //计算循环次数
    cycleNum := int(subSecond / cicleSectionNum)
    //计算任务所在的slots的下标
    ix := subSecond % cicleSectionNum
    //把任务加入tasks中
    tasks := dm.slots[ix]
    if _, ok := tasks[key]; ok {
        return errors.New("该slots中已存在key为" + key + "的任务")
    }
    tasks[key] = &Task{
        runTime:  t,
        cycleNum: cycleNum,
        curIndex: int(ix),
        exec:     exec,
        params:   params,
    }
    return nil
}

func main() {
    fmt.Println("abc")
    dm := NewDelayMessage()
    //添加任务
    dm.AddTask(time.Now().Add(time.Second*3), "test1", func(args ...any) {
        fmt.Println(args...)
    }, []any{1, 2, 3})

    dm.AddTask(time.Now().Add(time.Second*3), "test1", func(args ...any) {
        fmt.Println(args...)
    }, []any{2, 2, 3})

    dm.AddTask(time.Now().Add(time.Second*3), "test2", func(args ...any) {
        fmt.Println(args...)
    }, []any{3, 2, 3})

    dm.AddTask(time.Now().Add(time.Second*11), "test11", func(args ...any) {
        fmt.Println(args...)
    }, []any{11, 2, 3})

    dm.AddTask(time.Now().Add(time.Second*12), "test11", func(args ...any) {
        fmt.Println(args...)
    }, []any{11, 2, 3})

    //40秒后关闭
    time.AfterFunc(time.Second*40, func() {
        dm.Stop()
    })
    dm.Start()
}