简介：这份资源面向PHP后端开发者与消息队列学习者，提供一套基于Redis延时队列、结合Swoole多进程模型构建的高并发消费端实现，可用于订单超时关闭、定时任务触发等需要延迟处理的业务场景。压缩包为zip格式，整体约1.66MB，内含项目源码及相关资源文件，涉及Redis的Sorted Set延时队列设计、ZRANGEBYSCORE取数逻辑、Swoole Server多进程消费配置与消息处理等核心模块，适合具备一定PHP与Swoole基础的中高级开发者研读。目前已有1269人学习下载，读者可通过源码理解延时消息的入队、到期触发与并发消费全流程，掌握多进程下消息吞吐提升与资源消耗控制的实现思路，并借鉴其目录组织与配置方式，快速搭建或改造自己的延迟消息处理服务。

1. Redis延时消息队列配 Swoole 多进程消费端：为什么单进程扛不住，多进程又容易翻车

订单超时未支付要自动关单，优惠券到期要自动失效，这类需求背后都是同一件事：把一条消息延迟到未来某个时间点再消费。用 Redis 做延时消息队列，常见做法是用 ZSet 存消息体和到期时间戳，消费端轮询ZRANGEBYSCORE取出到期消息。单进程跑起来很顺，但一旦消息量上来，轮询和业务处理串在一条线上，吞吐立刻见底。这时候把消费端换成 Swoole 多进程，用多个 worker 并行拉取和处理，是很多 PHP 团队会走的路。但多进程一上，重复消费、消息丢失、进程假死、Redis 连接超时这些坑就全冒出来了。这篇笔记就围绕「Redis 延时消息队列 + Swoole 多进程消费端」这条链路，把选型理由、最小可跑代码、参数怎么调、坑在哪，一层层拆开讲清楚，适合正在做订单超时、定时任务、异步解耦的 PHP 后端照着落地。

2. 延时队列的数据结构选型与 Swoole 多进程消费模型

2.1 为什么用 ZSet 而不是 List 或 Stream

延时消息的核心诉求是「按到期时间排序，到点了才取」。Redis 的 ZSet 天然带 score 排序，把消息的到期时间戳当 score，消息体序列化后当 member，ZRANGEBYSCORE queue 0 now LIMIT 0 N就能捞出所有已到期消息。List 只能做先进先出，没法按时间排序；Stream 虽然支持消费组和 ACK，但延时语义要自己靠时间戳过滤，反而多一层复杂度。常见做法是 ZSet 存延时消息，处理成功后再从 ZSet 删除，配合一个 Hash 或独立 key 存消息体，避免 member 过大。

这里有个容易忽略的点：ZSet 的 member 必须唯一。如果两条消息内容完全一样，member 相同会被覆盖，消息就丢了。所以 member 里要带唯一 ID，比如msg:{uniqid}:{payload}，或者干脆 member 只存消息 ID，消息体放 Hash。我一般会把消息体单独放一个 Hash，ZSet 的 member 只存消息 ID，这样 ZSet 体积小，轮询快，也方便排查。

2.2 Swoole 多进程消费端的进程模型

Swoole 的多进程消费端一般有两种写法：一种是用Swoole\Process手动 fork 多个 worker，每个 worker 独立轮询；另一种是用Swoole\Server的 Task Worker 或者Swoole\Timer配合多进程。做延时队列消费，我更倾向手动 fork 固定数量的 worker，每个 worker 一个 Redis 连接，独立轮询，互不干扰。进程数一般设成 CPU 核数的 1 到 2 倍，太多反而因为 Redis 连接数和上下文切换拖慢整体。

进程模型的关键是「谁负责拉，谁负责处理，失败了怎么办」。简单做法是每个 worker 自己拉自己处理，拉的时候用ZRANGEBYSCORE加LIMIT控制批量大小，处理完再ZREM。但这样有个问题：如果 worker 拉到消息后进程崩了，消息还在 ZSet 里，下次会被重新拉到，相当于至少一次消费。要保证不丢，就得接受可能重复，业务侧做幂等。这是延时队列最常见的取舍，后面避坑章节会细说。

2.3 最小可跑的目录结构与依赖

先给一个能直接跑起来的最小结构，不依赖框架，纯 Swoole + phpredis 扩展。目录大概长这样：

