核心概念与架构

Asynq

核心概念与架构

1 核心概念速览

概念	说明	类比
Task	最小执行单元，由 Type（字符串标识）和 Payload（字节负载）组成	HTTP Request Body
Client	任务生产者，将 Task 写入 Redis 队列	HTTP Client
Server	任务消费者，从 Redis 拉取 Task 并调度 Worker 执行	HTTP Server
Handler	处理具体任务的业务逻辑，实现 `asynq.Handler` 接口	HTTP Handler
ServeMux	路由分发器，按 Task Type 将任务分派到对应 Handler	`http.ServeMux`
Queue	Redis 中的逻辑队列，可配置多个队列并分配不同权重	HTTP 路由分组
Middleware	任务执行前后的拦截器链，用于日志、追踪、指标采集	HTTP Middleware

2 整体架构

┌──────────────────────────────────────────────────────┐
│                      Client (Producer)               │
│  client.Enqueue(task) ──► Redis ──► Pending Queue    │
└──────────────────────────────────────────────────────┘
                               │
                               ▼
┌──────────────────────────────────────────────────────┐
│                      Redis                           │
│  ┌──────────┐  ┌──────────┐  ┌──────────┐          │
│  │ critical │  │ default  │  │   low    │  ...     │
│  └──────────┘  └──────────┘  └──────────┘          │
│  ┌──────────┐  ┌──────────┐  ┌──────────┐          │
│  │scheduled │  │  retry   │  │ archived │          │
│  └──────────┘  └──────────┘  └──────────┘          │
└──────────────────────────────────────────────────────┘
                               │
                               ▼
┌──────────────────────────────────────────────────────┐
│                   Server (Consumer)                  │
│                                                      │
│  ┌──────────────┐  ┌──────────────┐                  │
│  │   Processor  │  │  Forwarder   │                  │
│  │  (executes   │  │  (scheduled  │                  │
│  │   handlers)  │  │  → pending)  │                  │
│  └──────────────┘  └──────────────┘                  │
│  ┌──────────────┐  ┌──────────────┐                  │
│  │ Heartbeater  │  │  Recoverer   │                  │
│  │  (leases)    │  │  (orphaned)  │                  │
│  └──────────────┘  └──────────────┘                  │
│  ┌──────────────┐  ┌──────────────┐                  │
│  │HealthChecker │  │   Janitor    │                  │
│  │  (liveness)  │  │  (cleanup)   │                  │
│  └──────────────┘  └──────────────┘                  │
│  ┌──────────────┐                                    │
│  │  Aggregator  │                                    │
│  │  (grouping)  │                                    │
│  └──────────────┘                                    │
└──────────────────────────────────────────────────────┘

核心角色

Client：生产者。调用 Enqueue 将 Task 写入 Redis，支持即时、延迟、定时三种入队方式。
Server：消费者。从 Redis 拉取任务并分发给 Worker goroutine 并行处理。
Redis：唯一的存储和协调中间件。所有任务数据、状态信息、Worker 租约均在 Redis 中维护。

3 Server 内部组件详解

Asynq Server 启动后，内部会运行多个后台 goroutine 协同工作：

3.1 Forwarder（转发器）

将 scheduled 和 retry 队列中到期的任务移动到 pending 队列。

定期扫描 scheduled set 和 retry set（ZSET，score 为处理时间）
将 score <= now() 的任务出队，写入 pending list
配置项：DelayedTaskCheckInterval

3.2 Processor（处理器）

从 pending 队列拉取任务，交给 Handler 执行业务逻辑。

使用 Semaphore（带缓冲 channel）控制并发 Worker 数
从各优先级队列按权重概率拉取任务
执行前将任务写入 active list（表示正在处理中）
执行后根据返回值决定标记完成或进入 retry/archived

3.3 Heartbeater（心跳维护）

定期续约正在处理的任务的租约（lease），防止被 Recoverer 误判为孤儿任务。

每个活跃任务在 Redis 中有一个 lease 记录（ZSET，score 为过期时间）
Heartbeater 定期将 lease 的过期时间向后推延
Worker 崩溃后，lease 过期，Recoverer 将其回收

3.4 Recoverer（孤儿恢复）

检测因 Worker 崩溃而产生的孤儿任务，重新入队或归档。

扫描 active list 中所有任务
对比其 lease：已过期的即为孤儿任务
根据剩余重试次数决定重新入队（retry）或归档（archived）

🔬 深入原理：Asynq 通过 Redis 的 ZSET + Lua 脚本实现了分布式的 Worker 存活检测。每个 Worker 会定期将自己的活跃任务 lease 续约。如果一个 Worker 崩溃（进程退出、网络断开），其对应任务的 lease 将在配置的超时后过期，被同一 Redis 下的其他 Server 实例的 Recoverer 检测并回收。

3.5 HealthChecker（健康检查）

定期检查 Redis 连接健康状态，允许用户注册自定义健康检查回调。

配置项：HealthCheckInterval、HealthCheckFunc

3.6 Janitor（清理器）

定期清理过期数据，包括已完成但超过 Retention 时间的任务。

扫描 completed set，删除 score 过期的条目
配置项：JanitorInterval、JanitorBatchSize

