最佳实践与陷阱

// ✅ 推荐：domain:action 格式
const (
    TypeEmailSend        = "email:send"
    TypeEmailSendWelcome = "email:send:welcome"  // 子类型
    TypeOrderCancel      = "order:cancel"
    TypeOrderRefund      = "order:refund"
    TypeReportExport     = "report:export"
    TypeImageProcess     = "image:process"
)

// ✅ 版本化（当需要不兼容变更时）
const (
    TypeEmailSendV1 = "email:send:v1"
    TypeEmailSendV2 = "email:send:v2"
)

// ❌ 避免：无命名空间，容易冲突
const TypeSend = "send"

// 策略 1：数据库唯一约束
func HandleOrderCancel(ctx context.Context, t *asynq.Task) error {
    var p OrderCancelPayload
    json.Unmarshal(t.Payload(), &p)

    // INSERT ... ON DUPLICATE KEY 或 UPSERT
    err := db.CancelOrder(ctx, p.OrderID)
    if errors.Is(err, ErrAlreadyCancelled) {
        return nil // 已取消，幂等
    }
    return err
}

// 策略 2：分布式锁
func HandleUniqueAction(ctx context.Context, t *asynq.Task) error {
    lockKey := fmt.Sprintf("lock:task:%s", t.Type())
    locked, err := redisClient.SetNX(ctx, lockKey, "1", 30*time.Second).Result()
    if err != nil || !locked {
        return nil // 正在处理，跳过
    }
    defer redisClient.Del(ctx, lockKey)

    // 执行实际逻辑
    return doWork(ctx, t)
}

// 策略 3：状态机
func HandleStateTransition(ctx context.Context, t *asynq.Task) error {
    var p Payload
    json.Unmarshal(t.Payload(), &p)

    // 只有状态为 "pending" 时才执行
    rowsAffected, err := db.TransitionState(ctx, p.ID, "pending", "processing")
    if rowsAffected == 0 {
        return nil // 已被其他 Worker 处理，幂等
    }
    return err
}

asynq.Config{
    Queues: map[string]int{
        "critical":      6,  // 核心业务
        "default":       3,  // 常规任务
        "external_api": 2,  // 调用外部 API（可单独暂停此队列）
        "low":           1,  // 低优先级
    },
}

asynq pause external_api

// ✅ 推荐：提供上下文
return fmt.Errorf("send email to %s: smtp error: %w", p.To, err)

// ✅ 推荐：标明是否可重试
return fmt.Errorf("transient db error: %w", err)         // 可重试
return fmt.Errorf("invalid user: %w", asynq.SkipRetry)   // 不重试

// ❌ 避免：无上下文的错误
return err

// ❌ 在 Handler 中做大量同步 IO
func BadHandler(ctx context.Context, t *asynq.Task) error {
    // 每个任务都创建新的 HTTP Client
    resp, _ := http.Get("https://api.example.com")
    return nil
}

// ✅ 复用 HTTP Client
var httpClient = &http.Client{
    Timeout: 30 * time.Second,
    Transport: &http.Transport{
        MaxIdleConns:    100,
        IdleConnTimeout: 90 * time.Second,
    },
}

func GoodHandler(ctx context.Context, t *asynq.Task) error {
    resp, _ := httpClient.Get("https://api.example.com")
    return nil
}

// ❌ 在 Handler 中连接数据库（频繁创建连接）
func BadHandler(ctx context.Context, t *asynq.Task) error {
    db, _ := sql.Open("postgres", dsn)
    defer db.Close()
    // ...
}

// ✅ 注入连接池
type Handler struct {
    db *sql.DB // 全局连接池
}

# Worker Dockerfile
FROM golang:1.24-alpine AS builder
WORKDIR /app
COPY . .
RUN go build -o /worker ./cmd/worker

FROM alpine:3.20
COPY --from=builder /worker /usr/local/bin/worker
ENTRYPOINT ["/usr/local/bin/worker"]

srv := asynq.NewServer(redisOpt, asynq.Config{
    Concurrency:     20,
    ShutdownTimeout: 30 * time.Second, // 关键：给活跃任务足够时间
})

// K8s: 确保 terminationGracePeriodSeconds > ShutdownTimeout

# Kubernetes Pod 配置
spec:
  terminationGracePeriodSeconds: 60  # 应大于 ShutdownTimeout
  containers:
  - name: worker
    lifecycle:
      preStop:
        exec:
          command: ["/bin/sh", "-c", "sleep 5"]  # 等待 Endpoint 更新