docs(skill): 新增队列设计技能（DB 原子抢占/区间游标/Redis Stream 生产者消费者三模式）

2026-10-29 08:58:22 +08:00 · 2026-10-02 01:15:30 +00:00
parent 0ea627583b
commit 504d9a38b9
1 changed files with 462 additions and 0 deletions
--- a/.agents/skills/ulthon-queue/SKILL.md
+++ b/.agents/skills/ulthon-queue/SKILL.md
@@ -0,0 +1,462 @@
+---
+name: "ulthon-queue"
+description: "基于内置定时器设计任务队列与数据消费的规范（DB 原子抢占、区间游标、Redis Stream 生产者消费者三种模式）。需要异步处理数据表（消费队列、批量并行处理、worker 竞争）、设计队列表或生产者消费者、解决任务重复消费、并发竞争、超时回收，或排查 id 取余分片索引失效时调用。"
+---
+
+# queue（基于定时器的任务队列设计）
+
+前置技能：[ulthon-timer](../ulthon-timer/SKILL.md)。本技能建立在 timer 的 `site` 任务 + 时间窗口循环模式之上，先读懂定时器机制再读本技能。
+
+本技能回答的问题：**有一张表的数据需要异步/并行处理（消费），怎么设计才正确**。
+
+## 核心原则（先记住 5 条）
+
+1) **并发数只控制"消费者数量"，绝不划分数据**。`concurrency` / `$this->concurrencyId` 是消费者编号，不是数据分片号。禁止按 `concurrencyId` 预分配数据（如 `id % N` 取余分片，见反模式）
+2) **数据竞争用原子条件 UPDATE 消解（乐观锁/CAS）**，不用长事务悲观锁。抢占的锁生命周期 = 一条 UPDATE 的执行时长（微秒级），处理数据时不持任何锁
+3) **候选查询必须走索引区间扫描**：`WHERE status = 0 ORDER BY id LIMIT N` 配 `(status, id)` 复合索引，只读前 N 条，与表总量无关。索引是本技能一切模式的性能命根子
+4) **at-least-once 语义**：超时重投、重复消费不可避免，业务回写必须幂等（条件 UPDATE 兜底）
+5) **无序是并发的前提**：任务间有顺序依赖就不能并发消费，退回单消费者（`concurrency = 1`）
+
+## 选型决策表
+
+三个问题快速定位：
+
+| 问题 | 答案 | 用哪个模式 |
+|------|------|-----------|
+| 数据是持续产生还是存量一次性？ | 持续产生（源源不断的新数据要异步处理） | 模式 A 或 C |
+| | 存量一次性跑批 | 模式 B |
+| 消费速率波动大 / 吞吐要求高 / 需要削峰？ | 是 | 模式 C |
+| 能否接受引入（共享）Redis 依赖？ | 不能或没必要 | 模式 A |
+
+| 场景 | 推荐 | 理由 |
+|------|------|------|
+| 万级到几十万级、单表、持续消费（默认选择） | 模式 A | 零新增组件，复用 timer 全部能力 |
+| 存量数据跑一遍（数据修复、全量刷新） | 模式 B | 游标走主键索引，跑完即止 |
+| 高吞吐、削峰、多类任务复用队列设施 | 模式 C | 队列调度天然负载均衡，消费者弹性伸缩 |
+| 任务必须严格按序处理 | 都不适用 | 单消费者串行，放弃并发 |
+
+模式 A/B 只依赖 MySQL（5.7 兼容）；模式 C 依赖 Redis（Stream 需 5.0+，XAUTOCLAIM 需 6.2+）。
+
+## 模式 A：DB 原子抢占（默认推荐）
+
+### 锁思路（必须理解的原理）
+
+**错误写法（悲观锁长事务）**：
+
+```php
+// 反面教材：锁住 -> 处理 -> 才提交
+Db::transaction(function () {
+    $items = Db::name('task')->where('status', 0)->lock(true)->limit(20)->select();
+    // 在事务内处理业务（耗时数秒）...
+    // 期间其他消费者的 FOR UPDATE 全部阻塞等待，并发退化为串行
+});
+```
+
+**正确思路（CAS 乐观锁，三段生命周期）**：
+
+```
+抢占（一条原子 UPDATE，持锁微秒级）
+  -> 处理业务（不持任何锁，事务已提交）
+  -> 回写结果（又一条短的条件 UPDATE）
+```
+
+```php
+// 抢占：检查和修改压进同一条原子语句
+Db::name('task')
+    ->where('id', 'in', $candidates)
+    ->where('status', 0)              // 期望值条件：被别的消费者抢先的行不会被更新
+    ->update(['status' => 1, 'claim_token' => $token, 'claim_time' => time()]);
+```
+
+- InnoDB 的 UPDATE 原子执行，WHERE 里的 `status = 0` 让输掉的消费者自动少抢几行，无需等锁、无需重试锁
+- `affected_rows` 返回实际抢到的数量
+- 框架内已有同模式先例：timer 手动触发消费（`TimerBase::shouldExecuteTask` 的原子条件 UPDATE 单赢家）
+
+### 标准字段集
+
+对要消费的业务表加字段（表结构统一用 Scheme 定义，见 [ulthon-scheme-definition](../ulthon-scheme-definition/SKILL.md)；status 用 tinyint，禁止 ENUM）：
+
+| 字段 | 类型 | 必要性 | 说明 |
+|------|------|--------|------|
+| `status` | tinyint | 必须 | 0待处理 1处理中 2成功 3失败。若业务已有状态字段能表达"待处理"可复用，但独立字段不与业务状态机耦合，更稳 |
+| `claim_time` | int | 必须 | 抢占时间戳，超时回收的唯一依据 |
+| `claim_token` | varchar(32) | 批量抢占需要 | 本轮抢占令牌，区分"这轮我抢到哪几条"。每次只抢 1 条时可省 |
+| `worker_id` | varchar(64) | 可选 | 消费者标识，排查用 |
+| `error_message` | varchar/text | 可选 | 失败原因，排查用 |
+
+```sql
+-- 性能命根子：没有它，深积压时候选查询全表扫描
+ALTER TABLE task ADD INDEX idx_status_id (status, id);
+```
+
+说明：InnoDB 二级索引叶子节点自动附带主键，`(status)` 物理上已等效 `(status, id)`，显式声明是为了语义明确。有了它，`WHERE status = 0 ORDER BY id LIMIT 20` 从索引区间起点顺序读 20 条即可，不回表、不 filesort，深积压（百万条 status=0）时也只读 20 个索引项。
+
+### 消费者实现（完整模板）
+
+```php
+<?php
+
+namespace app\tools\controller\timer;
+
+use app\common\controller\TimerController;
+use app\common\service\HostService;
+use think\facade\Db;
+
+class NotifyConsumer extends TimerController
+{
+    protected $concurrency = 4;    // 消费者数（= 并发上限 cap，后台可在 cap 内动态调整）
+    protected $frequency = 5;
+    protected $maxRunTime = 3;     // 时间窗口，不超过 frequency 的一半
+
+    public function do()
+    {
+        $max_time = time() + $this->maxRunTime;
+
+        do {
+            $has_work = $this->doBatch();
+
+            if (!$has_work) {
+                sleep(1);          // 队列空时休眠，防空转；有数据时不要 sleep
+                break;
+            }
+
+            if (time() >= $max_time) {
+                break;
+            }
+        } while (true);
+
+        return 'done';
+    }
+
+    protected function doBatch(): bool
+    {
+        // 候选：走 (status, id) 索引，只读前 20 条
+        $candidates = Db::name('notify_task')
+            ->where('status', 0)
+            ->order('id')
+            ->limit(20)
+            ->column('id');
+
+        if (empty($candidates)) {
+            return false;
+        }
+
+        // 原子抢占：WHERE 带 status=0，被抢先的行 affected=0，竞态在此消解
+        $token = uniqid('', true);
+        Db::name('notify_task')
+            ->where('id', 'in', $candidates)
+            ->where('status', 0)
+            ->update([
+                'status'      => 1,
+                'claim_token' => $token,
+                'worker_id'   => $this->workerName(),
+                'claim_time'  => time(),
+            ]);
+
+        // 查回真正抢到的：必须带 id IN（主键定位），只按 claim_token 查会全表扫描
+        $mine = Db::name('notify_task')
+            ->where('id', 'in', $candidates)
+            ->where('claim_token', $token)
+            ->select();
+
+        foreach ($mine as $item) {
+            try {
+                // ... 单条业务处理 ...
+
+                // 幂等回写完成态：WHERE status=1 兜底重复消费
+                Db::name('notify_task')->where('id', $item['id'])
+                    ->where('status', 1)->update(['status' => 2]);
+            } catch (\Throwable $e) {
+                Db::name('notify_task')->where('id', $item['id'])
+                    ->where('status', 1)
+                    ->update(['status' => 3, 'error_message' => mb_substr($e->getMessage(), 0, 500)]);
+            }
+        }
+
+        return true;
+    }
+
+    protected function workerName(): string
+    {
+        return HostService::getNodeId() . '-c' . $this->concurrencyId;
+    }
+}
+```
+
+注册（`app/common/command/timer/config.php`）：
+
+```php
+[
+    'name' => 'notify_consumer',
+    'type' => 'site',
+    'target' => '/tools/timer.NotifyConsumer/do',
+    'frequency' => 5,
+    'concurrency' => 4,
+],
+```
+
+失败重试的扩展：示例中失败为终态（status=3）。需要自动重试时，加 `retry_count` / `next_retry_time` 字段做退避（重置 status=0 但候选查询带 `next_retry_time < now` 条件），防止失败任务被反复抢占形成热点。
+
+### 超时回收任务（必须配套）
+
+消费者崩溃后 status=1 的行会永久卡住。必须配一个独立回收任务：
+
+```php
+class NotifyReclaim extends TimerController
+{
+    protected $concurrency = 1;
+    protected $frequency = 60;
+
+    public function do()
+    {
+        $timeout = 600;   // 必须 > 单条最大处理时长，否则误回收正在处理的行
+
+        Db::name('notify_task')
+            ->where('status', 1)
+            ->where('claim_time', '<', time() - $timeout)
+            ->update(['status' => 0, 'claim_token' => null, 'worker_id' => null, 'claim_time' => null]);
+
+        return 'done';
+    }
+}
+```
+
+回收本身也是原子条件 UPDATE，幂等无害。多节点部署下回收任务 `run_type=auto`（一个节点跑即可）；消费者任务要各节点分摊时切 `run_type=all`。
+
+## 模式 B：区间游标（存量一次性跑批）
+
+适用：把存量数据跑一遍（数据修复、全量刷新），不需要状态字段，不需要竞争。
+
+```php
+class FixDataTask extends TimerController
+{
+    protected $concurrency = 4;      // 并行分段数
+    protected $frequency = 5;
+    protected $maxRunTime = 10;
+
+    public function do()
+    {
+        // 启动时按 min/max 划分连续区间，分片 i 处理 [lo, hi)，各自独立游标
+        [$lo, $hi] = $this->myRange();
+
+        $cursor_key = 'fix_cursor_' . $this->concurrencyId;
+        $cursor = (int) Cache::get($cursor_key, $lo);
+
+        $batch = Db::name('some_table')
+            ->where('id', '>=', $cursor)
+            ->where('id', '<', $hi)
+            ->order('id')             // 走主键 range scan；显式 ORDER BY 同时是执行计划稳定器
+            ->limit(500)
+            ->select();
+
+        if ($batch->isEmpty()) {
+            Cache::delete($cursor_key);   // 本区间完成
+            return 'range done';
+        }
+
+        foreach ($batch as $item) {
+            // ... 业务处理 ...
+        }
+
+        Cache::set($cursor_key, $batch[count($batch) - 1]['id']);
+
+        return 'batch done';
+    }
+
+    protected function myRange(): array
+    {
+        $min = (int) Db::name('some_table')->min('id');
+        $max = (int) Db::name('some_table')->max('id');
+        $step = (int) ceil(($max - $min + 1) / $this->concurrencyCount);
+
+        return [$min + $step * $this->concurrencyId, $min + $step * ($this->concurrencyId + 1)];
+    }
+}
+```
+
+要点：
+
+- `id > cursor ORDER BY id LIMIT N` 是教科书级主键 range scan，每批只读 N 行
+- 区间是**连续划分**不是取余，索引全程有效
+- 游标存在 Cache，可重入：进程重启后从断点继续
+- 区间静态划分仍有数据倾斜（id 空洞、单条耗时不均），一次性任务可接受；持续消费场景不要用本模式，用模式 A（竞争消费天然均衡）
+
+## 模式 C：生产者-消费者（Redis Stream）
+
+适用：高吞吐、削峰、消费速率波动大、多类任务复用同一套队列设施。
+
+```
+DB --[生产者 timer:游标捞取+入队,串行无竞争]--> Redis Stream --[N 个消费者:消费组]--> 处理 --> 回写 DB
+```
+
+价值：取数者只有一个（索引友好、无竞态），并发调度交给队列，消费端 push/pop 天然负载均衡；生产/消费速率解耦，各自的 frequency/concurrency 独立调节。
+
+引入前提：已有（或多节点必须有的）共享 Redis。不要为此单独引入一套新基础设施，也**不要用 MySQL MEMORY 表替代**（丢持久性、仍是 DB 竞争模式、锁并发不如 Redis 原子操作；不想用 Redis 就用模式 A）。
+
+### 生产者（concurrency=1，run_type=auto 防多节点重复投递）
+
+```php
+class NotifyProducer extends TimerController
+{
+    protected $concurrency = 1;    // 生产者永远单实例
+    protected $frequency = 5;
+
+    public function do()
+    {
+        $cursor = (int) Cache::get('notify_producer_cursor', 0);
+        $batch = Db::name('notify_task')
+            ->where('id', '>', $cursor)
+            ->where('status', 0)
+            ->order('id')
+            ->limit(200)
+            ->select();
+
+        if ($batch->isEmpty()) {
+            Cache::set('notify_producer_cursor', 0);   // 一轮跑完，从头巡检新数据
+            return 'done';
+        }
+
+        $redis = Cache::store('redis')->handler();
+        foreach ($batch as $item) {
+            // 顺序铁律：先标 queued 再入队（反过来崩溃窗口会丢数据）
+            Db::name('notify_task')->where('id', $item['id'])->where('status', 0)
+                ->update(['status' => 4]);             // 4=已入队
+            $redis->xAdd('notify_stream', '*', ['id' => $item['id']]);
+        }
+
+        Cache::set('notify_producer_cursor', $batch[count($batch) - 1]['id']);
+
+        return 'done';
+    }
+}
+```
+
+### 消费者（concurrency=N，动态可调）
+
+```php
+class NotifyStreamConsumer extends TimerController
+{
+    protected $concurrency = 4;
+    protected $frequency = 5;
+    protected $maxRunTime = 3;
+
+    public function do()
+    {
+        $redis = Cache::store('redis')->handler();
+        $this->ensureGroup($redis);
+        $consumer = 'c' . $this->concurrencyId;
+        $max_time = time() + $this->maxRunTime;
+
+        do {
+            $msgs = $redis->xReadGroup('notify_group', $consumer, ['notify_stream' => '>'], 10);
+
+            if (empty($msgs)) {
+                // 新消息为空 -> 认领超时未确认的（崩溃恢复）
+                $this->claimPending($redis, $consumer);
+                usleep(200000);
+                continue;
+            }
+
+            foreach ($msgs as $entry_id => $fields) {
+                $this->processOne($fields['id']);       // 内部含幂等回写
+                $redis->xAck('notify_stream', 'notify_group', [$entry_id]);
+            }
+        } while (time() < $max_time);
+
+        return 'done';
+    }
+
+    protected function ensureGroup(\Redis $redis): void
+    {
+        try {
+            $redis->xGroupCreate('notify_stream', 'notify_group', '0', true);
+        } catch (\Throwable $e) {
+            // BUSYGROUP: 组已存在，忽略
+        }
+    }
+
+    protected function claimPending(\Redis $redis, string $consumer): void
+    {
+        // XAUTOCLAIM 需要 Redis 6.2+；旧版本用 XPENDING + XCLAIM 组合
+        $result = $redis->xAutoClaim('notify_stream', 'notify_group', $consumer, 600000, '0-0', 10);
+        foreach ($result[1] ?? [] as $entry_id => $fields) {
+            $this->processOne($fields['id']);
+            $redis->xAck('notify_stream', 'notify_group', [$entry_id]);
+        }
+    }
+}
+```
+
+### 可靠性三原则（模式 C 的生死线）
+
+| 原则 | 原因 |
+|------|------|
+| **用 Stream 不用 List** | `RPOP/BRPOP` 是 at-most-once：弹出后消费者崩溃，消息永久丢失。Stream 消费组把"取出未确认"的消息留在 pending 列表，可被 XAUTOCLAIM 重投 |
+| **先标后投** | "从 DB 捞出"与"写入 Redis"之间崩溃会丢数据。先 `status=queued` 再 xAdd，配合回收（`status=4 AND update_time < now-600` 重置为 0）。反过来会重复投递——重复比丢失好处理（幂等可兜底），丢失是静默的 |
+| **幂等消费** | at-least-once 下消息可能投递两次（认领重试、生产者重复入队），回写用条件 UPDATE（`WHERE status != done`）兜底 |
+
+其他要点：
+
+- 消费组内每个 `$this->concurrencyId` 用独立 consumer 名，pending 认领才能跨消费者转移
+- 多节点部署：所有节点连同一个 Redis（共享 Stream），`run_type=all` 让各节点都起消费者
+- 消息体只放主键 id，业务数据回表取，避免消息体积膨胀与数据陈旧
+
+## 反模式清单（禁止）
+
+### 1. id 取余分片（索引失效）
+
+```php
+// 反面教材：WHERE id % 4 = 2
+SomeModel::whereRaw('id % ' . $this->concurrencyCount . ' = ' . $this->concurrencyId)->select();
+```
+
+`id % N = k` 对主键是表达式谓词，破坏索引有序性，命中行散布全键空间，优化器无法 range scan，只能全索引扫描；N 个分片各扫一遍 = 全表数据被过滤 N 次。MySQL 8.0.13+ 函数索引可解，但项目基线 5.7+ 不可依赖。
+
+正确替代：数据不预分配，用原子抢占（模式 A）或区间划分（模式 B）。
+
+### 2. find 后直接 save（竞态，重复消费）
+
+```php
+// 反面教材：读和标记之间不是原子的
+$item = SomeModel::where('status', 0)->find();
+// ... 处理 ...
+$item->status = 1;
+$item->save();
+```
+
+`concurrency > 1` 时多个消费者同时读到同一行、都处理一遍。读和标记必须压进一条条件 UPDATE（见模式 A）。
+
+### 3. ORDER BY RAND()
+
+全表 + 随机排序，O(N log N) 灾难。任何"随机取一批"的需求都用 `WHERE status = 0 ORDER BY id LIMIT N` + 抢占满足。
+
+### 4. 悲观锁长事务
+
+`SELECT FOR UPDATE` 锁住后慢慢处理再提交 = 并发退化为串行（见模式 A 的锁思路）。锁只允许出现在"抢"和"放"两个微秒级临界区。MySQL 8.0+ 可用 `FOR UPDATE SKIP LOCKED` 一条语句解决竞争，但基线 5.7 不可依赖，统一用条件 UPDATE。
+
+## 语义保证（设计前想清楚）
+
+- **至少一次（at-least-once）**：超时重投（模式 A 回收 / 模式 C 认领）+ 先标后投的崩溃窗口，决定了同一条数据可能被处理多次。这是可靠性的代价，换来不丢数据
+- **幂等是义务**：处理逻辑必须可安全重放。最低要求完成态回写带条件（`WHERE status = 1`）；涉及外部副作用（发短信、扣款）时用业务防重表/唯一键兜底
+- **顺序性**：并发消费不保证处理顺序。任务间有依赖（后一条依赖前一条结果）时只能单消费者；"失败重投无队头阻塞"是无序换来的自由，需要严格 FIFO 的业务不存在并发空间
+- **关于排序的性能事实**：在 `(status, id)` 索引下 `ORDER BY id` 是免费的（索引区间扫描天然有序），删掉它没有可测量的收益，反而丢失执行计划稳定性。性能的决定因素是索引，不是 ORDER BY 子句
+
+## 与 timer 机制的组合
+
+| 组合点 | 用法 |
+|--------|------|
+| `concurrency` | 消费者数量（吞吐旋钮），配合后台动态并发热调；扩容/缩容即时生效 |
+| `$this->concurrencyId` | 消费者编号，用于日志排查、worker 标识、Stream consumer 名；**不用于划分数据** |
+| 时间窗口模式 | `maxRunTime` + 循环消费 + 超时退出，见 [ulthon-timer](../ulthon-timer/SKILL.md)；`frequency >= maxRunTime * 2` |
+| `run_type` | 消费者任务默认 `auto`（单节点跑全部消费者）；要各节点分摊时切 `all`。回收/生产者任务恒为 `auto` 或 `main`（全局单份） |
+| 执行日志 | `/do` 自动重写为 `/execute`，每轮消费的耗时/结果进 `ul_system_timer_log`，配合定时器日志页排查 |
+
+## 常见坑位（快速自检）
+
+- **查回抢占结果只按 `claim_token` 查**：token 无索引会全表扫描。必须 `WHERE id IN (candidates) AND claim_token = token`（主键定位后内存过滤）
+- **回收阈值小于单条最大处理时长**：正在处理的行被误回收、重复消费。阈值保守设（如单条最长 30 秒则阈值 600 秒）
+- **候选查询没建 `(status, id)` 索引**：深积压时候选选择全表扫描，一切模式白搭。上线前 EXPLAIN 确认 ref/range
+- **失败任务直接重置 status=0 无退避**：失败任务立刻被反复抢占，形成热点空转。重试必须带 `retry_count` / `next_retry_time` 退避
+- **模式 C 多节点但 Redis 不共享**：各节点消费各自的 Stream，消息错乱。多节点必须共享同一个 Redis 实例
+- **消费者 `sleep` 写在处理分支里**：有数据时 sleep 会拖慢吞吐。只在队列空时休眠
+- **把 `concurrency` 当数据分片数理解**：又绕回 `id % N` 反模式。重读核心原则第 1 条