重试与错误处理

Asynq

重试与错误处理

1 默认重试机制

1.1 基本行为

默认最大重试次数：25 次
重试策略：指数退避（exponential backoff）
退避公式：min(initialDelay * 2^(n-1), maxDelay)，其中 initialDelay ≈ 1s，maxDelay ≈ 30min
重试次数耗尽后：任务进入 archived 状态

重试时间线（近似）：
  第 1 次失败 → 等待 ~1s  → 第 2 次失败 → 等待 ~2s
  → 第 3 次失败 → 等待 ~4s → ... → 第 25 次失败 → archived

1.2 修改最大重试次数

// 方式 1：在 Enqueue 时指定
client.Enqueue(task, asynq.MaxRetry(3))

// 方式 2：在 NewTask 时指定
task := asynq.NewTask("email:send", payload, asynq.MaxRetry(5))

2 错误分类与处理策略

2.1 SkipRetry —— 永久性错误

var SkipRetry = errors.New("skip retry")

返回此错误后，任务立即归档，不再重试。适用于无法通过重试修复的永久性错误。

func Handler(ctx context.Context, t *asynq.Task) error {
    var p Payload
    if err := json.Unmarshal(t.Payload(), &p); err != nil {
        // Payload 格式错误，重试无意义
        return fmt.Errorf("unmarshal payload: %w", asynq.SkipRetry)
    }

    user, err := db.GetUser(ctx, p.UserID)
    if errors.Is(err, sql.ErrNoRows) {
        // 用户不存在，重试无意义
        return fmt.Errorf("user %d not found: %w", p.UserID, asynq.SkipRetry)
    }

    return nil
}

🚨 注意 wrapping 方式：asynq.SkipRetry 必须通过 %w 包装（fmt.Errorf("...: %w", asynq.SkipRetry)）。Asynq 内部使用 errors.Is 判断。

2.2 普通 Error —— 临时性错误

返回普通 error 会触发自动重试：

func Handler(ctx context.Context, t *asynq.Task) error {
    resp, err := http.Get("https://api.example.com/data")
    if err != nil {
        // 网络错误 → 可能是临时的，触发重试
        return fmt.Errorf("api call failed: %w", err)
    }
    if resp.StatusCode == 429 {
        // 限流 → 稍后重试
        return fmt.Errorf("rate limited")
    }
    return nil
}

2.3 RevokeTask —— 撤销任务

v0.25.0+

// 直接从 Handler 中撤销任务（不进行重试，也不归档）
return fmt.Errorf("task revoked: %w", asynq.RevokeTask)

与 SkipRetry 的区别：RevokeTask 不会进入 archived，而是直接丢弃。

2.4 决策表

场景	应返回	行为
Payload JSON 解析失败	`SkipRetry`	直接归档
业务参数校验失败（用户不存在）	`SkipRetry`	直接归档
网络超时	普通 `error`	自动重试
数据库连接失败	普通 `error`	自动重试
下游服务限流（429）	普通 `error`	自动重试
任务已被业务逻辑取消	`RevokeTask`	直接丢弃
处理成功	`nil`	标记完成

3 自定义重试延迟

通过 RetryDelayFunc 自定义每次重试的等待时间：

asynq.Config{
    RetryDelayFunc: func(n int, e error, t *asynq.Task) time.Duration {
        // n 从 0 开始（第一次重试 n=0）
        // e 是 Handler 返回的错误
        // t 是当前任务
        return customDelay(n, e, t)
    },
}

3.1 按任务类型定制

RetryDelayFunc: func(n int, e error, t *asynq.Task) time.Duration {
    switch t.Type() {
    case "email:send":
        // 邮件任务：固定 10 秒间隔
        return 10 * time.Second
    case "order:cancel":
        // 订单任务：快速重试
        return time.Duration(n+1) * 2 * time.Second
    default:
        return asynq.DefaultRetryDelayFunc(n, e, t)
    }
},

3.2 按错误类型定制

RetryDelayFunc: func(n int, e error, t *asynq.Task) time.Duration {
    if errors.Is(e, ErrRateLimited) {
        // 被限流 → 等待更久
        return time.Duration(n+1) * 30 * time.Second
    }
    if errors.Is(e, ErrTimeout) {
        // 超时 → 立即重试
        return 1 * time.Second
    }
    return asynq.DefaultRetryDelayFunc(n, e, t)
},

3.3 自定义退避策略示例

// 线性退避：1s, 2s, 3s, 4s, ...
func linearBackoff(n int, e error, t *asynq.Task) time.Duration {
    return time.Duration(n+1) * time.Second
}

// 固定延迟：每次都是 30 秒
func fixedDelay(n int, e error, t *asynq.Task) time.Duration {
    return 30 * time.Second
}

// 分段策略：前 3 次快速，之后拉长
func stagedBackoff(n int, e error, t *asynq.Task) time.Duration {
    if n < 3 {
        return time.Duration(n+1) * 5 * time.Second  // 5s, 10s, 15s
    }
    return time.Duration(n-2) * 60 * time.Second      // 1min, 2min, 3min, ...
}

// 带随机抖动的指数退避（避免惊群效应）
func jitteredExponentialBackoff(n int, e error, t *asynq.Task) time.Duration {
    base := time.Duration(math.Pow(2, float64(n))) * time.Second
    jitter := time.Duration(rand.Int63n(int64(base) / 2))
    return base + jitter
}

