简介基于 Redis 延时队列与 Swoole 多进程模型实现的消息队列消费端面向已具备 PHP 基础、希望深入理解高并发消费架构的中高级开发者。它针对的问题是如何在保证延时精度的前提下让多个消费者并行、稳定地从队列中取件并处理。实现思路上以 Redis 有序集合承载待投递消息用时间戳作为 score到期后通过 ZRANGEBYSCORE 批量取出消费侧依托 Swoole Server 的多进程能力让每个子进程独立拉取与处理消息形成可横向扩展的消费集群同时借助非阻塞 I/O 降低空转开销。压缩包约 1.66MB主体为该消费服务的 PHP 工程源码涵盖 Redis 队列读写、多进程消费配置、消息处理与异常兜底等模块结构紧凑便于直接对照阅读与二次改造。已有 1270 人学习。1. 订单超时关单引出的一套消费端订单下单后 30 分钟未支付就要自动关单这是延时队列最常见的落地场景。很多团队第一反应是挂一个 crontab 每分钟扫一次数据库量小的时候没问题单表过了百万行之后扫描延迟、锁竞争、重复执行这些玄学问题就会一个接一个冒出来。这套 Redis 延时消息队列换了个思路用 ZSet 的 score 存到期时间戳用 Swoole 拉起多个常驻进程做消费端消息只有真正到期了才进入消费链路。它适合已经跑着 PHP 7.4 以上、Redis 有余量、又不想为了一个延时任务单独引入一套消息中间件的团队订单关单、支付超时提醒、失败重试、定时推送这类场景都能直接套上去。下面从架构一直拆到能跑的代码。2. 架构与进程模型Redis ZSet 加 Swoole 多进程怎么分工先把整套东西的骨架摆出来不然后面贴代码容易看散。这个项目的核心其实就是两个 Redis 结构加三类进程结构简单到你可以在一张纸上画完但每个角色的职责边界划得很清楚改起来不容易牵一发而动全身。2.1 为什么延时用 ZSet 而不是 ListList 只能做先进先出做不了「按时间到期」这件事。有人会用LPUSHBRPOP再配合sleep来模拟延时问题是 sleep 的精度和并发一塌糊涂多个消费者各自 sleep 不同时长消息顺序直接乱掉。ZSet 的 score 天然就是排序字段把「到期执行时间」写成毫秒时间戳塞进 score取消息时只取score 当前毫秒时间戳的成员语义完全对得上。常见做法是把延时队列拆成两个结构结构类型作用dq:delay:{topic}ZSet存未到期的消息score 是到期毫秒时间戳dq:ready:{topic}List存已经到期、等待消费的消息FIFOdq:dead:{topic}List重试次数耗尽的死信人工排查用中间那一步「从 ZSet 搬到 List」是整套设计的枢纽。为什么要多一个 List因为消费者需要阻塞等待而 ZSet 没有阻塞取。把到期消息先搬进 List消费端就能用BRPOP挂着等没消息时进程几乎不占 CPU比轮询 ZSet 省太多。提示score 一定用毫秒。用秒的话同一秒内到期的消息排序退化成字典序短延时比如 500 毫秒后执行会失去意义。2.2 三层结构生产者、调度器、消费进程整个链路是这样的生产者 push -- ZSet(delay) --[调度器扫描]-- List(ready) --[BRPOP]-- 消费进程 -- 业务处理 | -- 失败重投 ZSet / 进死信 List三个角色各自的职责生产者是业务代码里随手调的一个类负责组装消息体、算出到期时间戳、ZADD进 ZSet同步返回一个 jobId 给业务做幂等埋点。调度器是一个独立的常驻进程也挂在同一个 Swoole 进程池里固定占一个 worker每隔 100 毫秒左右扫一次 ZSet把到期的消息原子地搬进 ready List。它不执行业务逻辑只做搬运所以吞吐很高单进程足够撑住大部分场景。消费进程是多个 worker每个 worker 里跑一个BRPOP循环拿到消息就反序列化、调业务 handler、按结果决定 ACK 还是重投。worker 数量取决于你的业务耗时IO 密集型的调第三方接口、发短信可以开 816 个CPU 密集型的就少开点别把机器打满。这套分层最大的好处是调度和消费解耦。调度器挂了不影响已经在 ready List 里的消息被消费消费端慢了也只是 ready List 堆积可以通过 List 长度做监控告警。2.3 目录结构与进程模型项目目录我一般会这么组织方便后面二次开发时快速定位delay-queue/ ├── bin/ │ ├── server.php # 进程池入口拉起调度器和消费者 │ └── produce.php # 命令行投递脚本联调用 ├── src/ │ ├── Producer.php # 消息投递 │ ├── Scheduler.php # 到期扫描与搬运 │ ├── Worker.php # 消费循环 │ ├── Message.php # 消息体结构、序列化 │ ├── RetryPolicy.php # 退避策略 │ └── RedisFactory.php # 连接创建与断线重连 ├── lua/ │ └── move_expired.lua # 原子搬运脚本 ├── config/ │ └── queue.php # topic、并发数、重试参数 └── handler/ └── OrderCloseHandler.php # 业务处理器示例进程模型用的是Swoole\Process\Pool不是Swoole\Server。原因很实际这套东西不需要对外暴露端口不需要 TCP/HTTP 协议栈只是一个后台跑的任务集。Process\Pool 更轻管理进程数、重启子进程、转发信号这些事它都帮你做了代码量能少一半。进程分配上workerId 为 0 的那个进程固定跑调度器剩下的跑消费者。这样调度器天然单实例不用额外加分布式锁去防止重复扫描——当然就算你不小心开了两个调度器Lua 脚本里的ZREM返回值也能兜住后面会讲。3. 核心链路拆解从投递到出队的四段代码骨架清楚了接下来进代码。这四段是整个项目里最不能改错的顺序是投递 → 扫描搬运 → 消费 → 失败重投每一段都对应一个具体的坑。3.1 消息体设计与投递消息体不要只存业务 payload重试次数、最大重试、创建时间这些元数据必须一起带上不然后面做退避和死信判断时抓瞎。?php // src/Producer.php class Producer { private Redis $redis; private string $topic; public function __construct(Redis $redis, string $topic) { $this-redis $redis; $this-topic $topic; } /** * param array $payload 业务数据必须可 json 序列化 * param int $delayMs 延时毫秒数0 表示立刻到期 * param int $maxRetry 最大重试次数超过进死信 * return string jobId用于幂等与排查 */ public function push(array $payload, int $delayMs 0, int $maxRetry 3): string { $jobId bin2hex(random_bytes(8)); $message [ id $jobId, topic $this-topic, payload $payload, attempt 0, // 已经重试过几次 max_retry $maxRetry, born_at time(), // 第一次投递的秒级时间排查用 ]; // score 必须是毫秒时间戳保证同毫秒内也能正确排序 $score (int) round(microtime(true) * 1000) $delayMs; $this-redis-zAdd( dq:delay:{$this-topic}, $score, json_encode($message, JSON_UNESCAPED_UNICODE) ); return $jobId; } }这段代码有三个点值得说。第一id用 16 位随机十六进制不依赖自增多机部署时也不会撞。第二score用(int) round(microtime(true) * 1000)round是为了避开浮点乘法产生的.999截断问题不加的话偶尔会出现少 1 毫秒导致消息比预期早到期。第三整个消息体直接json_encode作为 ZSet 的 member好处是 ZSet 里每条消息自解释坏处是 message 大时内存占用高——如果你的 payload 超过 1KB建议改成 member 存 jobId、payload 另存一个 Hash这是二次开发时最常见的改造点。参数说明$delayMs传 0 表示立即到期适合做「异步化」而不需要真正延时的场景$maxRetry默认 3消费端会读这个字段决定什么时候进死信生产者可以按业务重要性单独指定。3.2 到期扫描与 Lua 原子搬运这一步是并发安全的重点。调度器用ZRANGEBYSCORE拿到期消息再用ZREM移除最后LPUSH进 ready List。如果这三步分开在 PHP 里做两个调度器实例比如滚动发布时新旧进程并存就可能同时读到同一条消息然后各ZREM一次、各LPUSH一次消息被消费两遍。解法是把三步塞进一个 Lua 脚本Redis 单线程执行 Lua天然原子-- lua/move_expired.lua -- KEYS[1] dq:delay:{topic} -- KEYS[2] dq:ready:{topic} -- ARGV[1] 当前毫秒时间戳 -- ARGV[2] 单批最大条数 local jobs redis.call(ZRANGEBYSCORE, KEYS[1], -inf, ARGV[1], LIMIT, 0, tonumber(ARGV[2])) if #jobs 0 then return 0 end local moved 0 for _, job in ipairs(jobs) do -- ZREM 返回 1 说明本次抢到了这条消息返回 0 说明已被别的进程移除 if redis.call(ZREM, KEYS[1], job) 1 then redis.call(LPUSH, KEYS[2], job) moved moved 1 end end return movedPHP 侧调用// src/Scheduler.php class Scheduler { private Redis $redis; private string $topic; private string $sha; public function __construct(Redis $redis, string $topic) { $this-redis $redis; $this-topic $topic; // 启动时一次性 load之后用 evalsha省带宽 $this-sha $redis-script(load, file_get_contents(__DIR__ . /../lua/move_expired.lua)); } public function tick(): int { $now (int) round(microtime(true) * 1000); try { return (int) $this-redis-evalSha($this-sha, [ dq:delay:{$this-topic}, dq:ready:{$this-topic}, $now, 200, // 单批最多搬 200 条防止大 key 拖垮 Redis ], 2); } catch (RedisException $e) { // NOSCRIPTRedis 重启或脚本被清空重新 load 一次 if (strpos($e-getMessage(), NOSCRIPT) ! false) { $this-sha $this-redis-script(load, file_get_contents(__DIR__ . /../lua/move_expired.lua)); } return 0; } } }ZREM返回值的判断是这段的核心。哪怕你有意无意开了两个调度器也只有ZREM 1的那个进程会把消息LPUSH进 ready List另一个进程拿到 0 直接跳过不会产生重复消息。这条血泪经验是从一次滚动发布事故里换来的——当时新旧两个调度进程并存了大概 40 秒一批关单任务被重复投递下游接口被打了两遍。参数说明ARGV[1]是当前毫秒时间戳由 PHP 传入而不是在 Lua 里取TIME因为 Lua 里调 Redis 的TIME命令多一次内部往返而且时间基准分散在两处不好排查。ARGV[2]是单批上限200 是个比较稳的值遇到大 key 时能把单次阻塞时间控制在毫秒级。调度器的主循环长这样100 毫秒一个 ticktick 之间用协程 sleep不占 CPU// 调度器主循环跑在 workerId 0 的进程里 while ($running) { $moved $scheduler-tick(); if ($moved 0) { // 没搬到东西就多歇一会儿避免空转刷 Redis \Swoole\Coroutine::sleep(0.1); } }3.3 消费、ACK 与失败重投消费端拿到消息后要做三件事反序列化、按 topic 找到 handler、根据结果决定 ACK 还是重投。这里最容易翻车的地方是「消息去哪了」——既没消费成功也没重投回去消息就这么凭空消失了。// src/Worker.php class Worker { private Redis $redis; private array $handlers; public function consume(string $topic): void { $delayKey dq:delay:{$topic}; $readyKey dq:ready:{$topic}; $deadKey dq:dead:{$topic}; while ($this-running) { // 阻塞 2 秒phpredis 的读超时必须大于这个值 $res $this-redis-brPop([$readyKey], 2); if (!$res) { continue; // 超时回头检查 running 标志 } [$key, $raw] $res; $message json_decode($raw, true); if (!is_array($message)) { // 脏数据直接进死信别让它无限循环 $this-redis-lPush($deadKey, $raw); continue; } $handler $this-handlers[$message[topic]] ?? null; if ($handler null) { $this-redis-lPush($deadKey, $raw); continue; } try { $handler($message[payload]); // 业务成功即认为消费成功Redis 侧无需额外 ACK } catch (\Throwable $e) { $this-retry($message, $delayKey, $deadKey, $e); } } } private function retry(array $message, string $delayKey, string $deadKey, \Throwable $e): void { $message[attempt]; $message[last_error] mb_substr($e-getMessage(), 0, 200); if ($message[attempt] $message[max_retry]) { $this-redis-lPush($deadKey, json_encode($message, JSON_UNESCAPED_UNICODE)); return; } // 指数退避2^n 秒封顶 60 秒避免下游刚崩就被重试打穿 $backoffMs min(60, 2 ** $message[attempt]) * 1000; $score (int) round(microtime(true) * 1000) $backoffMs; $this-redis-zAdd($delayKey, $score, json_encode($message, JSON_UNESCAPED_UNICODE)); } }这里采用的设计是「BRPOP 出队即视为投递成功」不额外做 ack 机制。原因是 List 的BRPOP本身就是原子弹出消息一旦出来就只有当前进程持有如果进程在处理中途崩了消息确实会丢。想做到「至少一次」语义常规改法是换用RPOPLPUSH到处理中的备份 List处理成功再LREM掉进程重启时先扫描备份 List 里的残留消息。这套改法我放在最后一章的二次开发里细讲。参数说明brPop第二个参数是阻塞秒数设 2 秒是为了让while ($this-running)能在收到退出信号后最多 2 秒内跳出循环配合优雅停机。max_retry从消息体里读而不是从配置里读这样不同来源的消息可以有不同的重试上限。退避公式min(60, 2 ** attempt)给出的是 2s、4s、8s、16s、32s、60s 这样的序列attempt从 1 开始算。为什么要封顶 60 秒下游服务挂掉恢复通常要几分钟重试间隔封顶后不会出现「第 3 次重试和第 4 次重试差了 8 分钟」的尴尬情况死信里也没那么多。4. 消费端工程化进程管理、信号退出与失败重试逻辑跑通只是及格线真上线还得解决三件事进程怎么拉起来、发布时怎么不丢消息、出问题怎么查。这三件事在本地开发时完全看不出来一上生产就集中爆发。4.1 Process\Pool 拉起多进程?php // bin/server.php require __DIR__ . /../vendor/autoload.php; $config require __DIR__ . /../config/queue.php; $workerNum $config[worker_num]; // 消费者进程数不含调度器 \Swoole\Runtime::enableCoroutine(SWOOLE_HOOK_ALL); $pool new Swoole\Process\Pool($workerNum 1, SWOOLE_IPC_NONE, 0, true); $pool-on(WorkerStart, function (Swoole\Process\Pool $pool, int $workerId) use ($config) { // 每个进程独立建连接绝不能跨进程共享 Redis 对象 $redis RedisFactory::make($config[redis]); if ($workerId 0) { // 0 号进程专职做调度 $scheduler new Scheduler($redis, $config[topic]); $scheduler-run(); return; } $worker new Worker($redis, $config[handlers]); $worker-consume($config[topic]); }); $pool-on(WorkerStop, function (Swoole\Process\Pool $pool, int $workerId) { // 只需要打日志连接释放交给进程退出 fwrite(STDERR, worker {$workerId} stopped\n); }); $pool-start();new Swoole\Process\Pool($workerNum 1, ...)里的1就是给调度器留的那个槽位。SWOOLE_IPC_NONE表示进程之间不建消息队列因为这套架构完全靠 Redis 通信进程间不需要 IPC省掉这块能少不少内存开销。第四个参数true是开启协程化让Swoole\Coroutine::sleep和 hook 后的 IO 能正常工作。这里最要命的一条Redis 连接绝对不能在WorkerStart外面 new 好了再传进去。fork出来的子进程会继承父进程的 socket 文件描述符多个进程读同一个 fd数据会串到别的进程去表现出来就是「消息被消费了但 handler 收到的参数是别人的」。这个坑我看过至少三次每次都是排查半天。4.2 信号处理与优雅停机发布时如果直接kill -9正在处理的消息就丢了。正确的姿势是接SIGTERM把running置为 false让循环自己走完当前这一条再退出。// 在 Worker::consume 之前注册 \Swoole\Process::signal(SIGTERM, function () { $this-running false; }); \Swoole\Process::signal(SIGINT, function () { $this-running false; });Swoole\Process::signal只能在WorkerStart回调里注册进程池启动前注册不生效。$this-running一开始是 true收到信号后置 falsewhile循环里下一次判断就跳出了。因为brPop阻塞了 2 秒最坏情况下进程会在 2 秒后才真正退出这 2 秒就是「还在处理中的消息」的保护窗口。配合 supervisor 的stopwaitsecs把等待时间设成 10 秒以上[program:delay-worker] commandphp /path/to/bin/server.php process_name%(program_name)s numprocs1 autostarttrue autorestarttrue stopasgrouptrue killasgrouptrue stopwaitsecs15stopasgrouptrue和killasgrouptrue很关键因为 Process\Pool 会再 fork 子进程supervisor 直接 stop 主进程的话子进程会变成孤儿进程继续跑下次启动就出现两批 worker 抢同一个 Redis List 的情况。4.3 监控、日志与连接管理没有监控的队列就是个黑匣子。最少要盯三个数指标取值方式告警阈值建议待消费堆积LLEN dq:ready:{topic}持续 5 分钟 1000延时未到期量ZCARD dq:delay:{topic}突增 10 倍生产者异常刷死信量LLEN dq:dead:{topic}有任何增长就通知日志方面每条消息处理完打一行结构化日志字段至少带jobId、topic、attempt、cost_ms、status。jobId是生产者返回给业务的那个用它能在全链路日志里串起来排查「这条消息到底被处理了几次」的时候特别管用。连接管理上每个 worker 一个 Redis 连接但要处理断线重连// src/RedisFactory.php class RedisFactory { public static function make(array $conf): Redis { $redis new Redis(); $redis-connect($conf[host], $conf[port], 2.0); if (!empty($conf[password])) { $redis-auth($conf[password]); } $redis-select($conf[db] ?? 0); // 读超时必须大于 brPop 的阻塞时长否则会抛 RedisException $redis-setOption(Redis::OPT_READ_TIMEOUT, $conf[read_timeout] ?? 5.0); $redis-setOption(Redis::OPT_SERIALIZER, Redis::SERIALIZER_NONE); return $redis; } }OPT_READ_TIMEOUT必须大于brPop的阻塞秒数这是 phpredis 的一个老坑默认读超时比较小brPop阻塞 2 秒就会抛read error on connection进程直接挂掉supervisor 又给你拉起来看起来像是「队列一直在重启」。5. 落地排查启动、并发与数据一致性上的 5 个坑前四章讲的是设计这一章讲的是我实际踩过的。每一条都是「现象 → 原因 → 解决」的完整链条遇到同类问题时可以直接对号入座。5.1 坑一worker 一启动就反复重启现象supervisor 里进程状态不停在RUNNING和BACKOFF之间跳日志里刷read error on connection。原因brPop的阻塞时长超过了 phpredis 的读超时。默认配置里读超时可能只有 1 秒而brPop([key], 2)要阻塞 2 秒客户端等不到数据就认为连接断了。解决把Redis::OPT_READ_TIMEOUT显式设成阻塞时长的 2 倍以上比如brPop阻塞 2 秒就把读超时设成 5 秒。两边参数不要靠猜写在配置文件里一起改。5.2 坑二同一条消息被消费了两次现象订单关单日志里同一个订单出现两次关闭记录或者短信接口被调了两遍。原因滚动发布时新旧调度进程短暂并存两边的 PHP 代码各自ZRANGEBYSCORE拿到同一批消息然后各自ZREMLPUSH。解决把「读、删、推」三步写进 Lua 脚本靠ZREM返回值 1 或 0 判断归属。哪怕有十个调度器同时跑也只有第一个ZREM成功的进程会往下走。如果业务侧还需要更强的幂等在 handler 里再按jobId做一次去重比如 Setnx 一个 5 分钟过期的 key。5.3 坑三消息处理到一半进程没了消息也丢了现象某些订单该关单的没关查日志发现消息被BRPOP出来之后就没有下文了。原因BRPOP是出队即删除语义。消息从 List 弹出来之后如果进程在处理中途被kill -9或者机器重启这条消息在 Redis 里已经不存在了。解决换成RPOPLPUSH或BLMOVE把消息推进一个dq:processing:{topic}的备份 List处理成功再LREM删除备份。进程启动时先扫一遍 processing List把遗留消息重新LPUSH回 ready List。这是从「至多一次」提升到「至少一次」的标准做法代价是多一次 Redis 往返。5.4 坑四延时消息提前执行了现象设置 30 分钟延时的消息几分钟后就被消费了。原因score 用了秒级时间戳或者用了time()而不是毫秒。秒级时间戳下同一秒内到期的消息排序不稳定而调度器又可能因为多次 tick 反复扫到边界值。解决统一用(int) round(microtime(true) * 1000)并且在调度器的 Lua 脚本里用而不是。生产者和消费者两地的时间基准必须一致服务器时间不同步的话NTP 没配会整体偏移这时候要看date命令输出的时区和偏移量。5.5 坑五死信队列越堆越多但没人发现现象LLEN dq:dead:{topic}从 0 涨到几万业务方反馈「有一批用户没收到通知」。原因死信只是个 List没有任何告警。消费失败进死信之后除了日志里有一行 error没有任何人会主动去看。解决死信数量纳入监控任何增长都触发告警同时死信 List 设一个上限比如 10000 条超过之后RPOP掉最老的并单独打一条「死信溢出」日志避免把 Redis 内存撑爆。更好的做法是死信单独进一个可以手工重放的入口运维能按topic和时间段筛选重投。6. 二次开发从「能用」到「敢用」的三个改造方向代码能跑起来之后接下来考虑的是这套东西怎么撑住更复杂的业务。我一般会按「先补一致性、再加优先级、最后做压测」的顺序改每次只动一个维度改完跑一轮回归避免多个变更叠在一起出问题时定位不了。第一个改造是补上「至少一次」语义。前面提过把BRPOP换成BLMOVE到 processing List处理成功再删。改造后 worker 启动时需要一段补偿逻辑// 进程启动时的遗留消息补偿放在 consume 循环之前 private function recover(string $topic): int { $processingKey dq:processing:{$topic}; $readyKey dq:ready:{$topic}; $count 0; // 每次最多补偿 500 条防止启动瞬间把 ready List 打爆 while ($count 500 ($raw $this-redis-rPopLPush($processingKey, $readyKey))) { $count; } return $count; }逻辑上就是把 processing List 里残留的消息原样推回 ready 等待重新消费。注意这里用rPopLPush循环而不是一次性LRANGE全量搬因为启动时 List 可能有几十万条一次性搬会长时间阻塞 Redis。500 条一批剩下的让消费循环自然消化。第二个改造是加优先级。现在的实现所有 topic 共用一个 ready List先进先出。如果业务里存在「实时性要求高的消息」可以按优先级拆成dq:ready:high和dq:ready:low两个 List消费者用brPop([high, low], 2)。phpredis 的brPop支持多个 key会按参数顺序优先弹出天然就是优先级队列不用额外写代码。第三个改造是压测。别用「感觉挺快」来判断吞吐。写一个压测脚本投递 10 万条 0 延时的消息观察 ready List 从峰值降到 0 的耗时以及 worker 的 CPU 占用# 投递 10 万条立即到期的消息 php bin/produce.php --topicorder_close --count100000 --delay0 # 每秒采样一次 ready List 长度直到归零 while true; do len$(redis-cli LLEN dq:ready:order_close) echo $(date %s) $len [ $len -eq 0 ] break sleep 1 done根据这个曲线能算出实际 TPS再乘上单个 worker 的平均处理耗时反推出需要几个 worker。如果发现 ready List 一直降不下来、worker CPU 又没跑满八成是 Redis 连接被阻塞在别处了回去看第 5 章的坑一。最后一个习惯每次改 Lua 脚本或者换 Redis 客户端配置我都会先用redis-cli --latency看一眼延迟再跑一轮上面的压测脚本。有一次改完ZRANGEBYSCORE的批次大小从 200 调到 2000单机测试毫无问题上了预发之后 Redis 延迟从 0.3ms 直接跳到 40ms整条链路的消费速度反而慢了。从那以后我每次动这几个参数都强制走一遍压测加延迟对比宁可多花半天也不想在半夜被叫起来看监控。希望这套思路能帮你少走几个弯路。本文还有配套的精品资源点击获取 
长河君 · 2026/11/6 5:33:01 · 阅读 1.2k Redis ZSet与Swoole多进程延时消息队列实践
简介基于 Redis 延时队列与 Swoole 多进程模型实现的消息队列消费端面向已具备 PHP 基础、希望深入理解高并发消费架构的中高级开发者。它针对的问题是如何在保证延时精度的前提下让多个消费者并行、稳定地从队列中取件并处理。实现思路上以 Redis 有序集合承载待投递消息用时间戳作为 score到期后通过 ZRANGEBYSCORE 批量取出消费侧依托 Swoole Server 的多进程能力让每个子进程独立拉取与处理消息形成可横向扩展的消费集群同时借助非阻塞 I/O 降低空转开销。压缩包约 1.66MB主体为该消费服务的 PHP 工程源码涵盖 Redis 队列读写、多进程消费配置、消息处理与异常兜底等模块结构紧凑便于直接对照阅读与二次改造。已有 1270 人学习。1. 订单超时关单引出的一套消费端订单下单后 30 分钟未支付就要自动关单这是延时队列最常见的落地场景。很多团队第一反应是挂一个 crontab 每分钟扫一次数据库量小的时候没问题单表过了百万行之后扫描延迟、锁竞争、重复执行这些玄学问题就会一个接一个冒出来。这套 Redis 延时消息队列换了个思路用 ZSet 的 score 存到期时间戳用 Swoole 拉起多个常驻进程做消费端消息只有真正到期了才进入消费链路。它适合已经跑着 PHP 7.4 以上、Redis 有余量、又不想为了一个延时任务单独引入一套消息中间件的团队订单关单、支付超时提醒、失败重试、定时推送这类场景都能直接套上去。下面从架构一直拆到能跑的代码。2. 架构与进程模型Redis ZSet 加 Swoole 多进程怎么分工先把整套东西的骨架摆出来不然后面贴代码容易看散。这个项目的核心其实就是两个 Redis 结构加三类进程结构简单到你可以在一张纸上画完但每个角色的职责边界划得很清楚改起来不容易牵一发而动全身。2.1 为什么延时用 ZSet 而不是 ListList 只能做先进先出做不了「按时间到期」这件事。有人会用LPUSHBRPOP再配合sleep来模拟延时问题是 sleep 的精度和并发一塌糊涂多个消费者各自 sleep 不同时长消息顺序直接乱掉。ZSet 的 score 天然就是排序字段把「到期执行时间」写成毫秒时间戳塞进 score取消息时只取score 当前毫秒时间戳的成员语义完全对得上。常见做法是把延时队列拆成两个结构结构类型作用dq:delay:{topic}ZSet存未到期的消息score 是到期毫秒时间戳dq:ready:{topic}List存已经到期、等待消费的消息FIFOdq:dead:{topic}List重试次数耗尽的死信人工排查用中间那一步「从 ZSet 搬到 List」是整套设计的枢纽。为什么要多一个 List因为消费者需要阻塞等待而 ZSet 没有阻塞取。把到期消息先搬进 List消费端就能用BRPOP挂着等没消息时进程几乎不占 CPU比轮询 ZSet 省太多。提示score 一定用毫秒。用秒的话同一秒内到期的消息排序退化成字典序短延时比如 500 毫秒后执行会失去意义。2.2 三层结构生产者、调度器、消费进程整个链路是这样的生产者 push -- ZSet(delay) --[调度器扫描]-- List(ready) --[BRPOP]-- 消费进程 -- 业务处理 | -- 失败重投 ZSet / 进死信 List三个角色各自的职责生产者是业务代码里随手调的一个类负责组装消息体、算出到期时间戳、ZADD进 ZSet同步返回一个 jobId 给业务做幂等埋点。调度器是一个独立的常驻进程也挂在同一个 Swoole 进程池里固定占一个 worker每隔 100 毫秒左右扫一次 ZSet把到期的消息原子地搬进 ready List。它不执行业务逻辑只做搬运所以吞吐很高单进程足够撑住大部分场景。消费进程是多个 worker每个 worker 里跑一个BRPOP循环拿到消息就反序列化、调业务 handler、按结果决定 ACK 还是重投。worker 数量取决于你的业务耗时IO 密集型的调第三方接口、发短信可以开 816 个CPU 密集型的就少开点别把机器打满。这套分层最大的好处是调度和消费解耦。调度器挂了不影响已经在 ready List 里的消息被消费消费端慢了也只是 ready List 堆积可以通过 List 长度做监控告警。2.3 目录结构与进程模型项目目录我一般会这么组织方便后面二次开发时快速定位delay-queue/ ├── bin/ │ ├── server.php # 进程池入口拉起调度器和消费者 │ └── produce.php # 命令行投递脚本联调用 ├── src/ │ ├── Producer.php # 消息投递 │ ├── Scheduler.php # 到期扫描与搬运 │ ├── Worker.php # 消费循环 │ ├── Message.php # 消息体结构、序列化 │ ├── RetryPolicy.php # 退避策略 │ └── RedisFactory.php # 连接创建与断线重连 ├── lua/ │ └── move_expired.lua # 原子搬运脚本 ├── config/ │ └── queue.php # topic、并发数、重试参数 └── handler/ └── OrderCloseHandler.php # 业务处理器示例进程模型用的是Swoole\Process\Pool不是Swoole\Server。原因很实际这套东西不需要对外暴露端口不需要 TCP/HTTP 协议栈只是一个后台跑的任务集。Process\Pool 更轻管理进程数、重启子进程、转发信号这些事它都帮你做了代码量能少一半。进程分配上workerId 为 0 的那个进程固定跑调度器剩下的跑消费者。这样调度器天然单实例不用额外加分布式锁去防止重复扫描——当然就算你不小心开了两个调度器Lua 脚本里的ZREM返回值也能兜住后面会讲。3. 核心链路拆解从投递到出队的四段代码骨架清楚了接下来进代码。这四段是整个项目里最不能改错的顺序是投递 → 扫描搬运 → 消费 → 失败重投每一段都对应一个具体的坑。3.1 消息体设计与投递消息体不要只存业务 payload重试次数、最大重试、创建时间这些元数据必须一起带上不然后面做退避和死信判断时抓瞎。?php // src/Producer.php class Producer { private Redis $redis; private string $topic; public function __construct(Redis $redis, string $topic) { $this-redis $redis; $this-topic $topic; } /** * param array $payload 业务数据必须可 json 序列化 * param int $delayMs 延时毫秒数0 表示立刻到期 * param int $maxRetry 最大重试次数超过进死信 * return string jobId用于幂等与排查 */ public function push(array $payload, int $delayMs 0, int $maxRetry 3): string { $jobId bin2hex(random_bytes(8)); $message [ id $jobId, topic $this-topic, payload $payload, attempt 0, // 已经重试过几次 max_retry $maxRetry, born_at time(), // 第一次投递的秒级时间排查用 ]; // score 必须是毫秒时间戳保证同毫秒内也能正确排序 $score (int) round(microtime(true) * 1000) $delayMs; $this-redis-zAdd( dq:delay:{$this-topic}, $score, json_encode($message, JSON_UNESCAPED_UNICODE) ); return $jobId; } }这段代码有三个点值得说。第一id用 16 位随机十六进制不依赖自增多机部署时也不会撞。第二score用(int) round(microtime(true) * 1000)round是为了避开浮点乘法产生的.999截断问题不加的话偶尔会出现少 1 毫秒导致消息比预期早到期。第三整个消息体直接json_encode作为 ZSet 的 member好处是 ZSet 里每条消息自解释坏处是 message 大时内存占用高——如果你的 payload 超过 1KB建议改成 member 存 jobId、payload 另存一个 Hash这是二次开发时最常见的改造点。参数说明$delayMs传 0 表示立即到期适合做「异步化」而不需要真正延时的场景$maxRetry默认 3消费端会读这个字段决定什么时候进死信生产者可以按业务重要性单独指定。3.2 到期扫描与 Lua 原子搬运这一步是并发安全的重点。调度器用ZRANGEBYSCORE拿到期消息再用ZREM移除最后LPUSH进 ready List。如果这三步分开在 PHP 里做两个调度器实例比如滚动发布时新旧进程并存就可能同时读到同一条消息然后各ZREM一次、各LPUSH一次消息被消费两遍。解法是把三步塞进一个 Lua 脚本Redis 单线程执行 Lua天然原子-- lua/move_expired.lua -- KEYS[1] dq:delay:{topic} -- KEYS[2] dq:ready:{topic} -- ARGV[1] 当前毫秒时间戳 -- ARGV[2] 单批最大条数 local jobs redis.call(ZRANGEBYSCORE, KEYS[1], -inf, ARGV[1], LIMIT, 0, tonumber(ARGV[2])) if #jobs 0 then return 0 end local moved 0 for _, job in ipairs(jobs) do -- ZREM 返回 1 说明本次抢到了这条消息返回 0 说明已被别的进程移除 if redis.call(ZREM, KEYS[1], job) 1 then redis.call(LPUSH, KEYS[2], job) moved moved 1 end end return movedPHP 侧调用// src/Scheduler.php class Scheduler { private Redis $redis; private string $topic; private string $sha; public function __construct(Redis $redis, string $topic) { $this-redis $redis; $this-topic $topic; // 启动时一次性 load之后用 evalsha省带宽 $this-sha $redis-script(load, file_get_contents(__DIR__ . /../lua/move_expired.lua)); } public function tick(): int { $now (int) round(microtime(true) * 1000); try { return (int) $this-redis-evalSha($this-sha, [ dq:delay:{$this-topic}, dq:ready:{$this-topic}, $now, 200, // 单批最多搬 200 条防止大 key 拖垮 Redis ], 2); } catch (RedisException $e) { // NOSCRIPTRedis 重启或脚本被清空重新 load 一次 if (strpos($e-getMessage(), NOSCRIPT) ! false) { $this-sha $this-redis-script(load, file_get_contents(__DIR__ . /../lua/move_expired.lua)); } return 0; } } }ZREM返回值的判断是这段的核心。哪怕你有意无意开了两个调度器也只有ZREM 1的那个进程会把消息LPUSH进 ready List另一个进程拿到 0 直接跳过不会产生重复消息。这条血泪经验是从一次滚动发布事故里换来的——当时新旧两个调度进程并存了大概 40 秒一批关单任务被重复投递下游接口被打了两遍。参数说明ARGV[1]是当前毫秒时间戳由 PHP 传入而不是在 Lua 里取TIME因为 Lua 里调 Redis 的TIME命令多一次内部往返而且时间基准分散在两处不好排查。ARGV[2]是单批上限200 是个比较稳的值遇到大 key 时能把单次阻塞时间控制在毫秒级。调度器的主循环长这样100 毫秒一个 ticktick 之间用协程 sleep不占 CPU// 调度器主循环跑在 workerId 0 的进程里 while ($running) { $moved $scheduler-tick(); if ($moved 0) { // 没搬到东西就多歇一会儿避免空转刷 Redis \Swoole\Coroutine::sleep(0.1); } }3.3 消费、ACK 与失败重投消费端拿到消息后要做三件事反序列化、按 topic 找到 handler、根据结果决定 ACK 还是重投。这里最容易翻车的地方是「消息去哪了」——既没消费成功也没重投回去消息就这么凭空消失了。// src/Worker.php class Worker { private Redis $redis; private array $handlers; public function consume(string $topic): void { $delayKey dq:delay:{$topic}; $readyKey dq:ready:{$topic}; $deadKey dq:dead:{$topic}; while ($this-running) { // 阻塞 2 秒phpredis 的读超时必须大于这个值 $res $this-redis-brPop([$readyKey], 2); if (!$res) { continue; // 超时回头检查 running 标志 } [$key, $raw] $res; $message json_decode($raw, true); if (!is_array($message)) { // 脏数据直接进死信别让它无限循环 $this-redis-lPush($deadKey, $raw); continue; } $handler $this-handlers[$message[topic]] ?? null; if ($handler null) { $this-redis-lPush($deadKey, $raw); continue; } try { $handler($message[payload]); // 业务成功即认为消费成功Redis 侧无需额外 ACK } catch (\Throwable $e) { $this-retry($message, $delayKey, $deadKey, $e); } } } private function retry(array $message, string $delayKey, string $deadKey, \Throwable $e): void { $message[attempt]; $message[last_error] mb_substr($e-getMessage(), 0, 200); if ($message[attempt] $message[max_retry]) { $this-redis-lPush($deadKey, json_encode($message, JSON_UNESCAPED_UNICODE)); return; } // 指数退避2^n 秒封顶 60 秒避免下游刚崩就被重试打穿 $backoffMs min(60, 2 ** $message[attempt]) * 1000; $score (int) round(microtime(true) * 1000) $backoffMs; $this-redis-zAdd($delayKey, $score, json_encode($message, JSON_UNESCAPED_UNICODE)); } }这里采用的设计是「BRPOP 出队即视为投递成功」不额外做 ack 机制。原因是 List 的BRPOP本身就是原子弹出消息一旦出来就只有当前进程持有如果进程在处理中途崩了消息确实会丢。想做到「至少一次」语义常规改法是换用RPOPLPUSH到处理中的备份 List处理成功再LREM掉进程重启时先扫描备份 List 里的残留消息。这套改法我放在最后一章的二次开发里细讲。参数说明brPop第二个参数是阻塞秒数设 2 秒是为了让while ($this-running)能在收到退出信号后最多 2 秒内跳出循环配合优雅停机。max_retry从消息体里读而不是从配置里读这样不同来源的消息可以有不同的重试上限。退避公式min(60, 2 ** attempt)给出的是 2s、4s、8s、16s、32s、60s 这样的序列attempt从 1 开始算。为什么要封顶 60 秒下游服务挂掉恢复通常要几分钟重试间隔封顶后不会出现「第 3 次重试和第 4 次重试差了 8 分钟」的尴尬情况死信里也没那么多。4. 消费端工程化进程管理、信号退出与失败重试逻辑跑通只是及格线真上线还得解决三件事进程怎么拉起来、发布时怎么不丢消息、出问题怎么查。这三件事在本地开发时完全看不出来一上生产就集中爆发。4.1 Process\Pool 拉起多进程?php // bin/server.php require __DIR__ . /../vendor/autoload.php; $config require __DIR__ . /../config/queue.php; $workerNum $config[worker_num]; // 消费者进程数不含调度器 \Swoole\Runtime::enableCoroutine(SWOOLE_HOOK_ALL); $pool new Swoole\Process\Pool($workerNum 1, SWOOLE_IPC_NONE, 0, true); $pool-on(WorkerStart, function (Swoole\Process\Pool $pool, int $workerId) use ($config) { // 每个进程独立建连接绝不能跨进程共享 Redis 对象 $redis RedisFactory::make($config[redis]); if ($workerId 0) { // 0 号进程专职做调度 $scheduler new Scheduler($redis, $config[topic]); $scheduler-run(); return; } $worker new Worker($redis, $config[handlers]); $worker-consume($config[topic]); }); $pool-on(WorkerStop, function (Swoole\Process\Pool $pool, int $workerId) { // 只需要打日志连接释放交给进程退出 fwrite(STDERR, worker {$workerId} stopped\n); }); $pool-start();new Swoole\Process\Pool($workerNum 1, ...)里的1就是给调度器留的那个槽位。SWOOLE_IPC_NONE表示进程之间不建消息队列因为这套架构完全靠 Redis 通信进程间不需要 IPC省掉这块能少不少内存开销。第四个参数true是开启协程化让Swoole\Coroutine::sleep和 hook 后的 IO 能正常工作。这里最要命的一条Redis 连接绝对不能在WorkerStart外面 new 好了再传进去。fork出来的子进程会继承父进程的 socket 文件描述符多个进程读同一个 fd数据会串到别的进程去表现出来就是「消息被消费了但 handler 收到的参数是别人的」。这个坑我看过至少三次每次都是排查半天。4.2 信号处理与优雅停机发布时如果直接kill -9正在处理的消息就丢了。正确的姿势是接SIGTERM把running置为 false让循环自己走完当前这一条再退出。// 在 Worker::consume 之前注册 \Swoole\Process::signal(SIGTERM, function () { $this-running false; }); \Swoole\Process::signal(SIGINT, function () { $this-running false; });Swoole\Process::signal只能在WorkerStart回调里注册进程池启动前注册不生效。$this-running一开始是 true收到信号后置 falsewhile循环里下一次判断就跳出了。因为brPop阻塞了 2 秒最坏情况下进程会在 2 秒后才真正退出这 2 秒就是「还在处理中的消息」的保护窗口。配合 supervisor 的stopwaitsecs把等待时间设成 10 秒以上[program:delay-worker] commandphp /path/to/bin/server.php process_name%(program_name)s numprocs1 autostarttrue autorestarttrue stopasgrouptrue killasgrouptrue stopwaitsecs15stopasgrouptrue和killasgrouptrue很关键因为 Process\Pool 会再 fork 子进程supervisor 直接 stop 主进程的话子进程会变成孤儿进程继续跑下次启动就出现两批 worker 抢同一个 Redis List 的情况。4.3 监控、日志与连接管理没有监控的队列就是个黑匣子。最少要盯三个数指标取值方式告警阈值建议待消费堆积LLEN dq:ready:{topic}持续 5 分钟 1000延时未到期量ZCARD dq:delay:{topic}突增 10 倍生产者异常刷死信量LLEN dq:dead:{topic}有任何增长就通知日志方面每条消息处理完打一行结构化日志字段至少带jobId、topic、attempt、cost_ms、status。jobId是生产者返回给业务的那个用它能在全链路日志里串起来排查「这条消息到底被处理了几次」的时候特别管用。连接管理上每个 worker 一个 Redis 连接但要处理断线重连// src/RedisFactory.php class RedisFactory { public static function make(array $conf): Redis { $redis new Redis(); $redis-connect($conf[host], $conf[port], 2.0); if (!empty($conf[password])) { $redis-auth($conf[password]); } $redis-select($conf[db] ?? 0); // 读超时必须大于 brPop 的阻塞时长否则会抛 RedisException $redis-setOption(Redis::OPT_READ_TIMEOUT, $conf[read_timeout] ?? 5.0); $redis-setOption(Redis::OPT_SERIALIZER, Redis::SERIALIZER_NONE); return $redis; } }OPT_READ_TIMEOUT必须大于brPop的阻塞秒数这是 phpredis 的一个老坑默认读超时比较小brPop阻塞 2 秒就会抛read error on connection进程直接挂掉supervisor 又给你拉起来看起来像是「队列一直在重启」。5. 落地排查启动、并发与数据一致性上的 5 个坑前四章讲的是设计这一章讲的是我实际踩过的。每一条都是「现象 → 原因 → 解决」的完整链条遇到同类问题时可以直接对号入座。5.1 坑一worker 一启动就反复重启现象supervisor 里进程状态不停在RUNNING和BACKOFF之间跳日志里刷read error on connection。原因brPop的阻塞时长超过了 phpredis 的读超时。默认配置里读超时可能只有 1 秒而brPop([key], 2)要阻塞 2 秒客户端等不到数据就认为连接断了。解决把Redis::OPT_READ_TIMEOUT显式设成阻塞时长的 2 倍以上比如brPop阻塞 2 秒就把读超时设成 5 秒。两边参数不要靠猜写在配置文件里一起改。5.2 坑二同一条消息被消费了两次现象订单关单日志里同一个订单出现两次关闭记录或者短信接口被调了两遍。原因滚动发布时新旧调度进程短暂并存两边的 PHP 代码各自ZRANGEBYSCORE拿到同一批消息然后各自ZREMLPUSH。解决把「读、删、推」三步写进 Lua 脚本靠ZREM返回值 1 或 0 判断归属。哪怕有十个调度器同时跑也只有第一个ZREM成功的进程会往下走。如果业务侧还需要更强的幂等在 handler 里再按jobId做一次去重比如 Setnx 一个 5 分钟过期的 key。5.3 坑三消息处理到一半进程没了消息也丢了现象某些订单该关单的没关查日志发现消息被BRPOP出来之后就没有下文了。原因BRPOP是出队即删除语义。消息从 List 弹出来之后如果进程在处理中途被kill -9或者机器重启这条消息在 Redis 里已经不存在了。解决换成RPOPLPUSH或BLMOVE把消息推进一个dq:processing:{topic}的备份 List处理成功再LREM删除备份。进程启动时先扫一遍 processing List把遗留消息重新LPUSH回 ready List。这是从「至多一次」提升到「至少一次」的标准做法代价是多一次 Redis 往返。5.4 坑四延时消息提前执行了现象设置 30 分钟延时的消息几分钟后就被消费了。原因score 用了秒级时间戳或者用了time()而不是毫秒。秒级时间戳下同一秒内到期的消息排序不稳定而调度器又可能因为多次 tick 反复扫到边界值。解决统一用(int) round(microtime(true) * 1000)并且在调度器的 Lua 脚本里用而不是。生产者和消费者两地的时间基准必须一致服务器时间不同步的话NTP 没配会整体偏移这时候要看date命令输出的时区和偏移量。5.5 坑五死信队列越堆越多但没人发现现象LLEN dq:dead:{topic}从 0 涨到几万业务方反馈「有一批用户没收到通知」。原因死信只是个 List没有任何告警。消费失败进死信之后除了日志里有一行 error没有任何人会主动去看。解决死信数量纳入监控任何增长都触发告警同时死信 List 设一个上限比如 10000 条超过之后RPOP掉最老的并单独打一条「死信溢出」日志避免把 Redis 内存撑爆。更好的做法是死信单独进一个可以手工重放的入口运维能按topic和时间段筛选重投。6. 二次开发从「能用」到「敢用」的三个改造方向代码能跑起来之后接下来考虑的是这套东西怎么撑住更复杂的业务。我一般会按「先补一致性、再加优先级、最后做压测」的顺序改每次只动一个维度改完跑一轮回归避免多个变更叠在一起出问题时定位不了。第一个改造是补上「至少一次」语义。前面提过把BRPOP换成BLMOVE到 processing List处理成功再删。改造后 worker 启动时需要一段补偿逻辑// 进程启动时的遗留消息补偿放在 consume 循环之前 private function recover(string $topic): int { $processingKey dq:processing:{$topic}; $readyKey dq:ready:{$topic}; $count 0; // 每次最多补偿 500 条防止启动瞬间把 ready List 打爆 while ($count 500 ($raw $this-redis-rPopLPush($processingKey, $readyKey))) { $count; } return $count; }逻辑上就是把 processing List 里残留的消息原样推回 ready 等待重新消费。注意这里用rPopLPush循环而不是一次性LRANGE全量搬因为启动时 List 可能有几十万条一次性搬会长时间阻塞 Redis。500 条一批剩下的让消费循环自然消化。第二个改造是加优先级。现在的实现所有 topic 共用一个 ready List先进先出。如果业务里存在「实时性要求高的消息」可以按优先级拆成dq:ready:high和dq:ready:low两个 List消费者用brPop([high, low], 2)。phpredis 的brPop支持多个 key会按参数顺序优先弹出天然就是优先级队列不用额外写代码。第三个改造是压测。别用「感觉挺快」来判断吞吐。写一个压测脚本投递 10 万条 0 延时的消息观察 ready List 从峰值降到 0 的耗时以及 worker 的 CPU 占用# 投递 10 万条立即到期的消息 php bin/produce.php --topicorder_close --count100000 --delay0 # 每秒采样一次 ready List 长度直到归零 while true; do len$(redis-cli LLEN dq:ready:order_close) echo $(date %s) $len [ $len -eq 0 ] break sleep 1 done根据这个曲线能算出实际 TPS再乘上单个 worker 的平均处理耗时反推出需要几个 worker。如果发现 ready List 一直降不下来、worker CPU 又没跑满八成是 Redis 连接被阻塞在别处了回去看第 5 章的坑一。最后一个习惯每次改 Lua 脚本或者换 Redis 客户端配置我都会先用redis-cli --latency看一眼延迟再跑一轮上面的压测脚本。有一次改完ZRANGEBYSCORE的批次大小从 200 调到 2000单机测试毫无问题上了预发之后 Redis 延迟从 0.3ms 直接跳到 40ms整条链路的消费速度反而慢了。从那以后我每次动这几个参数都强制走一遍压测加延迟对比宁可多花半天也不想在半夜被叫起来看监控。希望这套思路能帮你少走几个弯路。本文还有配套的精品资源点击获取