delay-queue/ ├── producer.php # 生产端，往 ZSet 塞延时消息 ├── consumer.php # 消费端，Swoole 多进程轮询 └── composer.json # 只声明依赖，实际用 pecl 装 swoole 和 redis

依赖上，Swoole 用 pecl 装，Redis 扩展也用 pecl 装 phpredis。Windows 下装 Swoole 比较折腾，常见做法是走 WSL 或者 Docker，生产环境基本都是 Linux，所以下面命令以 Linux 为准。装完用php -m | grep -E 'swoole|redis'确认两个扩展都加载了。

pecl install swoole pecl install redis # 在 php.ini 里加上 extension=swoole.so 和 extension=redis.so php -m | grep -E 'swoole|redis'

参数说明：pecl install会编译安装，Swoole 编译时可以加--enable-openssl等选项，但延时队列用不到网络加密，默认即可。装完必须重启 PHP-FPM 或者 CLI 环境，否则php -m看不到。这一步翻车的多半是忘了在 CLI 的 php.ini 里加 extension，Web 和 CLI 用的 ini 可能不是同一个。

3. 生产端与消费端的代码实现：从塞消息到多进程拉取

3.1 生产端：把延时消息写进 ZSet 和 Hash

生产端要做两件事：生成唯一消息 ID，把消息体写进 Hash，把 ID 和到期时间戳写进 ZSet。下面是最小实现：

<?php // producer.php $redis = new Redis(); $redis->connect('127.0.0.1', 6379); $redis->auth('yourpassword'); // 没有密码就删掉这行 function pushDelayMessage(Redis $redis, array $payload, int $delaySeconds): string { $msgId = uniqid('msg_', true); // 唯一 ID，避免 ZSet member 冲突 $expireAt = time() + $delaySeconds; // 到期时间戳，作为 score // 消息体放 Hash，字段名用 msgId，值用 JSON $redis->hSet('delay_queue:payload', $msgId, json_encode($payload, JSON_UNESCAPED_UNICODE)); // ZSet 只存 msgId，score 是到期时间 $redis->zAdd('delay_queue:zset', $expireAt, $msgId); return $msgId; } // 示例：10 秒后处理一条订单超时消息 pushDelayMessage($redis, ['order_id' => 10086, 'action' => 'close'], 10); echo "pushed\n";

逻辑说明：uniqid('msg_', true)带 more_entropy 参数，生成带小数点的唯一串，基本不会重复。hSet把消息体存 Hash，key 是delay_queue:payload，field 是 msgId。zAdd的 score 是到期时间戳，member 是 msgId。这样 ZSet 里只有短字符串，轮询时传输量小。参数上，$delaySeconds是延迟秒数，实际业务里可能是strtotime($expireTime) - time()。注意 Hash 不会自动清理，消费成功后要hDel，否则内存一直涨。

3.2 消费端：Swoole 多进程轮询与批量拉取

消费端是重点。下面这个 consumer.php 用Swoole\Processfork 出 N 个 worker，每个 worker 独立连 Redis，循环用ZRANGEBYSCORE拉到期消息，处理成功后ZREM并hDel。

<?php // consumer.php $workerNum = 4; // worker 数量，建议 CPU 核数 1~2 倍 $batchSize = 50; // 每次最多拉多少条 $intervalMs = 200; // 轮询间隔，毫秒 for ($i = 0; $i < $workerNum; $i++) { $process = new Swoole\Process(function (Swoole\Process $worker) use ($batchSize, $intervalMs) { $redis = new Redis(); $redis->connect('127.0.0.1', 6379); $redis->auth('yourpassword'); $redis->setOption(Redis::OPT_READ_TIMEOUT, 2); // 读超时，避免卡死 while (true) { $now = time(); // 拉取 score <= now 的消息，按 score 升序，最多 batchSize 条 $msgIds = $redis->zRangeByScore('delay_queue:zset', 0, $now, [ 'limit' => [0, $batchSize], ]); if (empty($msgIds)) { usleep($intervalMs * 1000); continue; } foreach ($msgIds as $msgId) { $payloadJson = $redis->hGet('delay_queue:payload', $msgId); if ($payloadJson === false) { // 消息体没了，说明已被其他 worker 处理，直接删 ZSet $redis->zRem('delay_queue:zset', $msgId); continue; } $payload = json_decode($payloadJson, true); try { handleMessage($payload); // 业务处理 // 处理成功，删 ZSet 和 Hash $redis->zRem('delay_queue:zset', $msgId); $redis->hDel('delay_queue:payload', $msgId); } catch (Throwable $e) { // 处理失败，不删，等下次重试；记录日志 fwrite(STDERR, "handle fail: {$msgId} " . $e->getMessage() . "\n"); } } } }, false, 0); // 不启用协程，普通进程模式 $process->start(); } // 父进程等待，防止退出 Swoole\Process::wait();