3.7 Aggregator（聚合器）

将同一 Group 的任务聚合为批次，统一回调给 Handler 处理。

配置项：GroupGracePeriod、GroupMaxDelay、GroupMaxSize、GroupAggregator

4 任务生命周期

一个任务从创建到最终完成，经历以下状态流转：

                    ┌──────────┐
                    │  Client   │
                    │ .Enqueue()│
                    └─────┬────┘
                          │
              ┌───────────┼───────────┐
              │           │           │
              ▼           ▼           ▼
        ┌──────────┐ ┌──────────┐ ┌──────────┐
        │ pending  │ │scheduled │ │aggregating│
        │ (立即执行) │ │(延迟/定时)│ │ (聚合等待) │
        └────┬─────┘ └────┬─────┘ └────┬─────┘
             │            │            │
             │     (到期后由Forwarder移动)   │
             │            │            │
             └────────────┼────────────┘
                          │
                          ▼
                    ┌──────────┐
                    │  active  │ ◄── Worker 正在处理
                    └────┬─────┘
                         │
              ┌──────────┼──────────┐
              │          │          │
              ▼          ▼          ▼
        ┌──────────┐ ┌──────────┐ ┌──────────┐
        │completed │ │  retry   │ │ archived │
        │   (成功)  │ │ (失败重试)│ │(重试耗尽) │
        └──────────┘ └────┬─────┘ └──────────┘
                          │
                    (到期后重新进入 active)
                          │
                          ▼
              ┌─────────────────────┐
              │ retry 耗尽 ──► archived │
              └─────────────────────┘

七种任务状态

状态	枚举值	字符串	含义
`TaskStateActive`	1	`"active"`	正在被 Worker 处理
`TaskStatePending`	2	`"pending"`	已就绪，等待 Worker 拉取
`TaskStateScheduled`	3	`"scheduled"`	已排期，等待到达执行时间
`TaskStateRetry`	4	`"retry"`	执行失败，等待重试
`TaskStateArchived`	5	`"archived"`	重试耗尽或 SkipRetry，已归档
`TaskStateCompleted`	6	`"completed"`	执行成功（在 Retention 时间内保留）
`TaskStateAggregating`	7	`"aggregating"`	等待同组任务聚合

TaskInfo 结构

Client.Enqueue() 返回 *asynq.TaskInfo，包含以下字段：

字段	类型	说明
`ID`	`string`	任务唯一标识（UUID）
`Queue`	`string`	所属队列名称
`Type`	`string`	任务类型标识
`Payload`	`[]byte`	负载数据
`State`	`TaskState`	当前状态
`MaxRetry`	`int`	最大重试次数
`Retried`	`int`	已重试次数
`LastErr`	`string`	最后一次错误信息
`LastFailedAt`	`time.Time`	最后一次失败时间
`Timeout`	`time.Duration`	单次执行超时
`Deadline`	`time.Time`	绝对截止时间
`Group`	`string`	聚合组名
`NextProcessAt`	`time.Time`	下次处理时间
`IsOrphaned`	`bool`	是否为孤儿任务
`Retention`	`time.Duration`	完成后保留时长
`CompletedAt`	`time.Time`	完成时间
`Result`	`[]byte`	任务结果数据（由 ResultWriter 写入）

5 队列模型

Asynq 支持多队列，每个队列本质上对应 Redis 中的一个数据结构集合：

Queue "critical":
  ├── pending list   (LPUSH/RPOP)     ← 就绪任务
  ├── active list    (List)            ← 正在处理
  ├── scheduled zset (ZSET)            ← 延迟任务
  ├── retry zset     (ZSET)            ← 失败重试
  ├── archived zset  (ZSET)            ← 已归档
  ├── completed zset  (ZSET)           ← 已完成
  ├── lease zset     (ZSET)            ← Worker 租约
  └── aggregating set (SET)            ← 聚合等待

队列权重与优先级

Queues: map[string]int{
    "critical": 6,  // 权重 6 → 约 60% 的处理概率
    "default":  3,  // 权重 3 → 约 30% 的处理概率
    "low":      1,  // 权重 1 → 约 10% 的处理概率
}

默认模式（StrictPriority: false）：按权重概率随机选择队列消费，高权重队列获得更多处理机会。 严格模式（StrictPriority: true）：始终先消费高权重队列，直到清空后才处理低权重队列。

6 Asynq 为什么快

🔬 深入原理：

Redis 原生数据结构：pending 队列使用 Redis List（O(1) 的 LPUSH/RPOP），延迟/重试/归档使用 ZSET（O(log N) 的范围查询）。
Lua 脚本原子操作：关键操作（入队、出队、状态变更）均通过 Lua 脚本在 Redis 端原子执行，避免多次往返和竞态条件。
单次网络往返：每个任务的入队操作只需一次 Redis 命令（Lua 脚本内含多步逻辑）。
Go 原生并发：利用 goroutine + channel 实现高效 Worker 池，轻量且无锁竞争。
无中心化协调：所有协调通过 Redis 完成，Server 实例之间无直接通信，天然支持水平扩展。

概述与快速开始任务创建与入队