Redis延时消息队列与Swoole多进程消费端：从扫表到可靠执行的工程实践

简介：面向PHP开发者的Redis延时消息队列实现方案，基于Swoole多进程模型构建高并发消费端，解决定时任务与延迟消息处理中的吞吐瓶颈。资源来自Serve项目源码包（1.66MB），聚焦Redis有序集合（ZSet）的延时队列设计，以及Swoole Server多进程并行消费机制，适合需要掌握异步任务调度、队列削峰填谷的中高级开发者参考。该设计可应用于订单超时关闭、定时重试、异步通知等对时效要求较高的业务场景，从中可直观理解消息如何按score排序并触达消费者。压缩包约1.66MB，解压后可查看Redis交互、Swoole进程配置等核心代码。目前已有1269人学习，说明该方案具有一定实践参考价值。通过学习可理解如何利用ZRANGEBYSCORE命令获取到期消息，如何分配子进程提升消费能力，以及非阻塞I/O在PHP服务中的实际应用，为自建延时队列或优化现有消息中间件提供直接可借鉴的代码范本。

1. 现金券过期提醒的夹生饭：我为什么用Redis延时消息队列接住Swoole多进程消费端

某次做一个活动现金券的“过期前5分钟提醒”，最开始图省事用了crontab每分钟扫表。结果活动还没上线，测试环境压了一遍就发现问题：订单表倒是没多大，但券表一旦关联用户维度，查询一下就拖慢了主库的几个核心接口，而且扫表SQL在主从延迟时会把刚过期的数据漏掉。换方案那几天我翻了一圈，最后落到这份基于Redis有序集合的延时消息队列 + Swoole多进程消费端上。它解决的核心问题很直接：消息带一个未来时间戳存进Redis，消费端不扫数据库，只从zset里按分数范围拉取“到点”的消息，由Swoole拉起多个Worker进程并发处理，任务完成后从集合里移除。如果你正在做订单超时关闭、支付回调延时通知、券过期提醒这类需求，又不想引入重型MQ，这份资源是能直接落到业务里的参考实现。

2. 选型与整体架构：同样是延时，crontab扫表为什么会被Swoole消费端替换掉

2.1 先掂量一下：crontab扫库方案卡在哪

大部分PHP项目里的定时任务都是从crontab开始的。写一个脚本，每分钟执行一次，查询“到期时间在五分钟内”的数据，然后逐条处理。这个模式在数据量几百上千时没问题，但一旦业务量上来，痛点会一个一个浮出来。

第一个痛点是主从延迟。线上通常读写分离，扫表脚本如果连的是从库，主库刚写入的券记录还没同步到从库，这轮扫描就把它漏掉了。你以为延时任务被正确执行，实际上它可能被延迟到下一分钟，或者直接错过。第二个痛点是SQL的查询范围不受控。要扫到期数据，SQL必然写成“到期时间大于某个值且小于某个值”，这种条件如果在时间字段上没有合适索引，就是全表扫描。到了整点、凌晨这种集中到期的时间点，脚本可能一次查出几万条数据，内存和CPU都扛不住，还会拖累正常业务请求。

第三个痛点是没有重试机制。扫表脚本通常是一条数据处理失败就记日志，下轮扫描根本不会再处理这条。用户券没提醒到，运营那边还要做补偿，这是典型的“埋了雷”的写法。我当时还把这事儿写成了评审问题：扫表方案的精确定时、失败重试、消费速度可控，三个能力基本都没有，只有“每分钟跑一次”的蛮力。

2.2 方案对比：为什么是Redis zset而不是RabbitMQ延迟队列

先做了一张对比表，帮助当时组内定方案时快速对齐：

RabbitMQ的延迟队列实现其实很成熟，但前提是团队愿意多维护一套MQ基础设施。我所在的某项目组当时Redis已经承担了缓存和分布式锁，再让它承担延时消息队列，运维成本几乎为零。Redis的有序集合天然按score排序，score放时间戳，那zset里key最小的就是最先该被消费的消息。用ZRANGEBYSCORE按分数范围取数据，取出来做业务，最后ZREM移除，这个读写模型非常贴合延时队列。

