Golang高性能内存线程安全延迟队列插件库dqueue的使用

什么是dqueue?

dqueue是一个Golang实现的延迟任务队列库，它允许你创建在未来某个时间点执行的任务，比如"3分钟后执行这个任务"或"15秒后执行那个任务"。然后可以通过并发goroutine消费这些任务，任务会按照正确的顺序提供给消费者。

基本用法

package main

import (
	"fmt"
	"time"
	"github.com/vodolaz095/dqueue"
)

func main() {
	handler := dqueue.New() // 导入"github.com/vodolaz095/dqueue"

	// payload可以是任何类型 - 数字、字符串、buffer、结构体...
	something := "task"

	// 创建未来执行的任务
	handler.ExecuteAt(something, time.Now().Add(time.Minute)) // 在指定时间执行
	handler.ExecuteAfter(something, time.Minute) // 1分钟后执行

	// 获取准备就绪的任务
	task, ready := handler.Get()
	if ready { // 任务已就绪
		fmt.Printf("任务 %s 准备在 %s 执行\n",
			task.Payload.(string), 
			task.ExecuteAt.Format(time.Kitchen),
		)
	} else {
		fmt.Println("没有准备就绪的任务")
	}
	
	// 获取队列中剩余任务数量
	tasksInQueue := handler.Len()

	// 导出所有任务，例如在关闭应用前保存所有待处理队列
	tasks := handler.Dump()

	// 清空队列
	handler.Prune()
}

并发消费者示例

package main

import (
	"context"
	"fmt"
	"sync"
	"time"
	"github.com/vodolaz095/dqueue"
)

func ProcessTask(task dqueue.Task) error {
	// 处理任务的逻辑
	fmt.Printf("处理任务: %v\n", task.Payload)
	return nil
}

func main() {
	handler := dqueue.New()

	// 发布任务
	something := "task" // payload可以是任何类型
	handler.ExecuteAt(something, time.Now().Add(time.Minute))
	handler.ExecuteAfter(something, time.Minute)

	// 创建全局上下文，在应用停止时取消
	wg := sync.WaitGroup{}
	mainCtx, mainCancel := context.WithTimeout(context.Background(), 3*time.Second)
	defer mainCancel()

	// 启动并发消费者
	for j := 0; j < 10; j++ {
		wg.Add(1)
		go func(workerNumber int, initialCtx context.Context) {
			ctx, cancel := context.WithCancel(initialCtx)
			defer cancel()
			ticker := time.NewTicker(time.Millisecond)
			for {
				select {
				case <-ticker.C:
					task, ready := handler.Get()
					if ready { // 任务已就绪
						err := ProcessTask(task)
						if err != nil { // 例如，将消息重新排队在1分钟后交付
							handler.ExecuteAfter(something, time.Minute)
						}
					}
					break
				case <-ctx.Done():
					fmt.Printf("关闭worker %v, 队列中还有%v个任务\n", workerNumber, handler.Len())
					wg.Done()
					ticker.Stop()
					return
				}
			}
		}(j, mainCtx)
	}
	wg.Wait()

	// 查看剩余任务，以便应用重启时可以恢复
	tasks := handler.Dump()
}

主要功能

1. ExecuteAt(payload interface{}, when time.Time) - 在指定时间执行任务
2. ExecuteAfter(payload interface{}, duration time.Duration) - 在指定时间间隔后执行任务
3. Get() (Task, bool) - 获取准备就绪的任务
4. Len() int - 获取队列中任务数量
5. Dump() []Task - 导出所有任务
6. Prune() - 清空队列

这个库是线程安全的，适合在高并发环境下使用。如果你需要更高性能但更不灵活的解决方案，可以考虑基于泛型的替代方案。