Cron 批量任务行级 Claim Token：彻底消灭重复发送

某后台管理系统有一个「批量消息发送」模块：运营创建任务，Cron 每分钟扫描待执行任务，分批把消息推给每一个目标用户。上线初期一切正常，直到某次大批量发送，部分用户反馈收到了两条一模一样的消息。

翻日志，两条推送时间戳相差不到 3 秒，来自两个不同 Cron 进程。理论上已经有任务级 lock_at CAS 防重，怎么还会撞车？

现象

同一用户在同一任务下收到重复消息，时间窗口在 1~2 分钟内
数据库里该任务的目标行 status 最终为 SENT，但 send_at 有两个不同时间戳（最后一次写入覆盖了前一次）
人工重跑任务（--task-id 参数）与 Cron 自动续跑同时在线时，问题必现

根因

系统的锁只做到了任务级，遗漏了目标行级：

Cron A 抢到任务锁，开始执行，50 秒后主流程判断「任务耗时过长」，主动释放任务锁（status=PENDING, lock_at=NULL），让下一轮 Cron 接续
Cron B 在下一分钟触发，拿到任务锁，执行 SELECT WHERE status=PENDING LIMIT 100，取出 100 行目标
此时 Cron A 仍在处理它上一轮 SELECT 拿到的那 100 行——这批行尚未被标记为 SENT，两个进程持有完全相同的目标行集合
两者都调用发送 API，用户收到两条消息

坑：任务级 CAS 无法防止行级并发读

lock_at CAS 只能保证同一时刻只有一个进程「持有任务锁」，无法防止两个进程读到同一批目标行。只要存在「任务超时释放 → 新 Cron 接续」或「手动触发绕过 CAS」的路径，行级竞态就必然存在。即使是单机 Cron，--force / --task-id 手动触发与自动 Cron 并发时同样会命中这个坑。

正解

在目标表增加一列 cron_token 和一个中间状态 PROCESSING，用 UPDATE-then-SELECT 模式做行级声索（Claim）：

-- schema 改动
ALTER TABLE xxx_task_target
  ADD COLUMN cron_token VARCHAR(40) DEFAULT NULL COMMENT '进程声索令牌',
  ADD INDEX idx_task_token (task_id, cron_token);

// 1. 生成本进程唯一 token
$batchToken = self::uuid() . '_' . getmypid();
 
// 2. 先 UPDATE：原子性地将 PENDING 行声索为 PROCESSING
$claimed = Db::name(self::TABLE_TARGET)
    ->where('task_id', $taskId)
    ->where('status', self::TARGET_STATUS_PENDING)
    ->limit(self::BATCH_SIZE)
    ->update([
        'status'     => self::TARGET_STATUS_PROCESSING,
        'cron_token' => $batchToken,
        'send_at'    => date('Y-m-d H:i:s'),
    ]);
 
if ((int)$claimed === 0) {
    // 无 PENDING 行；若也无 PROCESSING 行则任务完成
    $remaining = Db::name(self::TABLE_TARGET)
        ->where('task_id', $taskId)
        ->whereIn('status', [TARGET_STATUS_PENDING, TARGET_STATUS_PROCESSING])
        ->count();
    if ((int)$remaining === 0) self::markFinished($taskId);
    return;
}
 
// 3. 再 SELECT：只取本进程声索的行
$batch = Db::name(self::TABLE_TARGET)
    ->where('task_id', $taskId)
    ->where('cron_token', $batchToken)
    ->select();
 
// 4. 逐行处理，完成后清空 token
foreach ($batch as $target) {
    try {
        $result = sendToUser($target['openid'], ...);
        Db::name(self::TABLE_TARGET)->where('id', $target['id'])->update([
            'status'     => self::TARGET_STATUS_SENT,
            'cron_token' => null,
            'send_at'    => date('Y-m-d H:i:s'),
        ]);
    } catch (\Exception $e) {
        Db::name(self::TABLE_TARGET)->where('id', $target['id'])->update([
            'status'     => self::TARGET_STATUS_FAILED,
            'cron_token' => null,
        ]);
    }
}

MySQL 的 UPDATE ... LIMIT N 是原子的：并发执行时两个进程各自拿到不重叠的 N 行，后续 SELECT WHERE cron_token=? 精确取回自己的那批，完全隔离。

还要处理「进程崩溃导致行卡在 PROCESSING」的僵尸情况——在每次任务启动时清理超时的 PROCESSING 行：

protected static function releaseProcessingZombies($taskId)
{
    $cut = date('Y-m-d H:i:s', time() - 600); // 10 分钟超时
    Db::name(self::TABLE_TARGET)
        ->where('task_id', $taskId)
        ->where('status', self::TARGET_STATUS_PROCESSING)
        ->where('send_at', '<', $cut)
        ->update(['status' => self::TARGET_STATUS_PENDING, 'cron_token' => null, 'send_at' => null]);
}

正解：UPDATE-then-SELECT 行级声索

先 UPDATE 写入进程唯一 cron_token（原子），再 SELECT 按 token 取回己方行。MySQL 的行锁保证两个并发 UPDATE 永不返回重叠的行集合，这是比 SELECT ... FOR UPDATE SKIP LOCKED（MySQL 8+）更兼容老版本的等效方案。

双层锁缺一不可

任务级 lock_at CAS（防止两个 Cron 抢同一个任务）和行级 cron_token Claim（防止两个 Cron 读同一批目标行）是两个正交的保护层，必须同时在位。只有任务级锁的系统，在「超时续跑」和「手动触发」场景下行级竞态必然暴露。

一句话外卖

批量任务防重复，要锁到「行」这一层：先 UPDATE 声索、再 SELECT 取回，任务级 CAS 永远替代不了行级 Claim Token。