选择Swoole多进程消费端的原因也简单。如果只是用CLI脚本跑while循环，单进程消费几十万条到期消息时处理速度不够，而且脚本一挂就没人拉起。Swoole的swoole_process可以让一个Master进程管理多个Worker子进程，子进程挂了能自动重拉，还能接管信号做优雅退出。它是PHP常驻内存运行的基础设施，而不是简单的一次性CLI。这里要特别说明一点：资源里的“多进程消费端”不是用pcntl_fork手工管理进程，而是用swoole_process封装好的进程模型，Master只负责调度和监控，真正的消息分发逻辑在Worker里。

2.3 进程模型：Master不干活，Worker只消费到点消息

整体数据流是这样：业务端调用生产接口，把消息ID、业务类型、回调参数、到期时间统一写入zset，score是未来某个毫秒级时间戳。消费端进程常驻在服务器上，有一个Master进程和一个Worker进程池。Master进程不参与业务，负责监听信号、拉起/重启Worker进程，顺带做Worker心跳监控。每个Worker内部是一个while循环，每隔几百毫秒执行一轮ZRANGEBYSCORE查询，把score大于0且小于等于当前毫秒时间戳的消息全部拉出来，然后逐条ZREM并从回调处理器执行。

为什么要独立一个Master进程？因为在多进程模型里，子进程异常退出是常态。Worker处理业务时可能因为第三方接口超时、PHP内存溢出甚至未知异常直接退出。如果没人管，消费端就静默死了。Master进程就是用来兜底的，它阻塞在swoole_process::wait()上等待子进程退出事件，收到退出事件后立刻重新创建新的Worker。这样整个消费端在进程层面是稳定的，业务异常不会停掉整个队列消费。从服务器进程视角看，它类似这样：

PID PPID COMMAND 5261 1 php easyswoole/consumer master 5262 5261 php easyswoole/consumer worker:0 5263 5261 php easyswoole/consumer worker:1 5264 5261 php easyswoole/consumer worker:2

Master的PPID是1，说明它被init进程托管了，服务器重启、SSH断开都不影响它。Worker的PPID指向Master，说明Worker的生命周期完全由Master管理。如果某个Worker因为业务异常退出，Master的wait()会立刻返回，重新拉起一个同样配置的Worker进程。整个进程树非常清晰，这也是这份资源我觉得最值得直接抄的部分。

3. 生产端与消费端核心实现：zset的score是时间戳，取消息要用zrangebyscore而不是zpopmin

3.1 生产者：消息入库与score设计

生产端做的事情很简单，但要设计好两个点：消息的value结构、score的时间单位。value我建议不要直接塞业务字符串，而是拼成一个JSON对象，至少包含job_id、topic、payload、retry_count、available_at五个字段。job_id用UUID，用于幂等判断；topic是消息类型，消费端根据topic分发到不同业务处理器；payload是业务上下文；retry_count是重试次数；available_at是这条消息第一次允许被消费的时间戳。

<?php // Producer.php 生产端入队示意 function pushDelayMessage($redis, string $topic, array $payload, int $delayMs): string { $jobId = md5(uniqid('', true)); $message = [ 'job_id' => $jobId, 'topic' => $topic, 'payload'=> $payload, 'retry_count' => 0, 'available_at' => (int)(microtime(true) * 1000) + $delayMs, ]; // zadd 的 score 是毫秒时间戳 $redis->zadd('delay_queue', $message['available_at'], json_encode($message)); return $jobId; }

这里一个我特别注意的细节是score必须用毫秒时间戳。如果只精确到秒，同一秒内入队的消息在zset里的顺序会无法区分。消费端扫描时同一个score下面可能挂几十条消息，ZRANGEBYSCORE取出来后还要在内存里按入队顺序排序。用了毫秒时间戳之后，score精度够了，zset天然有序，取出来直接就是按到期时间排列。另外，value里一定要带job_id而不是用score本身做唯一标识，因为多条消息可能score相同，但zset不允许成员重复，如果value就是业务ID，同一业务要被延时执行两次时会直接覆盖上一次的数据。

