简介这份资源面向PHP后端开发者与消息队列学习者聚焦高并发场景下的延迟消息处理难题给出基于Redis与Swoole多进程消费端的完整实现方案。其核心思路是用Redis的Sorted Set以时间戳为score暂存延时消息再借助Swoole Server的多进程模型并行消费从而提升系统吞吐量并降低阻塞开销。压缩包为zip格式整体约1.66MB内含项目源码及相关资源涉及Redis交互、Swoole多进程配置与消息处理等模块解压后可按目录结构对照研读。目前已有1269人学习下载适合希望理解延时队列原理、掌握Swoole多进程消费写法的中高级PHP开发者参考也可作为构建可扩展消息处理系统的实践样例。1. 当订单超时关闭遇上多进程消费这个 Redis 延时队列项目能帮你省掉一堆定时脚本电商后台最常见的需求之一下单后 30 分钟未支付自动关单同时释放库存。很多团队第一版都是 crontab 每分钟扫一次数据库量小的时候没问题一旦订单表上千万扫描 SQL 直接把数据库拖垮。这个项目就是冲着这类场景来的——用 Redis 的有序集合ZSET存延时任务用 Swoole 起多进程消费端把「到点触发」这件事从数据库轮询里彻底剥离出来。它适合正在用 PHP 做订单、通知、重试类业务的开发者尤其是已经上了 Swoole 但还没把延时队列跑稳的团队。整套代码不依赖额外中间件Redis 加 PHP 就能跑二次开发的门槛很低。2. 架构拆解ZSET 存任务、Swoole 多进程抢单的完整链路2.1 为什么是 ZSET 而不是 List延时队列的核心矛盾是「按时间排序」和「实时取出到期任务」。List 只能做到先进先出没法按 score 排序而 ZSET 的每个成员带一个 score天然就是按时间戳排序的优先队列。常见做法是把任务的执行时间戳毫秒级作为 score任务体序列化后作为 member。?php // 投递一个延时任务30 分钟后执行 $redis new Redis(); $redis-connect(127.0.0.1, 6379); $delay 30 * 60; // 秒 $executeAt (int)(microtime(true) * 1000) $delay * 1000; $task [ id uniqid(order_, true), type close_order, payload [order_id 1024], retry 0, ]; // score 是毫秒时间戳member 是 JSON 串 $redis-zAdd(delay_queue, $executeAt, json_encode($task));这里有几个参数值得注意。executeAt用毫秒而不是秒是因为秒级精度在高峰期会出现同一秒内大量任务堆积消费端一次拉太多容易打满内存。retry字段是给消费失败重试用的投递时初始化为 0。member 用 JSON 而不是 PHP serialize是为了跨语言消费时不用反序列化 PHP 对象后面做多语言扩展会省事。2.2 消费端怎么「到点才取」消费端不能无脑zRange全量拉那样每次都要扫整个 ZSET。正确姿势是用zRangeByScore带上限只取 score 小于等于当前时间的成员。?php // 每次只取到期的前 N 条 $now (int)(microtime(true) * 1000); $tasks $redis-zRangeByScore(delay_queue, 0, $now, [ limit [0, 100], // 偏移 0取 100 条 ]); foreach ($tasks as $member) { // 先删再执行避免多进程重复消费 $removed $redis-zRem(delay_queue, $member); if ($removed 0) { continue; // 已被其他进程抢走 } $task json_decode($member, true); dispatch($task); }zRem的返回值是关键ZSET 的删除是原子操作返回 1 表示这条任务确实被你删掉了返回 0 说明已经被别的进程抢走。这就是多进程「抢单」的实现方式不需要额外的分布式锁。limit的 100 是单次批量上限设太大单进程会长时间占用设太小空转次数变多一般按消费能力压测后定在 50 到 200 之间。2.3 Swoole 多进程的进程模型消费端用 Swoole 的Process起多个 worker每个 worker 独立跑一个循环。主进程只负责拉起和监控不参与消费。?php $workerNum 4; $workers []; for ($i 0; $i $workerNum; $i) { $process new Swoole\Process(function (Swoole\Process $worker) { // 每个子进程独立连 Redis避免连接共享 $redis new Redis(); $redis-connect(127.0.0.1, 6379); $redis-setOption(Redis::OPT_READ_TIMEOUT, -1); while (true) { $now (int)(microtime(true) * 1000); $tasks $redis-zRangeByScore(delay_queue, 0, $now, [limit [0, 100]]); if (empty($tasks)) { usleep(100000); // 100ms 空转间隔 continue; } foreach ($tasks as $member) { if ($redis-zRem(delay_queue, $member) 0) { continue; } dispatch(json_decode($member, true)); } } }, false, false); $pid $process-start(); $workers[$pid] $process; } // 主进程回收退出信号 Swoole\Process::wait(true);$workerNum设多少要看任务的平均执行耗时。如果单个任务执行 50ms4 个进程理论吞吐是 80 QPS如果任务里有 HTTP 调用耗时波动大建议先压测再定。usleep(100000)是空转间隔100ms 意味着最坏情况下任务会晚 100ms 执行对大多数业务够用如果要求更精确可以降到 10ms但 CPU 空转成本会上升。OPT_READ_TIMEOUT设为 -1 是防止 Redis 连接在长时间空闲后被服务端断开。3. 跑起来从零搭建到验证任务触发3.1 环境依赖和目录结构跑这个项目需要 PHP 7.4 以上、Swoole 4.8 以上、Redis 扩展。目录结构一般长这样delay-queue/ ├── src/ │ ├── Producer.php # 投递端 │ ├── Consumer.php # 消费端主逻辑 │ ├── Worker.php # 单进程消费循环 │ └── Handler/ │ └── CloseOrder.php # 具体任务处理器 ├── config/ │ └── redis.php ├── bin/ │ └── consumer.php # 启动入口 └── composer.jsonHandler目录是留给业务扩展的每种任务类型对应一个处理器类消费端根据type字段路由过去。这种设计的好处是新增任务类型不用改消费端核心代码只加一个 Handler 就行。3.2 启动消费端# 安装依赖 composer install # 启动 4 个消费进程 php bin/consumer.php start -w 4-w参数控制 worker 数量不传默认 2。启动后可以用ps aux | grep consumer确认进程数。如果进程起来后立刻退出先看 Redis 连接配置再看 Swoole 版本是否匹配——Swoole 4 和 5 的Process构造参数有差异这是最常见的翻车点。3.3 验证任务是否按时触发投递一条 10 秒后执行的任务观察日志输出?php // test_produce.php $redis new Redis(); $redis-connect(127.0.0.1, 6379); $executeAt (int)(microtime(true) * 1000) 10000; $redis-zAdd(delay_queue, $executeAt, json_encode([ id test_1, type close_order, payload [order_id 999], retry 0, ])); echo 投递完成10 秒后触发\n;执行后等 10 秒消费端日志里应该出现close_order的处理记录。如果没出现按这个顺序排查zRangeByScore的 score 上限是不是用了秒级时间戳和投递时的毫秒对不上、Redis 的 key 是不是写错了、消费进程有没有真的在跑。3.4 任务处理器的写法?php namespace Handler; class CloseOrder { public function handle(array $task): bool { $orderId $task[payload][order_id]; // 业务逻辑关单、释放库存 // 常见做法是加一层幂等判断防止重复执行 if ($this-isClosed($orderId)) { return true; } $this-close($orderId); return true; } }handle返回true表示消费成功返回false时消费端会把任务重新投递回 ZSETscore 设为当前时间加一个退避间隔。这个重试机制要配合retry字段做次数上限不然失败任务会无限循环。4. 避坑与排查多进程消费最容易踩的五个坑4.1 任务被重复消费现象同一个订单的关单逻辑执行了两次日志里出现两条相同 task id 的记录。原因通常是消费端先执行再删除或者删除和执行的顺序反了。解决方式是严格保证zRem在业务逻辑之前并且判断返回值。如果业务逻辑本身耗时较长zRem之后进程崩溃任务会丢失——这是「至少一次」和「至多一次」的取舍要精确一次就得在业务侧做幂等。4.2 消费进程 CPU 跑满现象top看到消费进程 CPU 占用 90% 以上。原因是空转间隔设得太小或者zRangeByScore的 limit 太大导致每次都在处理大量数据。把usleep调到 100ms 以上limit 降到 100 以内观察 CPU 是否回落。如果任务量本身就大应该加 worker 而不是减小间隔。4.3 任务延迟越来越大现象任务实际执行时间比预期晚了几分钟甚至更久。原因一般是 worker 数量不够或者单个任务执行太慢拖住了整个循环。先看日志里单任务的执行耗时如果某个 Handler 里有慢查询或外部调用把它拆成异步任务。另一个可能是 ZSET 里积压了大量过期任务消费端每次只取 100 条积压多的时候要临时调大 limit 或加进程。4.4 Redis 连接断开导致进程退出现象消费端跑一段时间后进程消失日志里有RedisException: read error on connection。原因是 Redis 服务端有timeout配置空闲连接会被断开。解决方式是设置OPT_READ_TIMEOUT为 -1或者在循环里加心跳。更稳妥的做法是捕获异常后重连不要让异常直接冒泡到进程顶层。4.5 时间戳精度不一致现象任务投递后不触发或者触发时间偏差很大。原因是投递端用了秒级时间戳消费端用毫秒级比较导致 score 永远大于当前时间。统一用毫秒并且在投递和消费两端都打印时间戳做对比。这个坑在新手项目里出现频率极高因为 PHP 的time()和microtime(true)很容易混用。5. 二次开发把延时队列改造成你自己的任务调度中心这个项目最值得改造的地方是 Handler 路由和重试策略。默认实现里dispatch是一个简单的 switch实际业务中任务类型会越来越多建议改成注册制?php // 在消费端启动时注册所有 Handler $handlers [ close_order \Handler\CloseOrder::class, send_sms \Handler\SendSms::class, retry_pay \Handler\RetryPay::class, ]; // dispatch 时查表 $handlerClass $handlers[$task[type]] ?? null; if (!$handlerClass) { // 未知类型直接丢弃避免死循环 return true; } (new $handlerClass())-handle($task);重试策略也可以从固定间隔改成指数退避。在任务体里加一个next_retry_at字段失败时按retry次数计算下次执行时间?php $retry $task[retry] 1; if ($retry 5) { // 超过上限落到死信队列人工处理 $redis-zAdd(delay_queue_dead, time() * 1000, json_encode($task)); return true; } $task[retry] $retry; $backoff min(300, pow(2, $retry)); // 2s, 4s, 8s... 最大 300s $nextAt (int)(microtime(true) * 1000) $backoff * 1000; $redis-zAdd(delay_queue, $nextAt, json_encode($task));死信队列这个设计我强烈建议加上。没有死信队列的延时队列失败任务要么无限重试拖垮消费端要么直接丢失无从排查。加一个delay_queue_dead配合一个简单的后台页面就能看到所有处理失败的任务排查效率提升非常明显。还有一个改造方向是加监控指标。每个 worker 在循环里定期往 Redis 写心跳和消费计数主进程汇总后暴露一个 HTTP 接口用 Prometheus 拉取。这样能实时看到队列积压量、消费速率、失败率。我一般会在Worker.php里加一个report()方法每处理 100 条任务上报一次成本很低但排查问题时能省大量时间。从那以后我每次上线延时队列都强制先跑一遍「投递 10 秒任务 观察日志 手动 kill 一个 worker 看任务是否被其他进程接管」这三步验证确认抢单和容错都正常再放量。希望帮到你。本文还有配套的精品资源点击获取 
长河君 · 2026/10/21 7:52:56 · 阅读 1.2k Redis ZSET + Swoole 多进程延时队列：订单超时关闭实战
简介这份资源面向PHP后端开发者与消息队列学习者聚焦高并发场景下的延迟消息处理难题给出基于Redis与Swoole多进程消费端的完整实现方案。其核心思路是用Redis的Sorted Set以时间戳为score暂存延时消息再借助Swoole Server的多进程模型并行消费从而提升系统吞吐量并降低阻塞开销。压缩包为zip格式整体约1.66MB内含项目源码及相关资源涉及Redis交互、Swoole多进程配置与消息处理等模块解压后可按目录结构对照研读。目前已有1269人学习下载适合希望理解延时队列原理、掌握Swoole多进程消费写法的中高级PHP开发者参考也可作为构建可扩展消息处理系统的实践样例。1. 当订单超时关闭遇上多进程消费这个 Redis 延时队列项目能帮你省掉一堆定时脚本电商后台最常见的需求之一下单后 30 分钟未支付自动关单同时释放库存。很多团队第一版都是 crontab 每分钟扫一次数据库量小的时候没问题一旦订单表上千万扫描 SQL 直接把数据库拖垮。这个项目就是冲着这类场景来的——用 Redis 的有序集合ZSET存延时任务用 Swoole 起多进程消费端把「到点触发」这件事从数据库轮询里彻底剥离出来。它适合正在用 PHP 做订单、通知、重试类业务的开发者尤其是已经上了 Swoole 但还没把延时队列跑稳的团队。整套代码不依赖额外中间件Redis 加 PHP 就能跑二次开发的门槛很低。2. 架构拆解ZSET 存任务、Swoole 多进程抢单的完整链路2.1 为什么是 ZSET 而不是 List延时队列的核心矛盾是「按时间排序」和「实时取出到期任务」。List 只能做到先进先出没法按 score 排序而 ZSET 的每个成员带一个 score天然就是按时间戳排序的优先队列。常见做法是把任务的执行时间戳毫秒级作为 score任务体序列化后作为 member。?php // 投递一个延时任务30 分钟后执行 $redis new Redis(); $redis-connect(127.0.0.1, 6379); $delay 30 * 60; // 秒 $executeAt (int)(microtime(true) * 1000) $delay * 1000; $task [ id uniqid(order_, true), type close_order, payload [order_id 1024], retry 0, ]; // score 是毫秒时间戳member 是 JSON 串 $redis-zAdd(delay_queue, $executeAt, json_encode($task));这里有几个参数值得注意。executeAt用毫秒而不是秒是因为秒级精度在高峰期会出现同一秒内大量任务堆积消费端一次拉太多容易打满内存。retry字段是给消费失败重试用的投递时初始化为 0。member 用 JSON 而不是 PHP serialize是为了跨语言消费时不用反序列化 PHP 对象后面做多语言扩展会省事。2.2 消费端怎么「到点才取」消费端不能无脑zRange全量拉那样每次都要扫整个 ZSET。正确姿势是用zRangeByScore带上限只取 score 小于等于当前时间的成员。?php // 每次只取到期的前 N 条 $now (int)(microtime(true) * 1000); $tasks $redis-zRangeByScore(delay_queue, 0, $now, [ limit [0, 100], // 偏移 0取 100 条 ]); foreach ($tasks as $member) { // 先删再执行避免多进程重复消费 $removed $redis-zRem(delay_queue, $member); if ($removed 0) { continue; // 已被其他进程抢走 } $task json_decode($member, true); dispatch($task); }zRem的返回值是关键ZSET 的删除是原子操作返回 1 表示这条任务确实被你删掉了返回 0 说明已经被别的进程抢走。这就是多进程「抢单」的实现方式不需要额外的分布式锁。limit的 100 是单次批量上限设太大单进程会长时间占用设太小空转次数变多一般按消费能力压测后定在 50 到 200 之间。2.3 Swoole 多进程的进程模型消费端用 Swoole 的Process起多个 worker每个 worker 独立跑一个循环。主进程只负责拉起和监控不参与消费。?php $workerNum 4; $workers []; for ($i 0; $i $workerNum; $i) { $process new Swoole\Process(function (Swoole\Process $worker) { // 每个子进程独立连 Redis避免连接共享 $redis new Redis(); $redis-connect(127.0.0.1, 6379); $redis-setOption(Redis::OPT_READ_TIMEOUT, -1); while (true) { $now (int)(microtime(true) * 1000); $tasks $redis-zRangeByScore(delay_queue, 0, $now, [limit [0, 100]]); if (empty($tasks)) { usleep(100000); // 100ms 空转间隔 continue; } foreach ($tasks as $member) { if ($redis-zRem(delay_queue, $member) 0) { continue; } dispatch(json_decode($member, true)); } } }, false, false); $pid $process-start(); $workers[$pid] $process; } // 主进程回收退出信号 Swoole\Process::wait(true);$workerNum设多少要看任务的平均执行耗时。如果单个任务执行 50ms4 个进程理论吞吐是 80 QPS如果任务里有 HTTP 调用耗时波动大建议先压测再定。usleep(100000)是空转间隔100ms 意味着最坏情况下任务会晚 100ms 执行对大多数业务够用如果要求更精确可以降到 10ms但 CPU 空转成本会上升。OPT_READ_TIMEOUT设为 -1 是防止 Redis 连接在长时间空闲后被服务端断开。3. 跑起来从零搭建到验证任务触发3.1 环境依赖和目录结构跑这个项目需要 PHP 7.4 以上、Swoole 4.8 以上、Redis 扩展。目录结构一般长这样delay-queue/ ├── src/ │ ├── Producer.php # 投递端 │ ├── Consumer.php # 消费端主逻辑 │ ├── Worker.php # 单进程消费循环 │ └── Handler/ │ └── CloseOrder.php # 具体任务处理器 ├── config/ │ └── redis.php ├── bin/ │ └── consumer.php # 启动入口 └── composer.jsonHandler目录是留给业务扩展的每种任务类型对应一个处理器类消费端根据type字段路由过去。这种设计的好处是新增任务类型不用改消费端核心代码只加一个 Handler 就行。3.2 启动消费端# 安装依赖 composer install # 启动 4 个消费进程 php bin/consumer.php start -w 4-w参数控制 worker 数量不传默认 2。启动后可以用ps aux | grep consumer确认进程数。如果进程起来后立刻退出先看 Redis 连接配置再看 Swoole 版本是否匹配——Swoole 4 和 5 的Process构造参数有差异这是最常见的翻车点。3.3 验证任务是否按时触发投递一条 10 秒后执行的任务观察日志输出?php // test_produce.php $redis new Redis(); $redis-connect(127.0.0.1, 6379); $executeAt (int)(microtime(true) * 1000) 10000; $redis-zAdd(delay_queue, $executeAt, json_encode([ id test_1, type close_order, payload [order_id 999], retry 0, ])); echo 投递完成10 秒后触发\n;执行后等 10 秒消费端日志里应该出现close_order的处理记录。如果没出现按这个顺序排查zRangeByScore的 score 上限是不是用了秒级时间戳和投递时的毫秒对不上、Redis 的 key 是不是写错了、消费进程有没有真的在跑。3.4 任务处理器的写法?php namespace Handler; class CloseOrder { public function handle(array $task): bool { $orderId $task[payload][order_id]; // 业务逻辑关单、释放库存 // 常见做法是加一层幂等判断防止重复执行 if ($this-isClosed($orderId)) { return true; } $this-close($orderId); return true; } }handle返回true表示消费成功返回false时消费端会把任务重新投递回 ZSETscore 设为当前时间加一个退避间隔。这个重试机制要配合retry字段做次数上限不然失败任务会无限循环。4. 避坑与排查多进程消费最容易踩的五个坑4.1 任务被重复消费现象同一个订单的关单逻辑执行了两次日志里出现两条相同 task id 的记录。原因通常是消费端先执行再删除或者删除和执行的顺序反了。解决方式是严格保证zRem在业务逻辑之前并且判断返回值。如果业务逻辑本身耗时较长zRem之后进程崩溃任务会丢失——这是「至少一次」和「至多一次」的取舍要精确一次就得在业务侧做幂等。4.2 消费进程 CPU 跑满现象top看到消费进程 CPU 占用 90% 以上。原因是空转间隔设得太小或者zRangeByScore的 limit 太大导致每次都在处理大量数据。把usleep调到 100ms 以上limit 降到 100 以内观察 CPU 是否回落。如果任务量本身就大应该加 worker 而不是减小间隔。4.3 任务延迟越来越大现象任务实际执行时间比预期晚了几分钟甚至更久。原因一般是 worker 数量不够或者单个任务执行太慢拖住了整个循环。先看日志里单任务的执行耗时如果某个 Handler 里有慢查询或外部调用把它拆成异步任务。另一个可能是 ZSET 里积压了大量过期任务消费端每次只取 100 条积压多的时候要临时调大 limit 或加进程。4.4 Redis 连接断开导致进程退出现象消费端跑一段时间后进程消失日志里有RedisException: read error on connection。原因是 Redis 服务端有timeout配置空闲连接会被断开。解决方式是设置OPT_READ_TIMEOUT为 -1或者在循环里加心跳。更稳妥的做法是捕获异常后重连不要让异常直接冒泡到进程顶层。4.5 时间戳精度不一致现象任务投递后不触发或者触发时间偏差很大。原因是投递端用了秒级时间戳消费端用毫秒级比较导致 score 永远大于当前时间。统一用毫秒并且在投递和消费两端都打印时间戳做对比。这个坑在新手项目里出现频率极高因为 PHP 的time()和microtime(true)很容易混用。5. 二次开发把延时队列改造成你自己的任务调度中心这个项目最值得改造的地方是 Handler 路由和重试策略。默认实现里dispatch是一个简单的 switch实际业务中任务类型会越来越多建议改成注册制?php // 在消费端启动时注册所有 Handler $handlers [ close_order \Handler\CloseOrder::class, send_sms \Handler\SendSms::class, retry_pay \Handler\RetryPay::class, ]; // dispatch 时查表 $handlerClass $handlers[$task[type]] ?? null; if (!$handlerClass) { // 未知类型直接丢弃避免死循环 return true; } (new $handlerClass())-handle($task);重试策略也可以从固定间隔改成指数退避。在任务体里加一个next_retry_at字段失败时按retry次数计算下次执行时间?php $retry $task[retry] 1; if ($retry 5) { // 超过上限落到死信队列人工处理 $redis-zAdd(delay_queue_dead, time() * 1000, json_encode($task)); return true; } $task[retry] $retry; $backoff min(300, pow(2, $retry)); // 2s, 4s, 8s... 最大 300s $nextAt (int)(microtime(true) * 1000) $backoff * 1000; $redis-zAdd(delay_queue, $nextAt, json_encode($task));死信队列这个设计我强烈建议加上。没有死信队列的延时队列失败任务要么无限重试拖垮消费端要么直接丢失无从排查。加一个delay_queue_dead配合一个简单的后台页面就能看到所有处理失败的任务排查效率提升非常明显。还有一个改造方向是加监控指标。每个 worker 在循环里定期往 Redis 写心跳和消费计数主进程汇总后暴露一个 HTTP 接口用 Prometheus 拉取。这样能实时看到队列积压量、消费速率、失败率。我一般会在Worker.php里加一个report()方法每处理 100 条任务上报一次成本很低但排查问题时能省大量时间。从那以后我每次上线延时队列都强制先跑一遍「投递 10 秒任务 观察日志 手动 kill 一个 worker 看任务是否被其他进程接管」这三步验证确认抢单和容错都正常再放量。希望帮到你。本文还有配套的精品资源点击获取