Go 消息队列实战：用 asynq 基于 Redis 做异步任务、定时与重试

为什么用 asynq

当邮件发送、图片压缩、报表生成这类耗时操作被塞进 HTTP 请求里，接口就会变慢甚至超时。asynq 是一个基于 Redis 的轻量任务队列，它用 List 存待处理任务、用 ZSet 存延迟与定时任务，提供入队、定时、重试、中间件等开箱即用的能力，适合中小型服务的异步化改造。

定义任务与处理器

任务由类型名和 JSON 负载组成，处理器实现 ProcessTask 接口。这里用泛型结构承载业务参数。

package main

import (
	"context"
	"encoding/json"
	"fmt"
	"log"

	"github.com/hibiken/asynq"
)

type EmailPayload struct {
	UserID int    `json:"user_id"`
	Title  string `json:"title"`
}

func handleEmailTask(ctx context.Context, t *asynq.Task) error {
	var p EmailPayload
	if err := json.Unmarshal(t.Payload(), &p); err != nil {
		return fmt.Errorf("解析负载失败: %w", err)
	}
	return sendEmail(p.UserID, p.Title)
}

func sendEmail(userID int, title string) error {
	log.Printf("向用户 %d 发送邮件: %s", userID, title)
	return nil
}

入队与延迟任务

客户端把任务推入 Redis。用 asynq.ProcessIn 设定相对延迟，用 asynq.MaxRetry 控制失败重试次数，asynq.Queue 指定队列优先级。

func enqueue(client *asynq.Client) error {
	payload, _ := json.Marshal(EmailPayload{UserID: 42, Title: "欢迎注册"})

	task := asynq.NewTask("email:welcome", payload)
	// 10 秒后发送，最多重试 3 次，进入高优队列
	_, err := client.Enqueue(
		task,
		asynq.ProcessIn(10*time.Second),
		asynq.MaxRetry(3),
		asynq.Queue("critical"),
	)
	return err
}

启动 Worker 与定时任务

Worker 从 Redis 拉取任务并交给对应处理器。Scheduler 负责周期性任务，例如每天凌晨清理一次日志。

func main() {
	redisOpt := asynq.RedisClientOpt{Addr: "127.0.0.1:6379"}

	srv := asynq.NewServer(redisOpt, asynq.Config{Concurrency: 10})
	mux := asynq.NewServeMux()
	mux.HandleFunc("email:welcome", handleEmailTask)

	// 定时任务：每天 00:00 执行
	scheduler := asynq.NewScheduler(redisOpt, &asynq.SchedulerOpts{})
	scheduler.Register("@daily", asynq.NewTask("cleanup:logs", nil))

	go func() { log.Fatal(scheduler.Run()) }()
	log.Fatal(srv.Run(mux))
}

中间件与错误处理

asynq 的 middleware 可以统一做日志、耗时统计和 panic 恢复，避免每个处理器重复写样板。

func loggingMiddleware(h asynq.Handler) asynq.Handler {
	return asynq.HandlerFunc(func(ctx context.Context, t *asynq.Task) error {
		start := time.Now()
		err := h.ProcessTask(ctx, t)
		log.Printf("任务 %s 耗时 %v 结果 %v", t.Type(), time.Since(start), err)
		return err
	})
}

// 启动时包裹：srv.Run(asynq.NewServerMiddleware(loggingMiddleware)(mux))

• 任务类型用清晰前缀如 email:welcome，便于在面板里区分业务。
• 失败任务靠 MaxRetry 自动重试，超过上限进入 dead queue 待人工排查。
• 重活放高并发队列，轻活放默认队列，用 Queue 做优先级隔离。