生产端常见的错误是把延时时间直接作为score，比如“延时五分钟”就把score设置成300。这是不对的，score必须是一个绝对到期时间点，因为消费端扫描时用的是“score小于等于当前时间”这个判断条件。如果score存成相对值300，那所有入队消息都是score=300，要么一入库就被消费掉，要么永远等不到扫描条件成立。

3.2 消费端主循环：ZRANGEBYSCORE + ZREM 的原子性分寸

消费端的核心循环就是扫描到点消息。最开始我图省事用了ZPOPMIN，想当然地以为取出score最小的消息就是最早到期的。这个想法在只做延时任务时能跑通，但有一个致命问题：ZPOPMIN会把还没到点的消息也拿出来。

举例来说，现在库里有一条消息，到期时间是5分钟之后，score是当前时间加30万毫秒。ZPOPMIN取score最小的那条，它会把这条5分钟后的消息直接弹出来，消费端一处理发现还没到点，要么抛回队列，要么就丢掉了。所以正确做法是ZRANGEBYSCORE，按0到当前毫秒时间戳的范围查询，只取已经到期的消息，配合LIMIT限制每次拉取条数。

<?php // ConsumerWorker.php Worker主循环示意 public function run(): void { while ($this->running) { // 只取score在0到当前毫秒时间戳之间的到期消息 $nowMs = (int)(microtime(true) * 1000); $messages = $this->redis->zrangebyscore('delay_queue', 0, $nowMs, [ 'limit' => [0, 200], ]); foreach ($messages as $messageJson) { $message = json_decode($messageJson, true); // 先移除，再分发，避免同一条消息在多个Worker间重复消费 $removed = $this->redis->zrem('delay_queue', $messageJson); if ($removed === 1) { $this->dispatch($message); } } usleep(200000); // 200ms 轮询间隔，可配置 } }

这段代码里最关键的是zset的操作顺序：先ZRANGEBYSCORE取出候选消息，遍历时先ZREM再分发。为什么先ZREM而不是先处理业务再ZREM？因为如果先分发、后移除，处理业务期间另一个Worker也拉到了同一条消息，就会重复执行业务动作。订单关闭、支付回调这类操作重复执行大概率会出问题。先ZREM的效果是“抢先占座”，谁remove成功谁负责处理这条消息，remove返回0说明被别人抢走了，直接跳过。

LIMIT参数也要提一下。这里限制200条，不是越多越好。如果一次拉取几千条到内存，PHP进程内存容易吃紧，同时Redis的响应时间也会明显变长。而且单轮循环处理200条如果消耗两秒，扫描的实时性就差了。200这个值大概是经验值，消息体平均1KB内，单worker单轮内存增量可控在几MB以内。轮询间隔usleep(200000)也是折中选择——太快会不断打Redis，太慢会影响延时精度，毕竟精度要求最高也就是100毫秒级别。

3.3 消息分发与失败重试：把消息踢回队列但不能淹死队列

去掉这层，Worker的dispatch就是根据topic找到对应的业务处理器类，然后调用handle方法。业务处理器处理失败时，不能直接丢弃消息，合理的做法是把消息重新放回zset，但score要设置成下次重试的时间点，同时retry_count加1。

<?php // 失败重试示意：重试次数小于3，打印日志并重新入队 private function retryMessage(array $message): void { $message['retry_count'] += 1; if ($message['retry_count'] > 3) { // 超过最大重试次数，进入死信日志 $this->logWriter->write(json_encode($message) . PHP_EOL, LOG_PATH.'/dead_'.date('Ymd').'.log'); return; } // 延迟5秒、30秒、60秒三级重试 $backoffMs = [$default][$message['retry_count'] - 1] ?? 60000; $message['available_at'] = (int)(microtime(true) * 1000) + $backoffMs; $this->redis->zadd('delay_queue', $message['available_at'], json_encode($message)); }

这里我把重试退避设计成三级：第一次失败等5秒，第二次等30秒，第三次等60秒。第三次还失败就写死信日志，不再无限重试。死信日志的作用非常重要，它至少要记录完整的消息json和失败原因。对线上问题做追溯时，没有死信日志的延时队列就像没有黑匣子的飞机，查问题只能靠猜。

