Swoole多进程+Redis ZSet延时队列：秒级订单超时关闭实战-育师

简介：这份资源面向PHP后端开发者与消息队列学习者，提供一套基于Redis延时队列与Swoole多进程模型构建的高并发消费端实现，可用于订单超时关闭、定时任务触发等需要延迟处理的业务场景。压缩包为zip格式，整体约1.66MB，内含项目源码及相关资源文件，涵盖Redis交互、Swoole多进程配置与消息消费逻辑等核心模块，便于读者直接解压研读与二次改造。目前已有1269人学习下载，说明其在PHP异步编程与队列实践领域具有一定参考价值。通过阅读源码，读者可以掌握用Sorted Set配合时间戳score实现延时投递、借助ZRANGEBYSCORE拉取到期消息，以及利用Swoole Server多子进程并行消费提升吞吐量的完整思路，同时理解非阻塞I/O在降低资源消耗方面的作用，适合作为进阶PHP工程师构建可靠消息处理系统的实战参考。

1. 从一次订单超时未关闭说起：这套 Redis 延时队列到底能干什么

电商后台最常见的翻车场景之一：用户下单后迟迟不付款，库存被锁死，订单状态却一直挂在“待支付”。你写个定时脚本每分钟扫一次数据库，订单量一上来，SELECT * FROM orders WHERE status=0 AND create_time < ?直接把慢查询打满，数据库 CPU 飙到 90%。更尴尬的是，扫描间隔设 1 分钟，用户实际超时时间就在 60 到 120 秒之间飘，业务方天天追着问“到底几分钟关单”。

这套基于 Swoole 多进程消费端的 Redis 延时消息队列，解决的就是这类“到点才触发”的定时任务。核心思路是把延时任务塞进 Redis 的有序集合（ZSet），score 存到期时间戳，消费端用 Swoole 起多个 worker 进程轮询到期任务并投递执行。相比纯 crontab 扫表，它把精度压到秒级，把数据库压力转移到内存，还能横向扩消费进程。适合谁？手上是 PHP 技术栈、已经在用 Redis、又不想为了一个延时功能去引 Kafka 或 RabbitMQ 的团队。下面我按“结构怎么设计 → 怎么跑起来 → 坑在哪 → 怎么验证”拆一遍。

2. 拆开看结构：ZSet 存任务、Swoole 多进程怎么分工

2.1 为什么是 ZSet 而不是 List

延时队列的本质是“按时间排序，到点取出来”。Redis 里能排序的结构就 ZSet 最合适：member 存任务内容（一般序列化后的 JSON），score 存到期时间戳（毫秒或秒，全项目统一即可）。取到期任务用ZRANGEBYSCORE queue 0 <now>，拿到就删，天然去重。

有人会问 List 行不行。List 是 FIFO，你没法按时间取，只能靠BLPOP阻塞等，但它不支持“延迟到某个时间点才可见”。用 List 做延时，常见做法是配合一个轮询线程把到期的搬到就绪 List，多了一层中转，反而更绕。ZSet 一步到位，代价是ZRANGEBYSCORE是 O(log N + M)，任务量百万级时单次拉取要控制 M 的大小，别一次拉几万条。

任务 member 的格式建议固定成带业务前缀的字符串，方便排查：

{ "job_id": "order_close_20240521_8891", "topic": "order.timeout", "payload": {"order_id": 8891, "uid": 100233}, "delay": 1800, "expire_at": 1716280000000, "retry": 0 }

expire_at是绝对到期时间，delay只作记录。为什么存绝对时间？因为任务可能在队列里躺很久，存相对 delay 的话消费端还得自己算基准，容易错。

2.2 Swoole 多进程的角色划分

单进程轮询有两个问题：一是ZRANGEBYSCORE和业务执行串在一起，业务一慢，拉取就卡；二是单核跑不满。Swoole 的进程模型刚好能拆开：

Master 进程：Swoole\Process或Swoole\Server的管理进程，负责拉起和回收 worker。
Timer 进程（1 个）：只做一件事，定时（比如每 200ms）从 ZSet 拉到期任务，投递到就绪队列。它不碰业务。
Worker 进程（N 个）：从就绪队列取任务，执行业务逻辑，失败按策略重试。

就绪队列可以用 Redis List，也可以用 Swoole 的Channel做进程间通信。用 Redis List 的好处是 worker 崩了任务不丢，坏处是多一次网络往返；用Channel快，但进程重启任务就没了。生产环境我一般选 Redis List，稳。

