diff --git a/.agents/skills/ulthon-queue/SKILL.md b/.agents/skills/ulthon-queue/SKILL.md index da533c3..2d9b3c9 100644 --- a/.agents/skills/ulthon-queue/SKILL.md +++ b/.agents/skills/ulthon-queue/SKILL.md @@ -78,7 +78,7 @@ Db::name('task') | 字段 | 类型 | 必要性 | 说明 | |------|------|--------|------| -| `status` | tinyint | 必须 | 0待处理 1处理中 2成功 3失败。若业务已有状态字段能表达"待处理"可复用,但独立字段不与业务状态机耦合,更稳 | +| `status` | tinyint | 必须 | 0待处理 1处理中 2成功 3失败 4已入队(4 仅模式 C 使用)。若业务已有状态字段能表达"待处理"可复用,但独立字段不与业务状态机耦合,更稳 | | `claim_time` | int | 必须 | 抢占时间戳,超时回收的唯一依据 | | `claim_token` | varchar(32) | 批量抢占需要 | 本轮抢占令牌,区分"这轮我抢到哪几条"。每次只抢 1 条时可省 | | `worker_id` | varchar(64) | 可选 | 消费者标识,排查用 | @@ -105,8 +105,9 @@ use think\facade\Db; class NotifyConsumer extends TimerController { protected $concurrency = 4; // 消费者数(= 并发上限 cap,后台可在 cap 内动态调整) - protected $frequency = 5; protected $maxRunTime = 3; // 时间窗口,不超过 frequency 的一半 + // 一般不设控制器级 $frequency:防刷会在窗口内拦截手动触发(详见「与 timer 机制的组合」), + // 调度节流已由定时器侧 Cache 节流承担;确需防外部刷 URL 再设,且应小于调度 frequency public function do() { @@ -231,9 +232,18 @@ class NotifyReclaim extends TimerController class FixDataTask extends TimerController { protected $concurrency = 4; // 并行分段数 - protected $frequency = 5; + protected $frequency = 20; // 调度频率,与 maxRunTime 保持 >= 2 倍关系 protected $maxRunTime = 10; + // 基类只注入 $concurrencyId 属性,concurrencyCount 需自行从请求参数取 + protected int $concurrencyCount = 1; + + public function initialize() + { + parent::initialize(); + $this->concurrencyCount = (int) $this->request->param('concurrency_count', 1); + } + public function do() { // 启动时按 min/max 划分连续区间,分片 i 处理 [lo, hi),各自独立游标 @@ -293,6 +303,24 @@ DB --[生产者 timer:游标捞取+入队,串行无竞争]--> Redis Stream --[N 引入前提:已有(或多节点必须有的)共享 Redis。不要为此单独引入一套新基础设施,也**不要用 MySQL MEMORY 表替代**(丢持久性、仍是 DB 竞争模式、锁并发不如 Redis 原子操作;不想用 Redis 就用模式 A)。 +### 环境前提(缺失会直接报错) + +1. **`config/cache.php` 需配置 `redis` store**——框架默认只带 `file` / `login` / `mcp_session`,未配置时 `Cache::store('redis')` 抛 `InvalidArgumentException: Store [redis] not found`: + +```php +// config/cache.php 的 stores 中新增 +'redis' => [ + 'type' => 'Redis', + 'host' => env('redis.host', '127.0.0.1'), + 'port' => env('redis.port', 6379), + 'password' => env('redis.password', ''), + 'select' => (int) env('redis.select', 0), +], +``` + +2. **需要 phpredis 扩展**——`handler()` 返回原生 `\Redis` 的前提(未装 phpredis 且未装 predis 时抛 `BadFunctionCallException: not support: redis`)。本节示例的 `xAdd` / `xReadGroup` / `xAck` / `xAutoClaim` 驼峰方法是 phpredis 扩展的 API,Predis 环境(`handler()` 返回 `\Predis\Client`)方法名不同,示例不适用 +3. **版本要求**:Redis 服务器 Stream 与消费组需 5.0+,`XAUTOCLAIM` 需 6.2+(旧版本用 `XPENDING` + `XCLAIM` 组合替代);phpredis 扩展需 5.3+ 基础 Stream 支持、6.0.0+ 才有 `xAutoClaim` + ### 生产者(concurrency=1,run_type=auto 防多节点重复投递) ```php @@ -337,7 +365,6 @@ class NotifyProducer extends TimerController class NotifyStreamConsumer extends TimerController { protected $concurrency = 4; - protected $frequency = 5; protected $maxRunTime = 3; public function do() @@ -348,16 +375,18 @@ class NotifyStreamConsumer extends TimerController $max_time = time() + $this->maxRunTime; do { + // 返回结构嵌套流名层:[流名 => [entry_id => [字段 => 值]]],遍历前先取 $msgs['notify_stream'] + // 无新消息时旧版 Redis 返回 false(nil 应答)、Redis 8+ 返回空数组,统一 truthy 判断 $msgs = $redis->xReadGroup('notify_group', $consumer, ['notify_stream' => '>'], 10); - if (empty($msgs)) { + if (empty($msgs['notify_stream'])) { // 新消息为空 -> 认领超时未确认的(崩溃恢复) $this->claimPending($redis, $consumer); usleep(200000); continue; } - foreach ($msgs as $entry_id => $fields) { + foreach ($msgs['notify_stream'] as $entry_id => $fields) { $this->processOne($fields['id']); // 内部含幂等回写 $redis->xAck('notify_stream', 'notify_group', [$entry_id]); } @@ -368,16 +397,18 @@ class NotifyStreamConsumer extends TimerController protected function ensureGroup(\Redis $redis): void { - try { - $redis->xGroupCreate('notify_stream', 'notify_group', '0', true); - } catch (\Throwable $e) { - // BUSYGROUP: 组已存在,忽略 + // phpredis 没有 xGroupCreate 方法,统一入口是 xGroup('CREATE', ...) + // 组已存在(BUSYGROUP)不抛异常而是返回 false,要用 getLastError() 判断 + $ret = $redis->xGroup('CREATE', 'notify_stream', 'notify_group', '0-0', true); + if ($ret === false && strpos((string) $redis->getLastError(), 'BUSYGROUP') === 0) { + $redis->clearLastError(); // 组已存在属预期情况,清除错误标记 } } protected function claimPending(\Redis $redis, string $consumer): void { - // XAUTOCLAIM 需要 Redis 6.2+;旧版本用 XPENDING + XCLAIM 组合 + // 返回 [0 => 下一游标, 1 => [entry_id => [字段 => 值]], 2 => 已删除id(仅 Redis 7.0+ 有第三元素)] + // XAUTOCLAIM 需要 phpredis >= 6.0.0 且 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']); @@ -392,7 +423,7 @@ class NotifyStreamConsumer extends TimerController | 原则 | 原因 | |------|------| | **用 Stream 不用 List** | `RPOP/BRPOP` 是 at-most-once:弹出后消费者崩溃,消息永久丢失。Stream 消费组把"取出未确认"的消息留在 pending 列表,可被 XAUTOCLAIM 重投 | -| **先标后投** | "从 DB 捞出"与"写入 Redis"之间崩溃会丢数据。先 `status=queued` 再 xAdd,配合回收(`status=4 AND update_time < now-600` 重置为 0)。反过来会重复投递——重复比丢失好处理(幂等可兜底),丢失是静默的 | +| **先标后投** | "从 DB 捞出"与"写入 Redis"之间崩溃会丢数据。先 `status=4` 再 xAdd,配合回收(`status=4 AND update_time < now-600` 重置为 0;依赖表有 `update_time` 字段,CURD 表默认自带,没有则补)。反过来会重复投递——重复比丢失好处理(幂等可兜底),丢失是静默的 | | **幂等消费** | at-least-once 下消息可能投递两次(认领重试、生产者重复入队),回写用条件 UPDATE(`WHERE status != done`)兜底 | 其他要点: @@ -449,6 +480,7 @@ $item->save(); | `$this->concurrencyId` | 消费者编号,用于日志排查、worker 标识、Stream consumer 名;**不用于划分数据** | | 时间窗口模式 | `maxRunTime` + 循环消费 + 超时退出,见 [ulthon-timer](../ulthon-timer/SKILL.md);`frequency >= maxRunTime * 2` | | `run_type` | 消费者任务默认 `auto`(单节点跑全部消费者);要各节点分摊时切 `all`。回收/生产者任务恒为 `auto` 或 `main`(全局单份) | +| 控制器 `$frequency`(防刷) | 消费型任务**一般不设**:手动触发 bypass 的是定时器侧节流,不 bypass 控制器防刷——防刷窗口内手动触发会被"请不要频繁请求"拦截。确需防外部刷再设,且应显著小于调度 frequency | | 执行日志 | `/do` 自动重写为 `/execute`,每轮消费的耗时/结果进 `ul_system_timer_log`,配合定时器日志页排查 | ## 常见坑位(快速自检) @@ -460,3 +492,7 @@ $item->save(); - **模式 C 多节点但 Redis 不共享**:各节点消费各自的 Stream,消息错乱。多节点必须共享同一个 Redis 实例 - **消费者 `sleep` 写在处理分支里**:有数据时 sleep 会拖慢吞吐。只在队列空时休眠 - **把 `concurrency` 当数据分片数理解**:又绕回 `id % N` 反模式。重读核心原则第 1 条 +- **用了 `$this->concurrencyCount` 却没定义**:`TimerControllerBase` 只注入 `$concurrencyId` 属性,`concurrency_count` 参数需子类自行从 `$this->request->param()` 取并存为属性,直接用未定义属性会在运算时抛 TypeError +- **未配置 `redis` store 就调 `Cache::store('redis')`**:框架默认 cache 只有 `file`/`login`/`mcp_session`,先按「环境前提」配置,否则抛 `InvalidArgumentException` +- **phpredis Stream API 两个易错点**:`xReadGroup` 返回结构嵌套流名层(`[流名 => [entry_id => [字段 => 值]]]`,遍历前先取 `$msgs['流名']`,否则会把流名当 entry_id 去 xAck);phpredis **没有** `xGroupCreate` 方法(统一入口 `xGroup('CREATE', $key, $group, $id, $mkstream)`),且组已存在的 BUSYGROUP 错误**不抛异常**而是返回 false,需用 `getLastError()` 判断 +- **模式 B 的 `frequency` 与 `maxRunTime` 比例失衡**:`maxRunTime >= frequency` 时两次调度窗口重叠、消费节奏错乱,保持 `frequency >= maxRunTime * 2`