介绍

delay queue是基于Redis Zset+Cron/Ticker实现的Golang版延时队列。

• 实现方案：任务定时执行时主动轮询小于当前时间的元素, 取出符合条件元素执行任务，完成任务后删除该元素。
• 延时执行：支持延迟多少秒和延迟到具体时间（秒时间戳）执行。
• 定时执行：支持cron和ticker定时执行，cron在每次执行时会开启一个协程，因此该方案存在执行方法Execute方法执行很慢时协程堆积问题。
• redis key: keyPrefix:jobID
• cron id: keyPrefix:jobID
• batchLimit: dq.New方法中的batchLimit是每次从zset中获取的元素数量，根据实际情况调整，设置为0则默认10000

安装

go get -u github.com/hanxiang-s/delay_queue

推荐使用go.mod

使用

type JobActionSMS struct{}

// ID 任务ID
func (j *JobActionSMS) ID() string {
    return "JobActionSMS"
}

// Scheduler 任务定时执行(cron方案)，value为cron表达式，执行时从zset中获取0<score<=当前时间的member去执行任务
func (j *JobActionSMS) Scheduler() pkg.Scheduler {
    return pkg.Scheduler{
        Type:  pkg.SchedulerTypeCron,
        Value: "@every 1s",
    }
}

/*
// Scheduler 任务定时执行(ticker方案)，value为时间间隔，单位秒，执行时从zset中获取0<score<=当前时间的member去执行任务
func (j *JobActionSMS) Scheduler() pkg.Scheduler {
    return pkg.Scheduler{
        Type:  pkg.SchedulerTypeTicker,
        Value: "1",
    }
}
*/

// Execute 任务执行方法
func (j *JobActionSMS) Execute(member any) error {
    phone, _ := member.(string)
    fmt.Printf("sending sms to %s,time:%v\n", phone, time.Now().Format("2006-01-02 15:04:05"))
    return nil
}

func main() {
    redisOpt := &redis.Options{Addr: "127.0.0.1:6379", Password: "password"}
    cli := dq.New("test", 0, redisOpt)
    if err := cli.Register(&JobActionSMS{}); err != nil {
        log.Fatal(err)
    }
    fmt.Println("add job: ", time.Now().Format("2006-01-02 15:04:05"))
    if err := cli.AddJob(pkg.DelayJob{
		ID:        (&JobActionSMS{}).ID(),
        Type:      pkg.DelayTypeDuration, //延迟N秒执行
        DelayTime: 10,                    //延迟秒数
        Member:    "138****0000",
    }); err != nil {
        log.Fatal(err)
    }
    if err := cli.AddJob(pkg.DelayJob{
        ID:        (&JobActionSMS{}).ID(),
        Type:      pkg.DelayTypeDate,      //延迟到具体时间执行
        DelayTime: time.Now().Unix() + 10, //执行时间的秒时间戳
        Member:    "138****1111",
	}); err != nil {
        log.Fatal(err)
    }
    time.Sleep(time.Second * 30)
}