业务处理器是同步阻塞的。我见过有人试图在Worker里用Swoole协程并发处理消息，但带来的复杂度远大于收益，因为有状态业务、数据库事务在协程里一旦出问题，排查成本很高。消费端用多进程并行，每个Worker同步处理自己的消息，已经能满足大部分场景。如果要提高吞吐，加Worker数量即可，不需要在单进程里玩协程。

4. 消费端进程治理与可靠性：Master管进程、Worker跑任务，失败重试最多三次

4.1 Master进程的创建与监控Swoole子进程

Swoole的swoole_process类提供了完整的进程管理能力。先看Master进程的骨架代码，它在启动时创建N个Worker子进程，然后进入事件循环等待子进程退出通知。

<?php // master.php Master进程创建Worker的示意 use Swoole\Process; $workerCount = 4; // 可配置，建议等于CPU核心数 $workers = []; for ($i = 0; $i < $workerCount; $i++) { $process = new Process(function (Process $worker) use ($i) { // 子进程里执行Worker逻辑，注意要为子进程创建独立的Redis连接 $consumer = new ConsumerWorker($i); $consumer->run(); }, false, SOCK_DGRAM, true); $process->start(); $workers[$process->pid] = $process; } // Master进程阻塞在这里，等待子进程退出事件 while (true) { $ret = Process::wait(); if ($ret) { $pid = $ret['pid']; if (isset($workers[$pid])) { unset($workers[$pid]); // 子进程挂了，立刻重拉一个新的 $process = createWorker(count($workers)); $process->start(); $workers[$process->pid] = $process; } } }

这里Swoole比原生pcntl好用的点在于Process::wait()封装了信号处理的细节。裸用pcntl时，SIGCHLD处理逻辑要自己写，阻塞在主循环里的accept/select容易被打断。swoole_process内部处理好了这些脏活，Master进程的逻辑就很干净。

创建worker时传入的参数里，第二个参数false表示不启用标准输入输出重定向，第三个参数SOCK_DGRAM表示创建数据报管道。我们的场景里Master和Worker之间没有需要传递的大量数据，只需要传递信号事件，所以管道类型无所谓。第四个参数true是开启协程，但在我们这里没有用到协程特性，可以保持false。

有一个很容易被忽略的点：Worker子进程内部的Redis连接不能继承Master进程的连接。Master进程如果在启动时往Redis里写了一条数据，连接对象的fd是Master的。Worker在fork之后拿到这个连接对象，往同一个fd写数据，在Redis服务端视角还以为是Master在操作，但TCP连接是全双工的，fork之后父子进程各自持有同一个socket，很容易出现数据交叉。我在案例里遇到过线上Worker处理消息时偶发“Connection reset by peer”，就是这种fd继承问题。解决方案是每个子进程启动时独立创建自己的消息Redis实例。

4.2 优雅退出与平滑重载技术

迭代功能要发版，必然要重启消费端。如果直接kill -9杀掉进程，正在处理中的消息还没ZREM，重启后会被重新消费，倒是不会丢，但可能重复执行。优雅退出要做的核心事情是“停接收新消息、处理完当前正在处理的消息再退出”。

Master进程收到退出信号后，通知所有Worker先结束当前消息处理，再退出循环。代码如下：

<?php // Master进程信号处理示意 use Swoole\Process; foreach (['SIGTERM', 'SIGINT'] as $signal) { Process::signal($signal, function () use ($workers) { // 通知所有Worker优雅退出 foreach ($workers as $process) { Process::kill($process->pid, SIGUSR1); } // Master自身退出 exit(0); }); }

<?php // ConsumerWorker中响应SIGUSR1信号 Process::signal(SIGUSR1, function () { $this->running = false; // 当前消息处理完后，主循环条件不成立，退出 });

这个设计里我踩过一个坑：第一次做的优雅退出没解耦“通知Worker”和“Master退出”，Master先退出了，Worker还在跑，结果Worker变成孤儿进程，继续被init进程接管，下个版本代码就加载不到。正确做法是Master收到退出信号后，先关掉自己的子进程管理，给Worker发SIGUSR1，再等一小段时间，确认Worker全部退出后再完全退出自己。