进程数怎么定？不是越多越好。worker 是 CPU 密集型就设成核数，IO 密集型（比如要调第三方接口）可以设到核数的 2 到 4 倍。Timer 进程永远只留 1 个，多个 Timer 会重复拉取，除非你在拉取时用ZREM的返回值做原子抢占。

2.3 拉取与投递的原子性

Timer 拉取时有个经典竞态：两个进程同时ZRANGEBYSCORE拿到同一批任务。解决办法是用ZREM的返回值判断——先ZRANGEBYSCORE拿 member，再对每个 member 执行ZREM，返回 1 说明抢到了，返回 0 说明被别人抢走，跳过。别用ZPOPMIN，它按 score 最小的弹，但你没法限定“只弹到期的”。

<?php // Timer 进程核心拉取逻辑 $now = (int)(microtime(true) * 1000); $jobs = $redis->zRangeByScore('delay_queue', 0, $now, ['limit' => [0, 200]]); foreach ($jobs as $job) { // ZREM 返回 1 才算抢到，避免多进程重复消费 if ($redis->zRem('delay_queue', $job) === 1) { $redis->lPush('ready_queue', $job); } }

limit控制单次拉取量，200 是个经验值，太大单次循环耗时长，太小空转多。$now用毫秒时间戳，和写入时保持一致，混用秒和毫秒是新手最容易犯的错。

3. 把队列跑起来：从写任务到消费的完整链路

3.1 投递一个延时任务

投递端就是往 ZSet 里ZADD，score 是当前时间加延时。封装成一个函数，统一入口，方便以后换存储。