4 IsFailure —— 控制重试计数

默认情况下，所有非 nil 错误都会消耗重试次数。通过 IsFailure 可以排除某些错误：

asynq.Config{
    IsFailure: func(err error) bool {
        // ErrResourceNotAvailable 不消耗重试次数
        // 任务会重试但不会计入 MaxRetry 限制
        if errors.Is(err, ErrResourceNotAvailable) {
            return false
        }
        return true
    },
}

💡 使用场景：

下游服务临时过载，期望无限重试直到恢复
等待外部资源就绪（如等待另一个异步操作完成）

🚨 陷阱：滥用 IsFailure: false 会导致任务无限重试，需要确保错误状态最终能自我恢复，或通过外部手段（如 Inspector）手动终止。

5 ErrorHandler —— 最终失败回调

当任务耗尽所有重试次数后，ErrorHandler 被调用：

asynq.Config{
    ErrorHandler: asynq.ErrorHandlerFunc(func(ctx context.Context, task *asynq.Task, err error) {
        taskID, _ := asynq.GetTaskID(ctx)
        queueName, _ := asynq.GetQueueName(ctx)

        // 1. 记录死信
        log.Printf("[DEAD LETTER] id=%s type=%s queue=%s err=%v payload=%s",
            taskID, task.Type(), queueName, err, task.Payload())

        // 2. 写入数据库死信表（便于后续排查和手动重试）
        db.SaveDeadLetter(ctx, DeadLetter{
            TaskID:    taskID,
            TaskType:  task.Type(),
            Payload:   task.Payload(),
            Error:     err.Error(),
            CreatedAt: time.Now(),
        })

        // 3. 发送告警
        alerting.Send(ctx, fmt.Sprintf("Task %s exhausted retries", task.Type()))
    }),
}

💡 ErrorHandler 最佳实践：

记录死信到持久化存储（数据库死信表）
发送告警通知（Slack、PagerDuty、飞书等）
保留原始 Payload 和完整错误堆栈
提供运维手动重试的入口

6 超时与取消

6.1 任务级超时

// 单次执行超过 30 秒则 context 取消
client.Enqueue(task, asynq.Timeout(30*time.Second))

Handler 中感知取消：

func Handler(ctx context.Context, t *asynq.Task) error {
    for i, item := range items {
        select {
        case <-ctx.Done():
            return ctx.Err() // 超时或 Server 关闭
        default:
            process(item)
        }
    }
    return nil
}

6.2 期限（Deadline）

// 超过 deadline 后，即使还有重试次数也不再调度
client.Enqueue(task, asynq.Deadline(time.Now().Add(2*time.Hour)))

6.3 Timeout vs Deadline

	Timeout	Deadline
作用范围	单次执行	整个任务生命周期
超时后	进入 retry	直接 archived
重试时	重新计时	不重新计时
典型场景	防止单次处理卡死	超过指定时间后任务无意义

7 IsPanicError

v0.25.0+

asynq.Config{
    ErrorHandler: asynq.ErrorHandlerFunc(func(ctx context.Context, task *asynq.Task, err error) {
        if errors.Is(err, asynq.IsPanicError) {
            // Handler 发生了 panic
            log.Printf("PANIC in task %s: %v", task.Type(), err)
        }
    }),
}

8 错误处理完整决策树

Handler 返回 error
    │
    ├── err == nil
    │   └── 标记为 completed ✓
    │
    ├── errors.Is(err, SkipRetry)
    │   └── 直接 archived（不重试）
    │
    ├── errors.Is(err, RevokeTask)
    │   └── 丢弃任务（不归档、不重试）
    │
    └── 其他 error
        ├── IsFailure(err) == false
        │   └── 重试但不消耗重试次数（无限重试直到错误恢复）
        │
        └── IsFailure(err) == true
            ├── 重试次数 < MaxRetry
            │   └── 等待 RetryDelayFunc 后进入 retry
            │
            └── 重试次数 >= MaxRetry
                └── archived + ErrorHandler 回调

9 完整配置示例

srv := asynq.NewServer(redisOpt, asynq.Config{
    // 最大并发
    Concurrency: 20,

    // 自定义重试延迟
    RetryDelayFunc: func(n int, e error, t *asynq.Task) time.Duration {
        // 按任务类型和错误类型混合策略
        if errors.Is(e, ErrRateLimited) {
            return time.Duration(n+1) * 30 * time.Second
        }
        switch t.Type() {
        case "order:cancel":
            return time.Duration(n+1) * 2 * time.Second
        default:
            return asynq.DefaultRetryDelayFunc(n, e, t)
        }
    },

    // 部分错误不消耗重试配额
    IsFailure: func(err error) bool {
        if errors.Is(err, ErrResourceNotAvailable) {
            return false
        }
        return true
    },

    // 最终失败处理
    ErrorHandler: asynq.ErrorHandlerFunc(func(ctx context.Context, t *asynq.Task, err error) {
        taskID, _ := asynq.GetTaskID(ctx)

        if errors.Is(err, asynq.IsPanicError) {
            log.Printf("[PANIC] task=%s id=%s", t.Type(), taskID)
        } else {
            log.Printf("[DEAD] task=%s id=%s err=%v", t.Type(), taskID, err)
        }

        // 入库
        saveDeadLetter(ctx, t, err)
        // 告警
        sendAlert(ctx, t, err)
    }),
})

任务处理与中间件周期性任务