高级特性
1 任务聚合（Task Aggregation）
1.1 概述
任务聚合将一段时间内产生的同组任务聚合成一个批次，统一交给 Handler 处理。适合以下场景：
- 日志批量写入
- 指标批量上报
- 消息批量推送
- 数据库批量插入
1.2 入队端
// 指定 Group 名称
client.Enqueue(task, asynq.Group("exports"))
client.Enqueue(task2, asynq.Group("exports"))
client.Enqueue(task3, asynq.Group("exports"))
// 三个任务将聚合为一个批次1.3 Server 端配置
srv := asynq.NewServer(redisOpt, asynq.Config{
GroupGracePeriod: 10 * time.Second, // 首批任务到后至少等 10 秒
GroupMaxDelay: 5 * time.Minute, // 单个任务最多等 5 分钟
GroupMaxSize: 100, // 积累 100 个任务后立即触发
})触发条件（满足任一即触发）：
条件 1：距组内首个任务达到 GroupMaxDelay（"等太久了，立即处理"）
条件 2：组内任务数量达到 GroupMaxSize（"够了，立即处理"）
条件 3：距组内上一个任务超过 GroupGracePeriod（"新任务不来了，处理吧"）时间线示例（GroupGracePeriod=10s, GroupMaxDelay=60s, GroupMaxSize=5）：
t=0s task-a 入组 ─── 开始计时
t=3s task-b 入组
t=7s task-c 入组
t=12s task-d 入组 ← 距 task-c 仅 5 秒，未超过 GracePeriod
t=13s 超时 ──────── ← 距 task-d 已 10 秒无新任务 → 触发聚合！
聚合结果：[task-a, task-b, task-c, task-d]1.4 聚合 Handler
// 聚合后的回调签名
type GroupAggregator interface {
Aggregate(group string, tasks []*Task) *Task
}默认聚合行为是将组内所有任务合并为一个任务，Payload 为各组员 Payload 的 JSON 数组。
1.5 完整示例：批量写入
// 1. 入队
func EnqueueLogEvent(client *asynq.Client, event LogEvent) error {
payload, _ := json.Marshal(event)
task := asynq.NewTask("log:batch_write", payload,
asynq.Group("logs"), // 归入 "logs" 组
asynq.MaxRetry(3),
)
_, err := client.Enqueue(task)
return err
}
// 2. Server 配置
srv := asynq.NewServer(redisOpt, asynq.Config{
GroupGracePeriod: 5 * time.Second,
GroupMaxDelay: 30 * time.Second,
GroupMaxSize: 200,
})
// 3. Handler
func HandleBatchLogWrite(ctx context.Context, t *asynq.Task) error {
var events []LogEvent
if err := json.Unmarshal(t.Payload(), &events); err != nil {
return fmt.Errorf("unmarshal: %w", asynq.SkipRetry)
}
// 批量写入数据库
return db.BatchInsertLogs(ctx, events)
}🚨 陷阱：聚合后的任务 Payload 结构与单个任务不同（变为 JSON 数组）。Handler 需要同时处理单任务和聚合任务的两种情况，或确保聚合后的 Handler 只接收聚合任务。
2 ResultWriter —— 任务结果回写
2.1 概述
ResultWriter 允许 Handler 将执行结果写回 Redis，供调用方异步查询。
注意：
ResultWriter仅在 Handler 中有效。通过NewTask创建的任务调用ResultWriter()返回nil。
2.2 基本用法
func HandleExportTask(ctx context.Context, t *asynq.Task) error {
var p ExportPayload
json.Unmarshal(t.Payload(), &p)
// 执行导出...
fileURL, err := exportService.Export(ctx, p)
if err != nil {
return fmt.Errorf("export failed: %w", err)
}
// 将结果写回任务
result, _ := json.Marshal(map[string]string{"file_url": fileURL})
n, err := t.ResultWriter().Write(result)
if err != nil {
return fmt.Errorf("write result: %w", err)
}
log.Printf("wrote %d bytes result", n)
return nil
}2.3 读取结果
// 通过 Inspector 读取
inspector := asynq.NewInspector(redisOpt)
taskInfo, err := inspector.GetTaskInfo("default", taskID)
if err != nil {
log.Fatal(err)
}
fmt.Printf("Result: %s\n", taskInfo.Result)2.4 结果持久化
任务结果存储在 Redis 中。必须配合 Retention 使用，否则任务完成后立即清理：
// 入队时设置 Retention，确保任务完成后结果保留足够长时间
client.Enqueue(task,
asynq.Retention(24*time.Hour), // 结果保留 24 小时
)2.5 ResultWriter 方法
| 方法 | 签名 | 说明 |
|---|---|---|
Write |
func (w *ResultWriter) Write(data []byte) (int, error) |
写入结果数据 |
TaskID |
func (w *ResultWriter) TaskID() string |
获取关联的任务 ID |
3 任务保留（Retention）
3.1 概述
默认情况下，任务完成后立即从 Redis 中清理。设置 Retention 可延长保留时间。
client.Enqueue(task, asynq.Retention(24*time.Hour))3.2 保留期间的行为
- 完成后进入
completed状态，数据保留在 Redis 的 completed ZSET 中 - 可通过 Inspector 查询任务状态和结果
- 可通过 Web UI / CLI 查看历史任务
- Janitor 会在 Retention 到期后清理过期数据
3.3 适用场景
| 场景 | 建议 Retention |
|---|---|
| 任务审计追踪 | 7~30 天 |
| 结果异步查询 | 1~24 小时 |
| 故障排查 | 3~7 天 |
| 高频任务（日志类） | 不保留或 1 小时内 |
🚨 陷阱：Retention 时间越长，Redis 内存占用越大。高频任务场景下，务必结合 Redis 内存容量合理设置。
4 唯一任务深入
4.1 去重原理
Client.Enqueue(task, Unique(TTL))
│
▼
计算去重 key = Hash(taskType + payload)
│
▼
Redis SETNX(key, TTL)
│
├── 成功：任务入队
│
└── 失败：已有相同任务 → 返回 ErrTaskIdConflict4.2 自定义唯一性范围
// 自定义 TaskID 实现业务去重
// 同一订单的取消操作 5 分钟内只允许一次
taskID := fmt.Sprintf("order:%d:cancel", orderID)
client.Enqueue(task,
asynq.TaskID(taskID),
asynq.Unique(5*time.Minute),
)4.3 Unique vs TaskID
| Unique | TaskID + Unique | |
|---|---|---|
| 去重 key | hash(taskType + payload) |
自定义 TaskID |
| 适用场景 | 防止完全相同的任务重复 | 业务语义去重 |
| 灵活性 | 低（依赖 Payload 内容） | 高（业务决定） |
5 Headers 深入
v0.26.0+
5.1 使用场景汇总
task := asynq.NewTask("my:type", payload,
asynq.Headers(map[string]string{
"X-Trace-ID": traceID, // 分布式追踪
"X-Tenant-ID": tenantID, // 多租户隔离
"X-Request-ID": requestID, // 请求溯源
"X-Idempotency-Key": idempotencyKey, // 幂等键
"X-Retry-Reason": "rate_limited", // 重试原因标记
"X-Source": "api-server", // 来源标识
}),
)5.2 中间件中传递
func HeaderPropagationMiddleware(next asynq.Handler) asynq.Handler {
return asynq.HandlerFunc(func(ctx context.Context, t *asynq.Task) error {
// 从 Headers 提取信息注入 context
if traceID := t.Headers["X-Trace-ID"]; traceID != "" {
ctx = context.WithValue(ctx, traceKey, traceID)
}
if tenantID := t.Headers["X-Tenant-ID"]; tenantID != "" {
ctx = context.WithValue(ctx, tenantKey, tenantID)
}
return next.ProcessTask(ctx, t)
})
}6 JSON vs Gob 序列化
| 维度 | JSON | Gob |
|---|---|---|
| 跨语言 | ✅ 是 | ❌ 仅 Go |
| 人类可读 | ✅ 是 | ❌ 否 |
| 版本兼容 | ✅ 优秀 | ❌ 不同 Go 版本可能不兼容 |
| 性能 | 中 | 高（Go 内使用） |
| 体积 | 中 | 小（二进制） |
| 推荐度 | ⭐⭐⭐⭐⭐ | ⭐ |
💡 强烈推荐使用 JSON。虽然 Gob 在 Go 内性能更优，但其版本兼容性问题在生产环境中是致命的风险——升级 Go 版本可能导致无法反序列化历史任务。
7 自定义 TaskID 策略
// 策略 1：UUID（默认）
task := asynq.NewTask("my:type", payload)
// 策略 2：业务 ID（推荐用于去重）
taskID := fmt.Sprintf("%s:%d:%s", taskType, entityID, action)
client.Enqueue(task, asynq.TaskID(taskID))
// 策略 3：时间戳 + 随机数（日志型任务）
taskID := fmt.Sprintf("%s:%d:%s", taskType, time.Now().UnixNano(), randString(8))
// 策略 4：内容哈希（幂等任务）
hash := sha256.Sum256(payload)
taskID := fmt.Sprintf("%s:%x", taskType, hash[:8])💡 生产建议：对于需要去重的业务任务，使用策略 2（业务 ID）；对于不需要去重的通知类任务，使用默认的 UUID。