简介这是一份面向PHP后端开发者的延时消息队列实现资源核心利用Redis有序集合ZSet存储带时间戳的延迟消息并结合Swoole多进程模型并行消费以提升高并发场景下的吞吐能力。压缩包大小1.66MB页面暂未提供文件总数与类型明细内容应为Serve项目的源码及相关配置重点涉及Redis交互、Swoole进程管理和消息处理逻辑。资源已有1269人学习适合对异步任务、延迟队列、常驻内存服务感兴趣的开发者参考。通过阅读代码可以掌握Sorted Set实现延迟调度的思路、Swoole多进程消费者协作机制以及可扩展的消息确认与重试设计。该实现将Redis作为可靠存储利用ZRANGEBYSCORE按时取出到期消息再由Swoole子进程并行消费适用于订单超时、定时通知、任务重试等场景对优化PHP高并发处理能力具有直接借鉴价值。1. Redis 延时消息队列 Swoole 多进程消费端这套组合到底解决什么问题接到一个再常见不过的活儿订单超过 15 分钟没支付要自动关闭优惠券到期前要提醒用户。不少团队的做法是写一个 MySQL 任务表每分钟扫一次到期数据数据量小能用订单量大了之后一次 UPDATE 影响上千行数据库响应变慢延时从几十秒膨胀到分钟级。我后来把方案切成 Redis 延时消息队列 Swoole 多进程消费端Redis 按到期时间投递消息Swoole 在常驻内存里开多个进程并发消费。这套组合解决的核心问题是延时精度、消费吞吐、进程稳定性适合订单超时、活动定时上架、缓存过期回调这类「到点再干活」的业务。本文先讲实现原理再给一套能直接跑起来的多进程消费端代码和参数最后讲多进程模式下最容易踩的坑。2. 延时队列的实现原理ZSet 为什么够用状态怎么流转2.1 为什么是 ZSet 而不是数据库轮询或消息中间件三种方案的取舍先说结论Redis 的有序集合ZSet是实现延时队列最轻量可靠的基础结构前提是你对消息丢失的容忍度不是零。数据库轮询方案也就是在 MySQL 里建一张 task 表记 execute_time然后定时任务每分钟扫一次。这个方案在小数据量下没毛病但越往后越难受每一次全表扫描都在跟业务 SQL 抢数据库连接需要执行的任务扎堆在整点分钟时数据库的锁等待会明显变长更麻烦的是延时精度被定时任务的频率锁死想做到秒级就要把扫描频率提到秒级代价是数据库压力成倍上涨。RabbitMQ 延迟插件是另一个常见选项。它对消息投递的可靠性、消费确认的支持都比 Redis ZSet 完整得多但引入它意味着团队要维护一套独立的消息中间件集群而且延迟粒度受插件实现限制运维成本不低。如果业务里已经有 RabbitMQ 在用顺手接入是可以的如果只是为了一个订单超时功能单独引入一套集群我一般不建议。Redis ZSet 的取舍逻辑很直接ZSet 的 score 直接就是到期时间戳ZRANGEBYSCORE 取 0 到当前时间戳之间的成员就是「已经到期」的消息复杂度是 O(log(N) M)对十万级、百万级的延迟消息完全够用。Redis 本身大部分团队已经在用不需要新增基础设施。它的短板是消费确认机制得自己造消息去重也要自己处理这些后续章节会给出对应方案。延时精度上ZSet 方案能做到秒级甚至毫秒级取决于消费端的轮询间隔。相比数据库轮询的分钟级、RabbitMQ 延迟插件的分钟到小时级这个精度对于订单超时、活动倒计时、缓存过期补刷这类场景已经足够。2.2 到期消息怎么投递和消费ZSet 与 ready 队列的状态流转延时队列有一套固定的消息状态流转理解这套流转是写对代码的前提。我这里把整条链路拆成四个动作。生产端投递时把消息内容作为 member把「当前时间戳 延迟秒数」作为 score 写入 ZSet。这一步只做写入不做任何校验消息在到期之前不会出现在任何消费端面前。消费端做的事情是周期性扫描 ZSet 中 score 小于等于当前时间戳的成员。拿到的这些成员就是「已经到期的消息」但注意从 ZSet 里取出来不等于消息就安全了真正要设计的核心是「取出之后、消费之前」这段时间内消息怎么保证不丢。最常见的做法是先把到期的消息推进一个 Redis List我习惯叫它 ready 队列再从 ZSet 里删除最后让消费者从 List 里用阻塞读或者普通弹出取消息。为什么不能反过来先删再入列因为先 ZREM 再 LPUSH 这两条命令之间如果有任何一步失败消息就凭空消失了先入列再删除即使删除失败消息最多是被重复投递业务侧做一次幂等就能兜住不会丢。这个「先入列再删除」的顺序是整个延时队列可靠性的基石。三个操作之间不是原子的所以我在第 4 章会给一段 Lua 脚本把它们包成一个原子操作。消息进入 ready 队列之后消费进程用 LPOP 或者 BLPOP 取走消息开始业务处理。处理成功消息的生命周期就结束了处理失败进入重试逻辑重试达到上限就挪到一个死信队列里后续人工或者补偿任务再处理。整个链路里真正会丢消息的只有一个口子消息从 ready 队列被 LPOP 取出、但业务处理还没完成时消费进程崩溃。这个口子在绝大部分业务场景下是可以接受的但如果你的业务要求极端可靠就需要 BRPOPLPUSH 先把消息挪到一个 processing 队列再处理处理完再删。这个做法我放到第 5 章展开。2.3 为什么消费端必须常驻内存PHP-FPM 和 crontab 为什么不行延时队列的消费端在业务里就是个「每隔几百毫秒醒一次看看有没有到期消息」的角色。这个角色对 PHP 来说有两个天然的坎。第一个坎是 PHP-FPM 的生命周期。FPM 模式下每个请求处理完进程里的所有资源都会被释放你无法维持一个常驻的 Redis 长连接也没法在内存里保存任何状态。每处理一条消息都要重新走一遍框架初始化连接要重新建类要重新加载这个开销对于高频小消息是灾难。第二个坎是 crontab 调度。如果不用 Swoole很多人会写一个 CLI 脚本丢进 crontab每分钟跑一次。这个方案的问题很明确延时精度被钉死在分钟级消息到期后最长要等 59 秒才会被消费而且 crontab 每分钟拉起一个 PHP 进程光进程启动、脚本解析、扩展加载就要花掉几百毫秒重一点的项目能到一秒。Swoole 的 Process 模块解决的就是这两件事进程常驻内存Redis 连接建立一次就一直复用多进程并行消费吞吐量从单进程变成 N 个进程叠加。常驻进程带来的内存泄漏、僵尸进程、平滑退出这些新问题其实是「换取性能必须付出的管理成本」这正是第 5 章要讲的重点。3. 用 Swoole 搭多进程消费端进程模型与最小可运行架构3.1 进程模型选型Swoole Process、TaskWorker 还是协程进入代码之前先把进程模型选型说清楚因为这里选错了后面会反复返工。Swoole 里能承载消费逻辑的有三套东西Process进程、TaskWorker任务进程、Coroutine协程。TaskWorker 是投递式模型Master 进程收到消息后投给 TaskWorker 去执行它擅长的是「有任务来了处理一下」而我们的延时队列消费端是「主动去 Redis 里抢消息」没有投递方所以 TaskWorker 天然不对位。协程适合单进程内做大量 IO 并发但延时队列这种「一个进程一个长连接、每轮拉一批消息」的模型协程的收益不明显反而让代码里到处是上下文切换的隐性坑。我的选型是 Swoole Process手动 fork 出 N 个子进程每个子进程独立持有 Redis 连接、独立跑轮询循环。理由有三个第一子进程之间天然隔离一个进程崩溃不影响其他进程第二进程数可以按 CPU 核数精确控制不依赖框架内部的调度策略第三信号处理和退出逻辑完全可控方便嵌入 Supervisor 这类守护工具。3.2 最小可运行的进程容器fork 出 N 个消费 Worker整套消费端的骨架是一个父进程加 N 个子进程。父进程负责创建子进程、监听信号、回收退出的子进程并重新拉起子进程只做一件事就是跑消费循环。?php use Swoole\Process; $workerNum 4; // 初始进程数后面按 CPU 核数调 $processList []; for ($i 0; $i $workerNum; $i) { $process new Process(function (Process $worker) use ($i) { // 每个子进程单独命名方便 ps 排查 Process::setProcessName(delay_consumer:worker_{$i}); // 必须在子进程内部重新创建 Redis 连接 $consumer new DelayConsumer(); $consumer-run(); }); $process-start(); $processList[$process-pid] $process; } // 父进程处理 SIGTERM通知所有子进程退出 pcntl_signal(SIGTERM, function () use ($processList) { foreach ($processList as $process) { $process-kill(); } }); // 父进程进入回收循环某个子进程挂了就重新拉一个 while (true) { $ret Process::wait(); if ($ret $ret[code] ! 0) { // 子进程异常退出重新启动一个补充进去 } pcntl_signal_dispatch(); }这段代码的逻辑拆开看Process 的构造函数里传了一个闭包这个闭包就是子进程的入口Swoole 内部走的是 fork 系统调用所以子进程会继承父进程的全部内存状态和文件句柄。这就引出关键注意点——Redis 连接必须在子进程内部重新创建不能直接复用父进程的连接对象否则多个子进程共享同一个 socket读写会互相错乱这是新手最容易撞上的隐性故障之一。父进程这个 while 循环里Process::wait 是阻塞回收子进程的调用等价于 C 语言的 waitpid。当它返回一个数组时说明有一个子进程退出了数组里的 code 字段是退出码。如果退出码不等于 0说明子进程不是正常退出常见做法是在这里统计退出码并重新 fork 一个补位。整个骨架不依赖 Swoole 的 Server 模式单独用 Process 类就能跑很适合嵌入已有的 PHP 项目。3.3 消费进程的自我修养信号处理与平滑退出进程骨架搭好之后紧接着要处理的是两个生命周期问题更新代码时怎么平滑重启、子进程卡死时怎么处置。先说平滑退出。最粗暴的做法是直接 kill 掉整个进程组但这会让正在处理的业务逻辑中断极端情况下会留下半截状态。正解是走信号链路运维或者守护进程给父进程发 SIGTERM父进程把信号转发给所有子进程子进程在收到信号后停止接收新消息、把手头这一条处理完再退出。子进程里的信号处理要配合消费循环来写不能只注册信号处理函数就算完。常见的做法是消费循环里维护一个 isShutdown 标志位定时器或者主循环每次迭代检查这个标志收到信号就把标志置为 true循环自然退出。第二说僵尸进程。Swoole 的 Process 如果不做 wait 回收子进程退出后会变成僵尸进程吃内存。上面骨架里的 Process::wait 就是为了避免这个问题。如果某些场景你不能阻塞在 wait 上可以用非阻塞的方式配合定时器轮询回收但最少保证有回收逻辑。否则跑几天后ps aux里会挂满 defunct 进程运维看到会直接来找你。4. 消费端核心代码与参数从能跑调到扛得住4.1 投递与消费主循环把「取 入列 删除」变成原子操作先看生产端。投递延时消息的代码很短但有个细节值得说明member 里要塞完整消息体而不是只塞一个 ID因为消费端取出来之后要直接处理减少一次回查数据库。?php // 生产端投递一条 15 分钟延时消息 $msgId uniqid(order_, true); $payload json_encode([ msg_id $msgId, type order_timeout, order_id $orderId, ]); $delaySeconds 900; // 15 分钟 $score time() $delaySeconds; $redis-zAdd(delay_queue, $score, $payload);score 的单位是秒这里用 time() 计算到期时间戳。如果你需要毫秒级精度score 统一用毫秒时间戳消费端轮询时取毫秒也保持一致否则会出现消息提前被扫出来的问题。生产端这一段没有任何特殊逻辑纯写入。再看消费端主循环。消费端每一轮做三件事取到期消息、把到期消息推进 ready 队列、从 ZSet 删除。这三件事必须在一个 Lua 脚本里完成否则两个并发 Worker 会同时取到同一条消息。-- 延时队列取到期消息的 Lua 脚本 -- KEYS[1]: delay_queue ZSet -- KEYS[2]: ready_queue List -- ARGV[1]: 当前时间戳 -- ARGV[2]: 批量拉取数量 local got redis.call(ZRANGEBYSCORE, KEYS[1], 0, ARGV[1], LIMIT, 0, ARGV[2]) if #got 0 then return {} end for i, v in ipairs(got) do redis.call(LPUSH, KEYS[2], v) end redis.call(ZREM, KEYS[1], unpack(got)) return got这段脚本里最关键的是顺序先 LPUSH 后 ZREM任何一步失败都不会造成消息丢失——LPUSH 成功而 ZREM 失败结果是消息既在 ready 队列里也在 ZSet 里消费时会重复处理一次但幂等能兜住反过来先 ZREM 后 LPUSH一旦 LPUSH 失败消息就彻底没了。Lua 脚本在 Redis 里是原子执行的所以多进程场景下不会出现两个进程同时取到同一条消息的问题。调用端用 Redis 的 eval 命令执行这段脚本注意把脚本内容交给 Redis 做 script load拿到 SHA 之后用 evalsha 调用减少每次传输脚本内容的网络开销。主循环里拿到返回的已到期消息数组之后逐条丢给业务处理函数。?php // 消费端主循环每个 Worker 进程内运行 public function run(): void { while (!$this-isShutdown) { // Lua 脚本取到期消息原子地推入 ready 队列并删除 $items $this-takeExpiredJobs($this-nowMs(), 200); if (empty($items)) { // 没有到期消息睡 500ms避免空转 usleep(500 * 1000); continue; } foreach ($items as $item) { $this-handle($item); } } }这里有个容易被忽略的点takeExpiredJobs 返回的数组元素是 ZSet 里的 member也就是生产端塞进去的 JSON 字符串。handle 方法里要做的第一件事就是 json_decode 拿到 msg_id 和业务参数。每处理一条消息前先检查这个 msg_id 有没有处理过用 Redis SETNX 或者业务库的唯一索引做幂等键这一步能挡住绝大部分重复消费场景。4.2 消费端必调参数Worker 数量、轮询间隔、批量大小参数是延时队列「能不能扛住」的分水岭。我给出一套初始值和建议的调优方向实际部署时按业务形态改。参数建议初始值调优方向Worker 进程数CPU 核数消费逻辑里 IO 多就加到核数的 2 倍纯 CPU 计算别超过核数轮询间隔500ms消息量大缩短到 200ms空闲时退避到 2s 以上批量拉取数200单条消息体大就减到 50消息体小可以加到 500Redis 连接超时3s连不上 Redis 要快速失败别让进程无脑重试业务处理超时30s超过这个时间视为处理失败走重试逻辑Worker 数量的确定不是拍脑袋。消费逻辑如果主要耗在业务 API 调用、数据库读写上它是 IO 密集型的一个 CPU 核能扛住多个进程轮转进程数可以到核数的两倍如果消费逻辑是大量的 JSON 解析、加解密、规则计算它是 CPU 密集型的进程数超过核数反而会因为上下文切换变慢。先用初始值跑再观察 CPU 利用率和消费延迟来调。轮询间隔直接影响延时精度和 CPU 空转的平衡。间隔 200ms 意味着消息到期后最多晚 200ms 被取出来但如果没有到期消息每个进程每秒钟会空扫 5 次 ZSet。所以完好的实现一定是「有消息快轮询、没消息慢轮询」我在主循环里 sleep 500ms 是一个折中值空闲时可以根据 ZSet 里的最小 score 动态退避到 2 秒。Redis 官网对这类轮询有个朴素的建议宁可多睡一点不要空转烧 CPU。批量拉取数的逻辑是一批取 200 条逐条处理单条处理如果耗时 50ms200 条就要 10 秒而这段时间里新到期的消息会堆在 ZSet 里没人管。所以批量和轮询间隔要联动调整——一批消息的总处理时长不要超过轮询间隔的 3 倍否则会出现「到期高峰时消费不过来空闲时进程全在睡」。4.3 消息确认、失败重试与死信队列消费端最后一块拼图消费端没有内置 ack所以确认机制要自己设计。我的设计是三段式处理成功即确认处理失败进重试重试耗尽进死信。处理成功的语义很简单handle 方法正常返回消息生命周期结束不需要额外操作 Redis。处理失败要区分两种一种是临时失败比如下游 API 超时、数据库连接断开这种值得重试另一种是永久失败比如业务参数非法这种重试多少次都没用直接进死信。?php public function handle(string $item): void { $data json_decode($item, true); if (!$data) { // 解析失败属于永久性错误直接进死信 $this-pushDeadQueue($item); return; } try { $this-doBusinessLogic($data); } catch (\Throwable $e) { $retryCount intval($data[retry] ?? 0); if ($retryCount 3) { // 重试重新投递回 delay_queue10 秒后再执行 $data[retry] $retryCount 1; $this-redis-zAdd(delay_queue, time() 10, json_encode($data)); } else { // 超过重试上限挪到死信队列 $this-redis-zAdd(dead_queue, time() 86400, json_encode($data)); } } }重试的 score 用了 time() 10表示 10 秒后再次到期。这个值可以根据业务敏感性调订单超时这种对时间不敏感的可以放宽到 30 秒避免短时间内反复重试把下游打垮。死信队列我用了另一个 ZSetscore 是「保留到某个时间点」一个专门的巡检任务每天扫一次 dead_queue 把数据导出告警人工处理或者丢弃。这里要提醒一个常见误用不要在处理失败的 catch 里直接调用消费端自己的 run 方法重试那会让重试消息跳过「到期检测」直接进入处理逻辑破坏了延时队列「到点再消费」的语义。重试的正确姿势永远是「重新投递回 ZSet 等待到期」消费端的一视同仁才能保证整个链条的一致性。5. 多进程消费端 5 个必踩的坑现象、原因与排查方法5.1 坑一多个 Worker 同时取到同一条消息业务收到双份回调现象订单超时关闭逻辑被触发了两次同一张订单生成了两条关单记录。查日志发现两个 Worker 进程几乎同时在处理同一个 order_id时间差在毫秒级别。原因在没有 Lua 脚本的版本里Worker A 和 Worker B 同时执行 ZRANGEBYSCORE都拿到了同一条到期消息然后各自走去处理。ZRANGE 只是读取操作两个进程之间没有任何互斥消息被双取是必然结果。解决把「取数据、入 ready 队列、从 ZSet 删除」三个操作放进 Lua 脚本原子执行脚本前移到 4.1 里那一段。加了 Lua 之后Redis 保证同一时刻只会有一个进程能把某条消息从 ZSet 里取走。但注意Lua 只解决了「取消息」这一步的重复解决不了「业务处理」这一步的重复——比如消费端拿到了消息、业务执行成功、但在返回响应前网络断开了Swoole 的 Worker 会重启消息已经在 ZSet 里被删掉了反倒是不会重复。真正的重复风险集中在 ready 队列消费环节那里依然要依赖业务幂等。5.2 坑二没有到期消息时 CPU 跑满Redis 成了空转的陪跑现象业务量不大的时候某台机器上几个 Worker 进程的 CPU 占用合计接近 200%Redis 的每秒命令数异常高但实际消费成功的消息很少。原因主循环里 if empty continue 之后没有 sleep或者 sleep 时间只有 10ms。当 ZSet 里没有到期消息时循环会以极快的速度反复执行 Lua 脚本Redis 被这些无效的 ZRANGEBYSCORE 请求塞满CPU 全耗在空查上。解决空结果时至少睡 500ms并且做一次指数退避——连续 N 次取到空结果就把间隔翻倍上限 5 秒一旦取到消息就立刻恢复短间隔。我见过把 usleep 写成 sleep导致到期消息被延迟 1 秒才消费的这里单位一定要核对usleep 的参数是微秒500 * 1000 才是 500ms直接写 500 那是 0.5ms等于没睡。5.3 坑三Worker 崩溃之后消息无人消费ZSet 和 ready 队列里都找不到现象凌晨某个 Worker 进程因为内存超限被系统杀掉日志里没有报错但早上发现大量订单超时提醒漏发了。原因崩溃发生在「从 ready 队列 LPOP 取出消息」和「业务处理完成」之间。此刻消息既不在 ZSet 里已经被 Lua 删除也不在 ready 队列里已经被弹出完整地消失了。解决对可靠送达有要求的业务消费端改成 BRPOPLPUSH 模式——从 ready 队列弹出消息的同时把它推入一个 processing 队列业务处理成功后再 LREM 删除 processing 里的对应记录。如果进程在处理中途崩溃消息会留在 processing 队列里由巡检任务扫描 processing 队列把停留超时的消息重新投递回 delay_queue。做这个改造的代价是每条消息多两次 Redis 操作吞吐会下降约 20%但对于「不能接受丢消息」的业务是值得的。5.4 坑四进程数一多Redis 连接数被打满现象Redis 日志刷出 max number of clients reached部分 Worker 报「Cannot assign requested address」消费停止。原因两个叠加的问题。第一fork 的子进程继承了父进程创建的 Redis 连接句柄多个子进程共享同一个 socket连接数没有增加但报错却像连接耗尽第二每个进程如果每次循环都 new 一个 Redis 连接而不复用N 个进程乘上循环次数连接数瞬间被占满。解决子进程内每条连接必须各自创建并保存在进程内复用绝不能在父进程里预建连接再 fork循环体内要复用同一个连接对象只在断连时重建。进程数、Redis 的 maxclients 配置、以及机器上其它业务连接数三者要一起核算。我一般把消费端的连接数预算控制在 maxclients 的 30% 以内留出余量给其它业务。5.5 坑五常驻进程内存越跑越大最终被 OOM Kill现象Worker 进程刚启动时 RSS 只有 30MB跑三天涨到 300MB然后被系统以 OOM 杀掉。Supervisor 重新拉起后又开始循环。原因常驻进程的内存问题几乎都是「循环里累积」造成的。比如把日志字符串拼在循环变量里忘了清空比如每次循环 new 一个大数组处理完没有 unset比如 Swoole 的某些缓冲没有及时释放。PHP 的垃圾回收器能处理引用计数归零的变量但全局变量和长生命周期对象的循环引用需要 gc_collect_cycles 主动触发。解决一是消费循环的结尾统一 unset 大变量日志写入走文件追加而不是攒在内存二是每个进程内维护一个简单的内存水位检查每处理 1000 条消息调用一次 memory_get_usage(true)超过阈值比如 128MB 就主动退出进程让父进程重新拉起三是配合 Supervisor 配置 memory_limit 限制进程超限自动重启。主动退出比重启更可控因为它不留下半截状态父进程回收时也更干净。6. 消费端进阶压测校准参数时间轮优化进程守护6.1 用一场压测校准参数算 p99 而不是看平均数上线前先给这套系统做一次最小规模的压测。做法是写一个脚本往 delay_queue 里灌 10000 条消息每条的延时设为 30 秒然后统计每条消息实际被消费的时间与投递时间之差。真正的到期消息应该在 30 秒后陆续出现取这个差值的 p50 和 p99。如果 p99 大于 30 秒加 3 秒说明消费吞吐跟不上优先加 Worker 进程数如果 p99 接近理论值但 CPU 已经打满说明当前配置的吞吐峰值就在这里需要把批量拉取数往上调。平均值会掩盖尾部延迟延时队列最要命的就是 p99 的那批消息被拖慢。6.2 消息量到了千万级时间轮分桶代替单 ZSet上层的 ZSet 在消息量极大时会面临同一个 key 过大、ZRANGEBYSCORE 单次操作耗时变长的问题。常见优化是把一个 ZSet 拆成按时间分桶的多个 ZSet比如 delay_queue:ts_{分钟级时间戳}消费者每分钟只扫描当前分钟对应的那个桶。代价是可读性变差、过期消息的清理要额外处理收益是单个 key 的规模被限制住。这个优化不要提前做消息量到了千万级再切不迟。6.3 进程守护与上线习惯常驻进程一定要交给出 Supervisor 来管不要自己裸跑。配置里盯住 autorestarttrue、startsecs10 这两个参数前者保证进程死了被拉起后者防止启动后立刻崩溃造成疯狂重启。我现在给团队搭延时队列上线前只问自己三个问题进程挂了谁拉起消息重复了能不能幂等Redis 失联时消费端是阻塞还是快速失败。这三个问题想清楚比调任何参数都管用。希望帮到你。本文还有配套的精品资源点击获取