简介这份资源面向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消息体序列化后当 memberZRANGEBYSCORE 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。我一般会把消息体单独放一个 HashZSet 的 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 里加上 extensionswoole.so 和 extensionredis.so php -m | grep -E swoole|redis参数说明pecl install会编译安装Swoole 编译时可以加--enable-openssl等选项但延时队列用不到网络加密默认即可。装完必须重启 PHP-FPM 或者 CLI 环境否则php -m看不到。这一步翻车的多半是忘了在 CLI 的 php.ini 里加 extensionWeb 和 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 只存 msgIdscore 是到期时间 $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把消息体存 Hashkey 是delay_queue:payloadfield 是 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 独立连 RediszRangeByScore的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 resultPHP 侧调用$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 原子弹出是能扛住的。希望帮到你。本文还有配套的精品资源点击获取 
长河君 · 2026/10/25 11:32:11 · 阅读 1.2k Redis延时消息队列配Swoole多进程消费端：避坑与实战
简介这份资源面向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消息体序列化后当 memberZRANGEBYSCORE 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。我一般会把消息体单独放一个 HashZSet 的 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 里加上 extensionswoole.so 和 extensionredis.so php -m | grep -E swoole|redis参数说明pecl install会编译安装Swoole 编译时可以加--enable-openssl等选项但延时队列用不到网络加密默认即可。装完必须重启 PHP-FPM 或者 CLI 环境否则php -m看不到。这一步翻车的多半是忘了在 CLI 的 php.ini 里加 extensionWeb 和 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 只存 msgIdscore 是到期时间 $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把消息体存 Hashkey 是delay_queue:payloadfield 是 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 独立连 RediszRangeByScore的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 resultPHP 侧调用$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 原子弹出是能扛住的。希望帮到你。本文还有配套的精品资源点击获取