任务处理与中间件
1 Handler 接口
type Handler interface {
ProcessTask(ctx context.Context, t *Task) error
}任何实现了 ProcessTask 方法的类型都可以作为 Handler。
1.1 HandlerFunc 适配器
type HandlerFunc func(context.Context, *Task) error
func (f HandlerFunc) ProcessTask(ctx context.Context, task *Task) error {
return f(ctx, task)
}使用方式：
// 直接传入函数
mux.HandleFunc("email:send", func(ctx context.Context, t *asynq.Task) error {
// 处理逻辑
return nil
})
// 或传入预定义的函数
mux.HandleFunc("email:send", tasks.HandleEmailTask)1.2 自定义 Handler 类型
当需要为 Handler 注入依赖（如 DB 连接、配置）时，使用结构体实现接口：
type EmailHandler struct {
mailer *mail.Client
db *sql.DB
template *template.Engine
}
func (h *EmailHandler) ProcessTask(ctx context.Context, t *asynq.Task) error {
var p EmailPayload
if err := json.Unmarshal(t.Payload(), &p); err != nil {
return fmt.Errorf("unmarshal: %w", asynq.SkipRetry)
}
tmpl, err := h.template.Load(p.TemplateID)
if err != nil {
return fmt.Errorf("template load: %w", err) // 可重试
}
return h.mailer.Send(ctx, p.To, tmpl.Subject, tmpl.Body)
}
// 注册
handler := &EmailHandler{mailer: mailer, db: db, template: tmpl}
mux.Handle(tasks.TypeEmailSend, handler)2 ServeMux 路由
mux := asynq.NewServeMux()
// HandleFunc：按 Task Type 字符串路由
mux.HandleFunc("email:send", handleEmailSend)
mux.HandleFunc("order:cancel", handleOrderCancel)
mux.HandleFunc("report:export", handleReportExport)
// Handle：注册 Handler 接口的实现
mux.Handle("image:process", &ImageProcessHandler{processor})2.1 路由匹配规则
ServeMux 使用精确匹配——Task Type 字符串必须完全一致。不同于 http.ServeMux，不支持前缀匹配或模式匹配。
"email:send" → handleEmailSend
"email:send.v2"→ 不匹配 "email:send"（完全不同的类型）💡 需要版本化任务类型时，使用显式版本号：
const (
TypeEmailSendV1 = "email:send:v1"
TypeEmailSendV2 = "email:send:v2"
)2.2 未匹配的处理
如果 Task Type 没有注册任何 Handler，ServeMux 返回 SkipRetry 错误，任务直接进入 archived。
3 中间件
type MiddlewareFunc func(Handler) Handler中间件在注册的 Handler 外层形成洋葱模型：
┌─────────────┐
│ Middleware1 │
│ ┌─────────┐ │
│ │ Mid2 │ │
│ │ ┌─────┐ │ │
│ │ │Handler│ │ │
│ │ └─────┘ │ │
│ └─────────┘ │
└─────────────┘3.1 内置常用中间件模式
日志中间件：
func LoggingMiddleware(next asynq.Handler) asynq.Handler {
return asynq.HandlerFunc(func(ctx context.Context, t *asynq.Task) error {
start := time.Now()
taskID, _ := asynq.GetTaskID(ctx)
queueName, _ := asynq.GetQueueName(ctx)
log.Printf("[START] type=%s id=%s queue=%s", t.Type(), taskID, queueName)
err := next.ProcessTask(ctx, t)
duration := time.Since(start)
if err != nil {
log.Printf("[FAIL] type=%s id=%s duration=%s err=%v", t.Type(), taskID, duration, err)
} else {
log.Printf("[DONE] type=%s id=%s duration=%s", t.Type(), taskID, duration)
}
return err
})
}Panic 恢复中间件：
func RecoveryMiddleware(next asynq.Handler) asynq.Handler {
return asynq.HandlerFunc(func(ctx context.Context, t *asynq.Task) (err error) {
defer func() {
if r := recover(); r != nil {
buf := make([]byte, 4096)
n := runtime.Stack(buf, false)
err = fmt.Errorf("panic: %v\n%s", r, buf[:n])
}
}()
return next.ProcessTask(ctx, t)
})
}链路追踪中间件：
func TracingMiddleware(next asynq.Handler) asynq.Handler {
return asynq.HandlerFunc(func(ctx context.Context, t *asynq.Task) error {
// 从 Headers 中提取 TraceID 并注入 context
traceID := t.Headers["X-Trace-ID"]
if traceID != "" {
ctx = context.WithValue(ctx, traceKey, traceID)
}
// 如果没有，生成新的
if traceID == "" {
traceID = uuid.New().String()
ctx = context.WithValue(ctx, traceKey, traceID)
}
span, ctx := opentracing.StartSpanFromContext(ctx, t.Type())
defer span.Finish()
return next.ProcessTask(ctx, t)
})
}指标采集中间件：
func MetricsMiddleware(next asynq.Handler) asynq.Handler {
return asynq.HandlerFunc(func(ctx context.Context, t *asynq.Task) error {
taskType := t.Type()
taskCounter.WithLabelValues(taskType, "received").Inc()
timer := prometheus.NewTimer(taskDuration.WithLabelValues(taskType))
err := next.ProcessTask(ctx, t)
timer.ObserveDuration()
if err != nil {
taskCounter.WithLabelValues(taskType, "failed").Inc()
} else {
taskCounter.WithLabelValues(taskType, "completed").Inc()
}
return err
})
}3.2 注册中间件
mux := asynq.NewServeMux()
// 中间件按 Use 顺序由外向内包裹
mux.Use(RecoveryMiddleware) // 最外层：panic 恢复
mux.Use(LoggingMiddleware) // 第二层：日志
mux.Use(TracingMiddleware) // 第三层：追踪
mux.Use(MetricsMiddleware) // 最内层：指标
// 注册 Handler
mux.HandleFunc("email:send", handleEmailSend)最终执行顺序：
Recovery → Logging → Tracing → Metrics → Handler → Metrics → Tracing → Logging → Recovery3.3 中间件中的错误处理
func RetrySuppressMiddleware(next asynq.Handler) asynq.Handler {
return asynq.HandlerFunc(func(ctx context.Context, t *asynq.Task) error {
err := next.ProcessTask(ctx, t)
// 对特定错误类型降级处理
if IsDegradedError(err) {
log.Printf("[DEGRADED] task=%s err=%v", t.Type(), err)
return nil // 标记为成功
}
return err
})
}4 Context 辅助函数
Asynq 在 Handler 的 context.Context 中注入了以下信息：
4.1 GetTaskID
func Handler(ctx context.Context, t *asynq.Task) error {
taskID, ok := asynq.GetTaskID(ctx)
if ok {
log.Printf("Processing task: %s", taskID)
}
// ...
}4.2 GetQueueName
func Handler(ctx context.Context, t *asynq.Task) error {
queueName, ok := asynq.GetQueueName(ctx)
if ok {
log.Printf("Task from queue: %s", queueName)
}
// ...
}5 BaseContext 注入全局信息
v0.22.0+
srv := asynq.NewServer(redisOpt, asynq.Config{
BaseContext: func() context.Context {
// 在所有 Handler context 中注入全局 logger
ctx := context.WithValue(context.Background(), loggerKey, appLogger)
// 注入 tracer
ctx = context.WithValue(ctx, tracerKey, appTracer)
return ctx
},
})Handler 中获取：
func Handler(ctx context.Context, t *asynq.Task) error {
logger := ctx.Value(loggerKey).(*slog.Logger)
logger.Info("processing task", "type", t.Type())
// ...
}6 并发安全的 Handler
Handler 会被多个 goroutine 并发调用，必须保证线程安全：
type SafeHandler struct {
mu sync.Mutex
cache map[string]*CachedData // 需要加锁访问
db *sql.DB // database/sql 的连接池已内置线程安全
}
func (h *SafeHandler) ProcessTask(ctx context.Context, t *asynq.Task) error {
// ✅ database/sql 的连接池是线程安全的，无需额外加锁
rows, err := h.db.QueryContext(ctx, "SELECT ...")
// ✅ 访问共享 map 需要加锁
h.mu.Lock()
data := h.cache[t.Type()]
h.mu.Unlock()
// ...
}🚨 陷阱：不要在 Handler 中直接写闭包捕获的共享变量。避免使用包级全局变量，改为通过 Handler 结构体注入。
7 完整示例：带依赖注入的 Worker
package main
import (
"context"
"log"
"os"
"os/signal"
"syscall"
"github.com/hibiken/asynq"
)
type AppHandler struct {
emailHandler *EmailHandler
orderHandler *OrderHandler
logger *slog.Logger
}
func NewAppHandler(db *sql.DB, mailer *mail.Client) *AppHandler {
return &AppHandler{
emailHandler: &EmailHandler{mailer: mailer},
orderHandler: &OrderHandler{db: db},
logger: slog.Default(),
}
}
func (h *AppHandler) Register(mux *asynq.ServeMux) {
// 注册中间件
mux.Use(h.loggingMiddleware)
mux.Use(h.recoveryMiddleware)
// 注册路由
mux.HandleFunc("email:send", h.emailHandler.ProcessTask)
mux.HandleFunc("order:cancel", h.orderHandler.ProcessTask)
}
func (h *AppHandler) loggingMiddleware(next asynq.Handler) asynq.Handler {
return asynq.HandlerFunc(func(ctx context.Context, t *asynq.Task) error {
start := time.Now()
err := next.ProcessTask(ctx, t)
h.logger.Info("task processed",
"type", t.Type(),
"duration", time.Since(start),
"error", err,
)
return err
})
}
func (h *AppHandler) recoveryMiddleware(next asynq.Handler) asynq.Handler {
return asynq.HandlerFunc(func(ctx context.Context, t *asynq.Task) (err error) {
defer func() {
if r := recover(); r != nil {
h.logger.Error("panic recovered", "type", t.Type(), "panic", r)
err = fmt.Errorf("panic: %v", r)
}
}()
return next.ProcessTask(ctx, t)
})
}
func main() {
srv := asynq.NewServer(
asynq.RedisClientOpt{Addr: "localhost:6379"},
asynq.Config{Concurrency: 10},
)
h := NewAppHandler(db, mailer)
mux := asynq.NewServeMux()
h.Register(mux)
if err := srv.Run(mux); err != nil {
log.Fatal(err)
}
}