平滑重载的做法类似：更新代码后，向Master进程发送SIGUSR2信号，Master启动一个新版本Worker池，然后让旧Worker处理完当前消息后退出。这样消息不丢、进程不停，从运维视角看是无感发布。不过这个方案要求Worker启动时必须重新加载PHP文件，不能把业务类载入Master内存后靠继承传给子进程。

4.3 过期消息与死信日志：重试机制之外的二层保险

延时队列有一个天然问题：消费者宕机时间较长，恢复后积压了大批到期消息。如果这些消息是通知类业务，全部集中立刻发送会导致下游系统被瞬时请求击穿。我一般会在消费端做一个“补偿限流”参数，如果单轮取到的消息数量超过阈值，就动态增大轮询间隔，比如从200毫秒提高到2秒，让下游有个缓冲时间。

<?php // 简单限流示意 $messageCount = count($messages); if ($messageCount >= 500) { usleep(2000000); // 积压时降到每秒处理一批 } else { usleep(200000); // 正常情况200ms轮询 }

死信日志的价值前面提过，这里补充一个隐藏功能：它还是延时队列的监控入口。每天凌晨处理死信日志，统计哪些topic失败最多、失败原因分布如何，能提前判断三方接口稳定性。常见做法是写一个统计脚本，按topic和错误码分组，输出到监控看板。我在某公司就靠死信日志里的“订单关闭通知连续失败”定位到下游接口的一次链路超时调整。

Redis内存策略也要在这里提一个注意点。zset里的延时消息如果长期未被消费且没有过期时间，Redis内存会被吃掉。延时队列不能给key设置整体TTL，因为消费是逐步的，但也不能把过期时间交给Redis的key过期策略。正确做法是为延时消息独立建一个Redis实例，并把maxmemory-policy设置成noeviction，防止内存淘汰机制误删尚未消费的延时消息。

5. 避坑指南：同一份代码我踩过的五个Redis+Swoole的坑

5.1 坑一：Redis连接被跨进程复用，业务方偶发“Redis server went away”

现象：线上消费端运行一段时间后，worker日志开始出现Redis server went away，并且不是全部worker都有，而是随机某个worker报错。业务消息偶发丢失或者重复处理。

原因：Worker进程启动时，我图省事把Master进程里已经建立好的Redis连接对象直接传给了Worker，也就是在fork之前创建了Redis连接，fork之后子进程繼續持有同一个socket连接。多个进程共享同一个Redis连接时，一个进程的关闭操作会释放连接，其他进程再用就拿到一个失效连接。

解决：Worker子进程内部重新创建自己独立的Redis实例，绝对不允许连接对象在进程间共享。在Swoole的Process回调函数中new pRedis()，通过单例按子进程ID区分，保证每条worker只使用自己那一份连接。

5.2 坑二：score时间单位不一致，任务提前或延迟执行

现象：某次把延时逻辑从秒级改成毫秒级时，只改了生产端，消费端的时间戳计算还停留在秒。结果所有消息在入队那一刻就被consumed，订单超时关闭变成下单立刻关闭。

原因：score用的时间单位是秒，7165233421这种值要比当前时间的毫秒值小得多，ZRANGEBYSCORE条件0 <= score <= now_ms在score为秒时永远成立，消息全部到点。

解决：全局统一用microtime(true) * 1000并强转int类型。我在生产者和消费者的Redis调用入口写了同一个时间获取函数，只要改这一处即可。排查时看一下入队消息的available_at字段值是不是超过当前毫秒时间戳，立刻能判断单位是否错了。

5.3 坑三：ZRANGEBYSCORE + ZREM不是原子操作，并发场景重复消费

现象：多Worker场景下，业务上偶尔出现“重复通知”“订单重复关闭”的投诉。排查发现同一条消息JSON出现在两个日志文件里。

原因：WorkerA和WorkerB同时扫到同一条消息，WorkerA先ZREM成功，WorkerB的ZREM也执行了但返回0。问题在于ZRANGEBYSCORE和ZREM之间没有锁，极端情况下B在A未ZREM前就处理了这条消息。

