redisqueue

package

v1.0.3 Latest Latest Go to latest Published: Jul 7, 2026 License: MIT Imports: 7 Imported by: 0

Details

Package redisqueue provides a Redis backed repository.TaskQueue.

It is an optional driver kept out of the repository/query core so that local/in-memory embedding does not pull in the go-redis dependency.

This section is empty.

This section is empty.

func StartRecoveryLoop(ctx context.Context, q *RedisQueue)

StartRecoveryLoop 周期性把超时未完成的任务重新推回主队列（进程重启/崩溃恢复）。阻塞运行，通常以 goroutine 启动；随 ctx 取消而退出。

type RedisQueue struct {
	// contains filtered or unexported fields
}

RedisQueue 主要用于Worker 崩溃时，保障任务ID不丢，可以恢复执行中任务

func New(
	client *redis.Client,
	queueKey,
	processingKey,
	deadKey string,
	timeout time.Duration,
) *RedisQueue

NewRedisQueue 使用 Redis 的 BRPOPLPUSH 或 RPOPLPUSH 模式：

主队列：video_task_queue（存放待执行任务）
处理中队列：video_task_processing（存放正在处理的任务）
执行流程： 1. Worker 从 video_task_queue 阻塞弹出任务 → push 到 video_task_processing 2. 执行任务 3. 执行成功 → 从 video_task_processing 删除 4. 执行失败 → 视重试策略可重新 push 回主队列

func (q *RedisQueue) Ack(ctx context.Context, taskID int64) error

Ack 删除 processing 队列任务（表示成功完成）

func (q *RedisQueue) MoveToDead(ctx context.Context, taskID int64) error

MoveToDead 重试3次失败，移动任务到到死信队列

func (q *RedisQueue) PopAndReserve(ctx context.Context) (int64, error)

func (q *RedisQueue) Push(ctx context.Context, taskID int64) error

Push 往主队列添加任务

func (q *RedisQueue) Recover(ctx context.Context) error

Recover 将超时未完成任务重新推回主队列只恢复“过期任务”