学完这篇，你能给自己的 PHP 项目搭一套「任务失败自动退避重试、超过次数后与主队列分开保存的死信处理」的队列机制，不用引入 Redis、RabbitMQ 也能跑起来。
下面以 MySQL 单表队列为例，所有代码都能直接放进项目。
第一步：建一张能记录「重试状态」的任务表
队列要能重试，其实靠三个字段支撑：当前尝试次数、下次可执行时间、抢占标记。
CREATE TABLE jobs (
id BIGINT UNSIGNED AUTO_INCREMENT PRIMARY KEY,
queue VARCHAR(32) NOT NULL DEFAULT 'default',
payload TEXT NOT NULL,
attempts TINYINT UNSIGNED NOT NULL DEFAULT 0,
max_attempts TINYINT UNSIGNED NOT NULL DEFAULT 5,
available_at INT UNSIGNED NOT NULL, -- 下次可执行时间戳
reserved_at INT UNSIGNED NULL, -- 被 worker 抢占的时间
last_error TEXT NULL,
created_at INT UNSIGNED NOT NULL,
KEY idx_ready (queue, reserved_at, available_at)
) ENGINE=InnoDB;
idx_ready 这个联合索引必须建，出队查询全靠它，否则任务一多就是全表扫描。
第二步：用抢占式出队，避免两个 worker 抢同一个任务
出队分两步：先修改数据行来抢占，再取出整行内容。MySQL 5.7 没有 SKIP LOCKED，用事务加 FOR UPDATE 即可：
function pop(PDO $pdo, string $queue = 'default'): ?array {
$pdo->beginTransaction();
$sql = "SELECT * FROM jobs
WHERE queue = ? AND reserved_at IS NULL AND available_at <= ?
ORDER BY id LIMIT 1 FOR UPDATE";
$stmt = $pdo->prepare($sql);
$stmt->execute([$queue, time()]);
$job = $stmt->fetch(PDO::FETCH_ASSOC);
if (!$job) { $pdo->commit(); return null; }
$pdo->prepare("UPDATE jobs SET reserved_at = ?, attempts = attempts + 1 WHERE id = ?")
->execute([time(), $job['id']]);
$pdo->commit();
$job['attempts']++; // 本次就是第几次执行
return $job;
}
注意：attempts 一定要在「执行前」自增，不能在执行后。否则进程被 kill、OOM、超时中断时，这一次失败不会计入次数，任务就会无限重试。
第三步：写指数退避函数
失败后不要固定间隔重试，也不要在 worker 里执行 sleep() 等待。正确做法是改 available_at，让任务自己「消失」一段时间。
function backoff(int $attempt, int $base = 10, int $cap = 3600): int {
$delay = min($cap, $base * (2 ** ($attempt - 1))); // 10,20,40,80...
$jitter = random_int(0, (int) max(1, $delay * 0.2)); // ±20% 抖动
return $delay + $jitter;
}
第 1 次失败等约 10 秒，第 5 次约 160 秒，加上 $cap 封顶，避免退避时间变成几天。
注意：抖动（jitter）不能省。一次批量任务集体失败时，没有抖动的任务会在同一秒同时重新入队，恢复瞬间又会把数据库打挂。
第四步：失败处理分流——重试还是进死信
worker 捕获异常后只有两条路：未超过次数就重新排期，超过次数就搬进死信表。
function fail(PDO $pdo, array $job, Throwable $e): void {
if ($job['attempts'] < $job['max_attempts']) {
$pdo->prepare("UPDATE jobs SET reserved_at = NULL, available_at = ?, last_error = ? WHERE id = ?")
->execute([time() + backoff($job['attempts']), mb_substr($e->getMessage(), 0, 2000), $job['id']]);
return;
}
// 超过上限：搬进死信
$pdo->beginTransaction();
$pdo->prepare("INSERT INTO dead_jobs (queue, payload, attempts, last_error, failed_at)
SELECT queue, payload, attempts, ?, ? FROM jobs WHERE id = ?")
->execute([mb_substr($e->getMessage(), 0, 2000), time(), $job['id']]);
$pdo->prepare("DELETE FROM jobs WHERE id = ?")->execute([$job['id']]);
$pdo->commit();
}
死信表 dead_jobs 的结构与 jobs 基本一致，多一个 failed_at 就行。它和主队列分开，worker 永远不会取到，这样即使坏任务永远失败，也不会堵住好任务。
第五步：回收「僵尸任务」
worker 取走任务后中途挂了，reserved_at 会一直留着，这个任务就永远卡死。在每次出队前加一行回收：
$pdo->prepare("UPDATE jobs SET reserved_at = NULL WHERE reserved_at IS NOT NULL AND reserved_at < ?")
->execute([time() - 600]); // 10 分钟没跑完视为僵尸
注意：回收后这个任务会拿到新的 attempts 机会。所以单次执行时间要远小于回收阈值，长任务需要自行延长 reserved_at。
第六步：死信的人工复位
死信队列的价值在于「能看见、能救回来」。后台提供一个列表页，展示 queue / payload / attempts / last_error / failed_at，并提供「重放」按钮：
$pdo->prepare("INSERT INTO jobs (queue, payload, attempts, max_attempts, available_at, created_at)
SELECT queue, payload, 0, max_attempts, ?, ? FROM dead_jobs WHERE id = ?");
重放时 attempts 归零，任务重新走一轮完整退避。
第七步：接上调度入口
worker 用一个常驻循环，或由系统 crontab 每分钟调用一次，每次跑若干轮：
while ($job = pop($pdo)) {
try {
(new $job['payload']['class'])->handle(...);
$pdo->prepare("DELETE FROM jobs WHERE id = ?")->execute([$job['id']]);
} catch (Throwable $e) {
fail($pdo, $job, $e);
}
}
如果你跑的是 Clara BBS，它的计划任务体系（Cron::register 注册、前台访问懒触发 + /cron 外部入口精准触发、文件锁防并发重跑、结果落库可查）正好可以当作这个 worker 的调度器，把上面的 while 循环包进一个 Cron 任务里就行，不用另配守护进程。
注意：任务体必须幂等。重试和重放都会导致同一个任务被执行两次以上，发钱、发积分这类操作用唯一索引或状态机兜住。
小结
- 重试的本质是三个字段：
attempts（第几次）、available_at（何时再跑）、reserved_at（谁在跑）。
attempts 在执行前自增，防进程崩溃导致无限重试。- 退避用
base * 2^(n-1) 并加随机抖动，设上限封顶。
- 超过
max_attempts 的任务搬进 dead_jobs，与主队列隔离，避免堵死正常任务。
- 定期回收超时的
reserved_at，否则 worker 一崩任务就永久卡死。
- 死信必须可查、可重放，重放时把
attempts 归零。
- 所有任务逻辑写成幂等，重试才安全。
正式会员
认证极客
见习用户