解决：把ZRANGEBYSCORE + ZREM整体迁移到Lua脚本中，让Redis保证原子性。简单做法是把扫描和移除通过Lua的redis.call联动执行，然后客户端只拿到已经移除成功的消息，不会发生竞争。

5.4 坑四：积压消息把下游打挂，Redis内存在早高峰飙升

现象：活动秒杀场景下，一个topic积累了20万条延时消息，到点后所有Worker一起满负荷消费，在2分钟内全部推送到了下游HTTP接口。接口瞬间超时，Redis内存也在消费高峰期飙升。

原因：消费端的并发数量没有做业务侧限制。20万条消息被4个Worker平均分配后，每个worker每分钟要处理几千条，数据库和HTTP接口承受不住。

解决：消费端引入令牌桶或简单计数器限流，每次取出后业务侧控制发送速率。我在消费循环里加了一个简单的“每秒最大处理量”配置，超出部分放下轮循环处理，不求快但求稳。死信和重试消息也要排队，不能插队。

5.5 坑五：Worker进程无异常日志但不再消费，消息堆积

现象：某次线上运维发现延时任务堆积，登录服务器看进程还在，日志文件却几个小时没更新。kill掉Master后进程又被系统拉起，但依然没有消费日志。

原因：Worker里某个业务方法出现了死循环或阻塞等待，进程没有退出，Master无法感知“假死”。进程看起来活着，实际卡在某个网络请求或IO等待上，消费者循环已经无法执行到ZRANGEBYSCORE。

解决：在Worker里加入心跳机制，每次正常循环结束时向Redis写入一个last_active时间点。Master进程周期性检查所有Worker的last_active，超过阈值就强制kill掉该Worker并拉起新的。这个方案能有效处理临界业务死锁场景，而且实现成本低，不依赖任何HTTP健康检查组件。

6. 进阶用法：消费端平滑无感更新代码，用信号配合Keepalive让队列服务越跑越稳

进程常驻之后，发布更新一直是个麻烦事。一开始的方案很粗暴，先pkill杀掉再启动新的，那段时间总会有几秒空窗，消息会被积压，有时还会因为杀进程的时机不当造成重复消费。后来我把平滑重载做成了标准操作：更新代码文件之后，向Master进程发送SIGUSR2信号。Master收到信号后，用新代码创建一组全新的Worker进程，让新Worker先在后台待命，然后通知旧Worker处理完当前消息退出，完成新旧交替。整个过程服务一直在线，消息不丢不重。这要求业务代码里不能有跨请求的静态缓存状态，每个Worker都是无状态启动。

要实现这个效果，Master里要做两处变更：一是把Worker进程对象按代际分组，新老进程的pid分开管理；二是等旧进程全部退出后再回收其资源。大概思路是定义一个generation标识，收到SIGUSR2时gen++，新Worker都挂在新的generation下，Master每次遍历时只清理旧generation下已经退出的进程。这个改造的代码量不大，但换来了随时发版、随时回滚的能力。

另一个让我比较省心的习惯是在每个Worker里加一个swoole_timer_tick心跳。它不只是写Redis时间戳，还在心跳里打印当前状态，包括workerHandle内存占用、最近一分钟消费消息数、失败次数。这样即使某个Worker假死，Master也能在黑匣子里看到最后心跳时的内存和消息数据。

<?php // 心跳定时器示意 swoole_timer_tick(10000, function () use ($consumer) { $memory = memory_get_usage(true); $this->logger->info(sprintf('worker=%d mem=%dMB total=%d failed=%d', $consumer->getWorkerId(), $memory >> 20, $consumer->getTotalConsumed(), $consumer->getTotalFailed() )); });

心跳配合Master的强制kill，这套组合在资源里是绑定使用的。从那以后我每次上线消费端，都强制走一遍“发信号-看日志-查心跳”三步：先确认平滑重载后新worker日志正常输出，再等10分钟确认消费速率和积压曲线回归正常，最后抽查死信日志有没有新增异常。这套动作坚持了半年，延时队列再没出过凌晨被人叫醒的幺蛾子，希望帮到你。

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