From 504d9a38b983877809143395b35ba0546df0daea Mon Sep 17 00:00:00 2001 From: augushong Date: Fri, 2 Oct 2026 01:15:30 +0000 Subject: [PATCH] =?UTF-8?q?docs(skill):=20=E6=96=B0=E5=A2=9E=E9=98=9F?= =?UTF-8?q?=E5=88=97=E8=AE=BE=E8=AE=A1=E6=8A=80=E8=83=BD=EF=BC=88DB=20?= =?UTF-8?q?=E5=8E=9F=E5=AD=90=E6=8A=A2=E5=8D=A0/=E5=8C=BA=E9=97=B4?= =?UTF-8?q?=E6=B8=B8=E6=A0=87/Redis=20Stream=20=E7=94=9F=E4=BA=A7=E8=80=85?= =?UTF-8?q?=E6=B6=88=E8=B4=B9=E8=80=85=E4=B8=89=E6=A8=A1=E5=BC=8F=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .agents/skills/ulthon-queue/SKILL.md | 462 +++++++++++++++++++++++++++ 1 file changed, 462 insertions(+) create mode 100644 .agents/skills/ulthon-queue/SKILL.md diff --git a/.agents/skills/ulthon-queue/SKILL.md b/.agents/skills/ulthon-queue/SKILL.md new file mode 100644 index 0000000..da533c3 --- /dev/null +++ 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 +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 条