<?php function pushDelayJob(Redis $redis, string $topic, array $payload, int $delaySec): string { $jobId = $topic . '_' . uniqid('', true); $job = [ 'job_id' => $jobId, 'topic' => $topic, 'payload' => $payload, 'delay' => $delaySec, 'expire_at' => (int)(microtime(true) * 1000) + $delaySec * 1000, 'retry' => 0, ]; // score 用到期毫秒时间戳，member 存 JSON $redis->zAdd('delay_queue', $job['expire_at'], json_encode($job, JSON_UNESCAPED_UNICODE)); return $jobId; } // 调用：30 分钟后关闭订单 pushDelayJob($redis, 'order.timeout', ['order_id' => 8891], 1800);

JSON_UNESCAPED_UNICODE别省，不然中文 payload 会变成\uXXXX，排查时看着头疼。job_id用uniqid保证唯一，方便日志追踪。

3.2 消费端进程启动

消费端用Swoole\Process起进程池，Timer 和 Worker 分开。下面是一个可跑的最小骨架。

<?php $workerNum = 4; $pool = new Swoole\Process\Pool($workerNum + 1); // 1 个 timer + N 个 worker $pool->on('WorkerStart', function (Swoole\Process\Pool $pool, int $workerId) { $redis = new Redis(); $redis->connect('127.0.0.1', 6379); $redis->auth('your_password'); if ($workerId === 0) { // Timer 进程：定时拉取到期任务 Swoole\Timer::tick(200, function () use ($redis) { $now = (int)(microtime(true) * 1000); $jobs = $redis->zRangeByScore('delay_queue', 0, $now, ['limit' => [0, 200]]); foreach ($jobs as $job) { if ($redis->zRem('delay_queue', $job) === 1) { $redis->lPush('ready_queue', $job); } } }); } else { // Worker 进程：阻塞取就绪任务并执行 while (true) { $job = $redis->brPop('ready_queue', 1); if (!$job) { continue; } $data = json_decode($job[1], true); try { handleJob($data); // 业务处理函数 } catch (Throwable $e) { retryJob($redis, $data, $e); // 失败重试 } } } }); $pool->start();

Swoole\Process\Pool的WorkerStart里每个进程独立连 Redis，别在on外面建连接再 fork，连接会被子进程共享导致数据错乱。brPop的超时设 1 秒，既能及时响应，又不会空转烧 CPU。handleJob和retryJob按业务自己实现。

3.3 重试与死信处理

任务失败不能无限重试。常见做法是重试次数写进 payload，超过阈值丢到死信队列（另一个 List 或 ZSet），人工介入。

<?php function retryJob(Redis $redis, array $data, Throwable $e): void { $data['retry'] = ($data['retry'] ?? 0) + 1; if ($data['retry'] > 3) { // 超过 3 次进死信，保留现场 $data['error'] = $e->getMessage(); $redis->lPush('dead_letter_queue', json_encode($data, JSON_UNESCAPED_UNICODE)); return; } // 指数退避：2^retry 秒后重试 $delay = pow(2, $data['retry']); $data['expire_at'] = (int)(microtime(true) * 1000) + $delay * 1000; $redis->zAdd('delay_queue', $data['expire_at'], json_encode($data, JSON_UNESCAPED_UNICODE)); }

指数退避避免失败任务瞬间打满重试。死信队列要配监控，不然任务悄悄堆着没人知道。重试次数阈值 3 是经验值，对时效敏感的业务可以降到 1 到 2。

3.4 关键参数对照

参数	建议值	说明
Timer 轮询间隔	200ms	越小精度越高，CPU 占用越大
单次拉取量 limit	100~500	按任务平均执行耗时调
Worker 进程数	CPU 核数 × 1~4	IO 密集取高值
brPop 超时	1s	兼顾响应与空转
最大重试次数	3	按业务容忍度调
退避基数	2	2^retry 秒

4. 避坑排查：这几个问题我踩过不止一次

4.1 任务重复消费

现象：同一个订单被关闭两次，日志里两条相同 job_id。原因：Timer 进程起了多个，或者拉取时没用ZREM返回值做原子判断，两个进程同时拿到同一 member。解决：Timer 只留 1 个进程；拉取必须ZRANGEBYSCORE加ZREM返回值判断，别用ZPOPMIN图省事。

4.2 时间戳单位混用

现象：任务要么立刻被消费，要么永远不触发。原因：写入用秒级time()，拉取用毫秒microtime(true)*1000，score 差三个数量级。解决：全项目统一毫秒，封装一个nowMs()函数，禁止各处自己写。

4.3 Worker 阻塞导致就绪队列堆积

现象：ready_queue长度持续上涨，任务延迟越来越大。原因：handleJob里有同步 HTTP 请求或大文件操作，worker 被占满。解决：把耗时操作异步化，或增加 worker 数；给handleJob加超时，超时任务走重试。

4.4 Redis 连接在 fork 后失效

现象：worker 启动后报Redis server went away。原因：在Pool外面建了 Redis 连接，fork 后子进程共用同一个 socket。解决：连接必须在WorkerStart回调里建，每个进程独立。

4.5 内存持续增长

现象：Redis 内存只涨不降，delay_queue里堆了大量过期任务。原因：Timer 进程挂了或拉取逻辑异常，到期任务没被清理。解决：给 Timer 进程加健康检查，挂了自动重启；定期用ZCOUNT delay_queue 0 <now>监控积压量，超阈值告警。

5. 验证与压测：怎么确认它真的按秒触发

写完不验证等于没写。我一般分三步：单任务精度、批量积压、故障恢复。

单任务精度：投一个 10 秒后触发的任务，记录投递时间和实际执行时间，差值应在 200ms 到 400ms 之间（取决于轮询间隔）。写个脚本循环 100 次，统计 P99 延迟。

<?php // 精度测试：投 100 个 5 秒任务，统计实际触发延迟 for ($i = 0; $i < 100; $i++) { pushDelayJob($redis, 'test.precision', ['seq' => $i], 5); } // 消费端记录每个任务的 (实际执行时间 - expire_at)，输出 P50/P99

批量积压：一次性投 10 万个任务，观察ready_queue长度变化和消费速率。如果消费速率上不去，先看 worker 是不是被慢业务卡住，再看 Redis 是不是成了瓶颈（SLOWLOG GET看慢命令）。

故障恢复：手动 kill 掉 Timer 进程，确认到期任务没丢（还在 ZSet 里），重启后能补消费。再 kill 一个 worker，确认就绪队列里的任务被其他 worker 接管。这两步过了，才算能上生产。

压测时用redis-cli --latency看 Redis 响应，用top -H看 Swoole 各进程 CPU 分布。Timer 进程 CPU 应该很低（个位数），如果它飙高，多半是轮询间隔太短或单次拉取量太大。

从那以后我每次上线延时队列，都强制先跑一遍精度测试和 kill 恢复测试，确认 P99 延迟和任务不丢，才敢接真实业务。希望帮到你。

本文还有配套的精品资源，点击获取