ulthon_admin/.agents/skills/ulthon-queue/SKILL.md

name, description

name	description
ulthon-queue	基于内置定时器设计任务队列与数据消费的规范（DB 原子抢占、区间游标、Redis Stream 生产者消费者三种模式）。需要异步处理数据表（消费队列、批量并行处理、worker 竞争）、设计队列表或生产者消费者、解决任务重复消费、并发竞争、超时回收，或排查 id 取余分片索引失效时调用。

queue（基于定时器的任务队列设计）

前置技能：ulthon-timer。本技能建立在 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 原子抢占（默认推荐）

锁思路（必须理解的原理）

错误写法（悲观锁长事务）：

// 反面教材：锁住 -> 处理 -> 才提交
Db::transaction(function () {
    $items = Db::name('task')->where('status', 0)->lock(true)->limit(20)->select();
    // 在事务内处理业务（耗时数秒）...
    // 期间其他消费者的 FOR UPDATE 全部阻塞等待，并发退化为串行
});

正确思路（CAS 乐观锁，三段生命周期）：

抢占（一条原子 UPDATE，持锁微秒级）
  -> 处理业务（不持任何锁，事务已提交）
  -> 回写结果（又一条短的条件 UPDATE）

// 抢占：检查和修改压进同一条原子语句
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；status 用 tinyint，禁止 ENUM）：

字段	类型	必要性	说明
status	tinyint	必须	0待处理 1处理中 2成功 3失败 4已入队（4 仅模式 C 使用）。若业务已有状态字段能表达"待处理"可复用，但独立字段不与业务状态机耦合，更稳
claim_time	int	必须	抢占时间戳，超时回收的唯一依据
claim_token	varchar(32)	批量抢占需要	本轮抢占令牌，区分"这轮我抢到哪几条"。每次只抢 1 条时可省
worker_id	varchar(64)	可选	消费者标识，排查用
error_message	varchar/text	可选	失败原因，排查用

-- 性能命根子：没有它，深积压时候选查询全表扫描
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

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 $maxRunTime = 3;     // 时间窗口，不超过 frequency 的一半
    // 一般不设控制器级 $frequency：防刷会在窗口内拦截手动触发（详见「与 timer 机制的组合」），
    // 调度节流已由定时器侧 Cache 节流承担；确需防外部刷 URL 再设，且应小于调度 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）：

[
    '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 的行会永久卡住。必须配一个独立回收任务：

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：区间游标（存量一次性跑批）

适用：把存量数据跑一遍（数据修复、全量刷新），不需要状态字段，不需要竞争。

class FixDataTask extends TimerController
{
    protected $concurrency = 4;      // 并行分段数
    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)，各自独立游标
        [$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）。

环境前提（缺失会直接报错）

1. config/cache.php 需配置 redis store——框架默认只带 file / login / mcp_session，未配置时 Cache::store('redis') 抛 InvalidArgumentException: Store [redis] not found：

// 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 防多节点重复投递）

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，动态可调）

class NotifyStreamConsumer extends TimerController
{
    protected $concurrency = 4;
    protected $maxRunTime = 3;

    public function do()
    {
        $redis = Cache::store('redis')->handler();
        $this->ensureGroup($redis);
        $consumer = 'c' . $this->concurrencyId;
        $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['notify_stream'])) {
                // 新消息为空 -> 认领超时未确认的（崩溃恢复）
                $this->claimPending($redis, $consumer);
                usleep(200000);
                continue;
            }

            foreach ($msgs['notify_stream'] 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
    {
        // 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
    {
        // 返回 [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']);
            $redis->xAck('notify_stream', 'notify_group', [$entry_id]);
        }
    }
}

可靠性三原则（模式 C 的生死线）

原则	原因
用 Stream 不用 List	RPOP/BRPOP 是 at-most-once：弹出后消费者崩溃，消息永久丢失。Stream 消费组把"取出未确认"的消息留在 pending 列表，可被 XAUTOCLAIM 重投
先标后投	"从 DB 捞出"与"写入 Redis"之间崩溃会丢数据。先 status=4 再 xAdd，配合回收（status=4 AND update_time < now-600 重置为 0；依赖表有 update_time 字段，CURD 表默认自带，没有则补）。反过来会重复投递——重复比丢失好处理（幂等可兜底），丢失是静默的
幂等消费	at-least-once 下消息可能投递两次（认领重试、生产者重复入队），回写用条件 UPDATE（WHERE status != done）兜底

其他要点：

• 消费组内每个 $this->concurrencyId 用独立 consumer 名，pending 认领才能跨消费者转移
• 多节点部署：所有节点连同一个 Redis（共享 Stream），run_type=all 让各节点都起消费者
• 消息体只放主键 id，业务数据回表取，避免消息体积膨胀与数据陈旧

反模式清单（禁止）

1. id 取余分片（索引失效）

// 反面教材：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（竞态，重复消费）

// 反面教材：读和标记之间不是原子的
$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；frequency >= maxRunTime * 2
run_type	消费者任务默认 auto（单节点跑全部消费者）；要各节点分摊时切 all。回收/生产者任务恒为 auto 或 main（全局单份）
控制器 $frequency（防刷）	消费型任务一般不设：手动触发 bypass 的是定时器侧节流，不 bypass 控制器防刷——防刷窗口内手动触发会被"请不要频繁请求"拦截。确需防外部刷再设，且应显著小于调度 frequency
执行日志	/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 条
• 用了 $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