mirror of
https://gitee.com/ulthon/ulthon_admin.git
synced 2026-11-10 13:00:49 +08:00
463 lines
20 KiB
Markdown
463 lines
20 KiB
Markdown
---
|
||
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 条
|