逻辑说明：每个 worker 独立连 Redis，zRangeByScore的limit控制批量，避免一次拉太多撑爆内存。OPT_READ_TIMEOUT设 2 秒，防止 Redis 卡住时 worker 永久阻塞。处理成功才zRem和hDel，失败就留着下次重试。这里有个并发问题：多个 worker 可能同时拉到同一条消息，因为zRangeByScore不是原子弹出。后面避坑章节会讲怎么用ZREM返回值做去重。

参数说明：$workerNum设 4 是保守值，实际按 CPU 和 Redis 负载调；$batchSize设 50，太大单次处理时间长，太小轮询频繁；$intervalMs设 200，没有消息时睡 200 毫秒，降低 Redis 压力。handleMessage是业务函数，自己实现，注意里面不要再连同一个 Redis 做长事务，容易和轮询连接互相影响。

3.3 用 ZREM 返回值做并发去重

上面代码有个隐患：worker A 和 worker B 同时拉到 msg_1，都去处理，业务就重复执行了。解决办法是利用ZREM的返回值：zRem返回实际删除的成员数，只有删到 1 的那个 worker 才算抢到消息。

foreach ($msgIds as $msgId) { // 先尝试抢占：只有删除成功（返回 1）的 worker 才处理 $removed = $redis->zRem('delay_queue:zset', $msgId); if ($removed !== 1) { continue; // 被别的 worker 抢走了 } $payloadJson = $redis->hGet('delay_queue:payload', $msgId); if ($payloadJson === false) { continue; } $payload = json_decode($payloadJson, true); try { handleMessage($payload); $redis->hDel('delay_queue:payload', $msgId); } catch (Throwable $e) { // 失败要放回去，否则消息丢了 $redis->zAdd('delay_queue:zset', time() + 5, $msgId); // 5 秒后重试 fwrite(STDERR, "retry: {$msgId} " . $e->getMessage() . "\n"); } }

逻辑说明：zRem是原子操作，多个 worker 同时删同一个 member，只有一个返回 1，其余返回 0。抢到的才继续处理，没抢到的直接跳过。失败时把消息重新zAdd回去，score 设成当前时间加 5 秒，实现延迟重试。注意重试次数要控制，否则一条坏消息会无限重试，常见做法是在 payload 里加retry_count，超过阈值就丢到死信队列。

参数说明：重试延迟 5 秒是经验值，业务慢可以调大。死信队列可以用另一个 ZSet 或 List，存超过重试上限的消息，人工排查。这一步是保证「至少一次消费」的关键，业务侧仍然要做幂等，因为进程崩溃等极端情况仍可能重复。

4. 避坑与排查：多进程消费端最容易翻车的 5 个点

4.1 现象：消息被重复消费，订单被关了两次

原因：多个 worker 同时zRangeByScore拉到同一条消息，没有做抢占去重，各自都执行了业务。解决：用zRem返回值做原子抢占，只有返回 1 的 worker 才处理，如 3.3 节所示。另外业务侧要加幂等，比如订单关单前先查状态，已关闭的直接跳过。

4.2 现象：Redis 报command timed out或read error on connection

原因：worker 轮询太频繁，或者单次zRangeByScore拉太多，Redis 连接读超时。也可能是 Redis 本身负载高，主从切换。解决：设OPT_READ_TIMEOUT，降低轮询频率，减小 batchSize。如果用的是 lettuce 这类 Java 客户端报RedisCommandTimeoutException，那是客户端侧超时，和 PHP 无关，但思路一样：调大超时、减少单次数据量、加连接池。PHP 这边 phpredis 没有连接池，多进程各自连，连接数等于 worker 数，一般可控。

4.3 现象：worker 进程假死，消息堆积但没人处理

原因：业务代码里有阻塞操作，比如同步 HTTP 请求没设超时，或者handleMessage里死循环。Swoole 普通进程模式下，一个 worker 卡住不影响其他 worker，但如果所有 worker 都卡住，队列就堆积。解决：给所有外部调用设超时，handleMessage里加最大执行时间检查，必要时用Swoole\Process::kill重启 worker。父进程可以加一个定时器，定期检查子进程状态，发现异常就重启。

4.4 现象：Hash 内存一直涨，Redis 内存告警

原因：消费成功后只zRem没hDel，消息体一直留在 Hash 里。或者处理失败的消息反复重试，Hash 里的 field 没清理。解决：成功处理必须同时zRem和hDel，失败重试的消息如果超过重试上限，也要hDel并转入死信队列。定期用HLEN delay_queue:payload监控 Hash 大小，和 ZSet 的ZCARD对比，差值大说明有泄漏。

4.5 现象：多进程启动后 CPU 飙高，Redis QPS 暴涨

原因：worker 数量太多，每个 worker 都在高频轮询，空轮询也占 CPU。解决：worker 数控制在 CPU 核数 1 到 2 倍，轮询间隔不要低于 100 毫秒。更好的做法是用 Redis 的BLPOP或者 Swoole Timer 做事件驱动，但延时队列用 ZSet 轮询是常见取舍，间隔调大一点，用usleep让出 CPU。监控上，用redis-cli info commandstats看zrangebyscore的调用次数，如果每秒几千次，说明轮询太密。

5. 进阶技巧：用 Lua 脚本把拉取和抢占合成原子操作

前面用zRangeByScore加zRem两步走，虽然zRem是原子的，但拉取和抢占之间仍有窗口，而且每个消息两次 Redis 往返。更稳的做法是写一个 Lua 脚本，在 Redis 服务端一次性完成「拉取到期消息 + 删除 ZSet + 返回消息体」，减少网络往返，也避免并发窗口。

-- pop_delay.lua -- KEYS[1] = zset key, KEYS[2] = payload hash key -- ARGV[1] = now timestamp, ARGV[2] = batch size local zsetKey = KEYS[1] local hashKey = KEYS[2] local now = tonumber(ARGV[1]) local batch = tonumber(ARGV[2]) local msgIds = redis.call('ZRANGEBYSCORE', zsetKey, 0, now, 'LIMIT', 0, batch) local result = {} for i, msgId in ipairs(msgIds) do local removed = redis.call('ZREM', zsetKey, msgId) if removed == 1 then local payload = redis.call('HGET', hashKey, msgId) if payload then redis.call('HDEL', hashKey, msgId) table.insert(result, msgId) table.insert(result, payload) end end end return result

PHP 侧调用：

$script = file_get_contents(__DIR__ . '/pop_delay.lua'); $now = time(); $result = $redis->eval($script, ['delay_queue:zset', 'delay_queue:payload', $now, 50], 2); // $result 是 [msgId1, payload1, msgId2, payload2, ...] for ($i = 0; $i < count($result); $i += 2) { $msgId = $result[$i]; $payload = json_decode($result[$i + 1], true); try { handleMessage($payload); } catch (Throwable $e) { // 失败重新入队，延迟 5 秒 $redis->zAdd('delay_queue:zset', time() + 5, $msgId); $redis->hSet('delay_queue:payload', $msgId, $result[$i + 1]); } }

逻辑说明：Lua 脚本在 Redis 里原子执行，ZRANGEBYSCORE拉取后立刻ZREM，只有删除成功的才HGET并HDEL，返回消息 ID 和消息体。PHP 侧一次eval拿到一批消息，处理失败再重新入队。这样每个消息只走一次网络往返，并发窗口也消除了。参数上，eval的第三个参数 2 表示前两个 key 是 KEYS，后面是 ARGV。注意 Lua 脚本里不要做太重的操作，批量大小控制在 100 以内，否则阻塞 Redis 主线程。

验证方法：起两个 worker，往 ZSet 里塞 100 条同时到期的消息，观察业务日志，确认每条消息只被处理一次。再用redis-cli monitor看命令，确认没有重复的ZREM。我自己的习惯是，任何多进程消费端上线前，先用小批量压测跑一遍，看重复率和堆积情况，没问题再放量。这套方案值不值得做，取决于你的消息量和实时性要求：如果每天几千条，单进程加长轮询就够；如果每秒几百条，Swoole 多进程加 Lua 原子弹出是能扛住的。希望帮到你。

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