简介这份资源提供的是基于Redis与Swoole的延时消息队列多进程消费端实现适合有一定PHP基础并希望深入异步队列、常驻服务开发的工程师。其核心原理是利用Redis有序集合为每条延迟消息保存执行时间戳当到达设定时间后通过ZRANGEBYSCORE取出再借助Swoole的多进程模型创建多个Worker并发处理配合非阻塞I/O避免大量连接阻塞更好地利用CPU多核。压缩包大小1.66MB内含基于Swoole Server编写的消费服务源码其中可以看到Redis客户端的延时队列操作、Worker进程数配置、消息回调注册与异常重试机制等关键模块。该实现还涉及子进程生命周期管理、避免重复消费等实战细节对构建稳定可靠的消息中间件很有参考价值。目前已有1269人学习下载适合用作品牌技术栈中PHP高并发队列服务的学习样本。1. Redis延时消息队列终于不用再写轮询脚本扫数据库了做订单超时关单这个需求的时候我一开始用的是每分钟跑一次定时任务去扫订单表里创建时间超过30分钟且未支付的记录。订单量到几十万的时候这个扫描慢得离谱而且每次扫到的绝大部分记录都是没到期的等于把整个表翻一遍只为找出那几百条该关的。后来换成基于Redis ZSet加Swoole多进程的延时消息队列把待处理任务按到期时间戳存进有序集合到期任务交给多个消费进程并发处理数据库不用扛轮询压力关单延迟从分钟级降到秒级这是这个项目最直接的收益。这套方案适合做延时任务、定时回调、缓存过期补偿这类场景尤其是PHP项目里想把延时任务从数据库扫表中解放出来的团队。项目本身不复杂核心就是Redis ZSet的到期时间排序加上Swoole提供的多进程能力看完这篇你就能自己跑起来。2. 技术选型与整体架构为什么是Redis ZSet加Swoole多进程2.1 延时任务选型对比数据库轮询、RabbitMQ DLX、Redis ZSet做延时队列之前我花了点时间对比了三种常见方案这里直接说结论。第一种是前面说的数据库轮询实现最简单但问题在于空扫太严重任务量越大扫描成本越高而且扫描间隔决定了延迟精度间隔设小了数据库扛不住设大了用户体验差。第二种是RabbitMQ的延迟插件或死信队列这个方案本身很成熟但如果项目里根本没有RabbitMQ为了一个延时队列引入一套消息中间件运维成本和依赖成本都上来了。第三种就是本项目的方案——Redis有序集合加Swoole多进程。Redis ZSet解决的核心问题是怎么知道哪些任务该执行了。把任务的到期时间戳作为score存进ZSetRedis天然按score排序要取出到期的任务只需要一条zrangebyscore命令按当前时间戳取出所有score小于等于当前时间的成员。这个操作的时间复杂度是O(log(N)M)性能上完全没问题。Swoole解决的是到期任务谁来执行、怎么并发执行通过Swoole的Process管理多个子进程每个子进程作为独立的Worker去消费到期任务互不阻塞。这套组合的好处是依赖极简——只要你有Redis和PHP的Swoole扩展就能跑不引入额外的消息中间件。坏处也很明显Redis里没有消息确认机制任务消费失败后需要自己想办法保证不丢这块我在第4章会详细讲。2.2 整体架构一条延时任务的生命周期这个项目的数据流转路径很清晰我先用文字描述一下完整链路方便后面看代码的时候有个整体感。任务投递方比如订单服务调用延时队列的delay()方法把任务ID和延时秒数传进来。延时队列内部做两件事把任务写入Redis的ZSetscore设置为当前时间戳加上延时秒数同时把任务详情写入一个Hash结构因为ZSet里只存任务ID详情需要单独存。接下来是一个扫描进程Scheduler这个进程循环执行zrangebyscore把到期的任务ID从ZSet里捞出来放入一个List队列。Worker进程组多个子进程通过brpop阻塞监听这个List拿到任务ID后从Hash里取详情执行具体的业务逻辑执行完删除Hash里的记录。为什么中间要加一个List因为多个Worker进程如果同时去ZSet里取任务会面临并发竞争问题——两个进程可能同时取到同一个任务ID。而Redis的List配合brpop天然支持多消费者场景一条消息只会被一个消费者取走这样就把并发安全地派发任务这件事交给了Redis自己不需要在PHP代码里加锁。整个系统涉及三个角色投递方、扫描进程、消费Worker。投递方是业务代码里直接调用的部分扫描进程和消费Worker则是Swoole管理的常驻进程通过命令行启动。项目里这两个角色都跑在Swoole的进程池里扫描进程是单进程消费Worker默认起4个可以在配置里调整。2.3 目录结构与启动方式下面说一下项目的目录结构这个是我拆这个项目时整理出来的实际布局后面写代码和二次开发时对照着看会方便很多。. ├── config.php # 配置文件Redis连接、Worker数量、扫描间隔 ├── queue.php # 核心封装延时任务投递与消费的入口类 ├── scheduler.php # 扫描进程扫描ZSet中到期任务并转入List ├── worker.php # 消费Worker从List中取任务并执行业务回调 ├── server.php # 进程管理用Swoole Process启动/管理所有子进程 ├── jobs/ # 业务任务类的存放目录 │ ├── OrderCloseJob.php # 示例任务订单超时关闭 │ └── StockReleaseJob.php # 示例任务库存释放 └── logs/ # 运行日志目录启动方式很简单命令行执行php server.php startSwoole的Process会拉起一个扫描进程和N个消费Worker。停止时执行php server.php stop会向所有子进程发送退出信号Worker处理完当前任务后自动退出避免任务做到一半被杀掉。3. 核心模块拆解一条延时任务从投递到触发的完整链路3.1 投递接口把任务写进ZSet和Hash先来看投递侧的代码这部分封装在queue.php里。核心方法delay()接收三个参数任务类型、任务数据、延时秒数。这里我把任务类型和任务ID拼成一个唯一的key方便后面按类型路由到不同的业务处理类。?php // queue.php 核心投递逻辑 class DelayQueue { private $redis; private $zsetKey delay:queue:zset; // 有序集合score为到期时间戳 private $hashKey delay:queue:hash; // 哈希存放任务详情 public function __construct($redis) { $this-redis $redis; } /** * 投递一个延时任务 * param string $type 任务类型例如 order_close * param array $data 任务数据例如 [order_id 1001] * param int $delay 延时秒数 * return bool */ public function delay($type, $data, $delay) { $taskId uniqid($type . _, true); $score time() $delay; // 到期时间戳作为score // 任务详情写入Hash字段名为taskId $this-redis-hSet($this-hashKey, $taskId, json_encode([ type $type, data $data, create_time time(), ])); // 任务ID写入ZSetscore为到期时间戳 $this-redis-zAdd($this-zsetKey, $score, $taskId); return true; } /** * 删除任务用于业务提前完成时取消延时任务 */ public function remove($taskId) { $this-redis-zRem($this-zsetKey, $taskId); $this-redis-hDel($this-hashKey, $taskId); } }这段代码里有个细节值得说下任务ID用的是uniqid($type . _, true)把任务类型拼到了ID前缀上这样在ZSet里按前缀就能看出这个任务属于哪个业务排查问题的时候不用把每个ID都挖出来看Hash里的type字段。zAdd的score就是任务的到期时间戳Redis会按这个score自动排序后面扫描的时候取小于当前时间戳的所有成员就是所有已经到期应该执行的任务。这里要注意投递延时任务时一定要确保ZSet和Hash两个写入都成功。我在实际使用中遇到过一个情况Hash写入成功但ZSet写入失败比如Redis连接超时导致任务永远在Hash里但永远不会被执行。常见的做法是把这两个写入包在一个事务里或者用Lua脚本保证原子性。项目里我用的方案是先写Hash再写ZSet如果zAdd返回false就走异常处理把Hash里那条清理掉。3.2 扫描调度到期任务怎么被捞出来扫描进程是整个延时队列的心跳它要不停地去ZSet里查有没有到期的任务有就转运到List里。这个进程在Swoole里作为独立的子进程运行循环间隔通过swoole_timer_tick控制默认每1秒扫一次。?php // scheduler.php 扫描进程核心逻辑 class Scheduler { private $redis; private $zsetKey delay:queue:zset; private $hashKey delay:queue:hash; private $listKey delay:queue:list; // List队列Worker从这里取任务 public function run() { // 每1000ms扫描一次到期的任务 swoole_timer_tick(1000, function () { $now time(); // 取出所有score小于等于当前时间戳的任务ID最多取200个 $taskIds $this-redis-zRangeByScore($this-zsetKey, 0, $now, [ limit [0, 200], ]); if (empty($taskIds)) { return; } // 用Lua脚本保证从ZSet移除并存入List两个操作原子执行 $script LUA local taskIds KEYS[1] local zsetKey KEYS[2] local listKey KEYS[3] local hashKey KEYS[4] for _, taskId in ipairs(ARGV) do redis.call(zrem, zsetKey, taskId) local detail redis.call(hget, hashKey, taskId) if detail then redis.call(rpush, listKey, taskId) end end return #ARGV LUA; $this-redis-eval($script, array_merge( [taskIds, $this-zsetKey, $this-listKey, $this-hashKey], $taskIds ), 4); }); } }这段扫描逻辑里最值得关注的是那个Lua脚本。如果不加这个脚本直接用zRangeByScore取出任务ID后再挨个zRem和rPush会有两个问题一是这两步之间如果进程崩溃任务可能已经从ZSet移除但没进List任务就丢了二是多个扫描进程同时跑的话虽然项目默认只有一个两个进程会重复处理同样的任务。用Lua脚本把zrem和rpush包成一个原子操作Redis单线程执行脚本天然避免了这两个问题。limit [0, 200]这块是批量处理的保护。如果同一时间有上万条任务到期一次性全部捞出来再转运Redis的响应会变慢而且Lua脚本内部循环处理大量元素也会阻塞Redis。每次最多处理200条下一轮扫描会继续处理剩余的这样把一次大任务拆成多次小任务对Redis的压力是平滑的。实际运行下来1000ms扫描一次每次200条吞吐量完全够用。3.3 消费Worker多进程并发取任务消费Worker是真正干活的地方。多个Worker进程同时用brpop阻塞监听List队列谁抢到谁处理。这里用brpop而不是rpop是为了避免Worker空转消耗CPU——brpop在没有任务时会阻塞住直到有任务进来或者超时我设置了超时时间2秒超时后继续下一轮监听这样还能顺便检查一下进程是否需要退出。?php // worker.php 消费Worker核心逻辑 class Worker { private $redis; private $listKey delay:queue:list; private $hashKey delay:queue:hash; private $jobMap []; // 任务类型映射到具体的Job类 public function __construct($redis, $jobMap) { $this-redis $redis; $this-jobMap $jobMap; } public function run() { while (true) { // 阻塞取任务超时2秒返回格式 [listKey, taskId] $result $this-redis-brPop([$this-listKey], 2); if (empty($result)) { // 超时且没有任务检查是否有退出信号 if ($this-shouldExit()) { break; } continue; } $taskId $result[1]; $detailJson $this-redis-hGet($this-hashKey, $taskId); if (empty($detailJson)) { // 任务详情不存在说明已经被处理或删除直接跳过 continue; } $detail json_decode($detailJson, true); $jobClass $this-jobMap[$detail[type]] ?? null; if (!$jobClass) { // 没有注册的任务类型记录日志并删除 error_log(未注册的任务类型: {$detail[type]}, taskId: {$taskId}); $this-redis-hDel($this-hashKey, $taskId); continue; } // 执行具体的业务逻辑 try { $job new $jobClass(); $job-handle($detail[data]); // 执行成功后删除Hash里的任务详情 $this-redis-hDel($this-hashKey, $taskId); } catch (\Exception $e) { // 记录失败日志任务保留在Hash里后续可以人工补偿 error_log(任务执行失败: {$taskId}, 错误: . $e-getMessage()); // 这里特意不删除Hash记录方便后续排查重试 } } } private function shouldExit() { // 通过读取一个进程文件或者信号量判断是否需要退出 // 具体实现在server.php中通过proc_open写入控制文件 return file_exists(/tmp/delay_queue_stop_ . posix_getpid()); } }执行流程是Worker从List取到taskId再去Hash里捞详情根据detail里的type字段找到对应的Job类实例化后调用handle方法。业务逻辑执行完从Hash删除详情。这个删除操作很关键它标志着这个任务真正结束了不然任务会在Hash里无限堆积。关于失败处理这里做了一个刻意的设计业务逻辑抛出异常时不删除Hash里的任务详情只记录日志。这意味着这个任务虽然没有被继续执行但它的现场被保留下来了排查问题时能从Hash里看到任务的数据。为什么要这样做而不是直接重试因为在订单关单、库存释放这类场景下任务失败往往意味着下游接口有问题比如支付系统不稳定立即重试大概率还是失败可能会把Worker卡在同一个任务上后面的任务全部堵住。保留现场、记录日志、由人工或者补偿脚本去处理比自动重试更稳妥。我在实际项目中遇到过自动重试把Redis连接打满的情况从那以后我对消费失败的态度就是先记录别硬刚。4. 多进程消费端实现进程管理与可靠性保障4.1 Swoole进程模型谁负责启动和控制子进程前面讲了扫描进程和Worker各自的逻辑这一章要解决的是这些进程怎么被拉起来、怎么被管理。项目里用的是Swoole的Process类在server.php里统一管理。先看启动部分的代码。?php // server.php 进程管理 require_once queue.php; require_once scheduler.php; require_once worker.php; $config require config.php; // 保存所有子进程的PID用于stop时精确关闭 $childPids []; // 启动扫描进程 $schedulerProcess new Swoole\Process(function () use ($config) { $redis createRedis($config[redis]); $scheduler new Scheduler($redis); $scheduler-run(); }, false, SOCK_STREAM); $schedulerProcess-start(); $childPids[] $schedulerProcess-pid; // 启动N个消费Worker子进程 for ($i 0; $i $config[worker_count]; $i) { $workerProcess new Swoole\Process(function () use ($config, $i) { $redis createRedis($config[redis]); $jobMap require config.php; // 这里应该是jobMap配置 $worker new Worker($redis, $jobMap[job_map]); $worker-run(); }, false, SOCK_STREAM); $workerProcess-start(); $childPids[] $workerProcess-pid; } // 主进程等待子进程退出回收资源 foreach ($childPids as $pid) { Swoole\Process::wait(); }这里有个关键点每个Worker子进程内部都创建了独立的Redis连接。这是因为Swoole子进程是fork出来的父进程创建的Redis连接在fork后不能直接复用——子进程之间共享同一个socket连接会导致数据错乱。我在第一次写这个项目的时候就踩过这个坑父进程创建了一个Redis连接传给所有子进程结果线上任务处理出现大量随机失败查了半天才发现是连接共享导致Redis响应串了。后来改成每个进程独立创建连接问题立刻消失。这是一条血泪经验任何用Swoole多进程的项目都要记住进程内资源要各自创建不能复用父进程的。Swoole\Process::wait()是主进程的收尾动作阻塞等待子进程退出。如果某个Worker异常崩溃了wait会返回它的PID这时候可以做重启策略——虽然项目默认没有做自动重启但我在二次开发时加了wait返回后检查退出码非0说明是异常退出直接重新fork一个新的Worker。4.2 Worker进程的优雅退出任务不丢失的关键进程管理不只是启动更重要的是停止。想象一个场景线上要发布新代码需要重启消费进程。如果直接kill -9杀掉WorkerWorker正在处理的任务就断在半路了——如果任务已经做了业务操作但还没来得及删除Hash记录重启后Hash里还有记录但List里已经没有了这个任务就永远停在Hash里既不会被执行也不会被清理。项目里对优雅退出做了两个层面的处理。第一层server.php收到stop命令时遍历所有子进程PID发送SIGTERM信号而不是直接杀。Worker的brpop设置了2秒超时它每2秒会醒来一次检查shouldExit()的状态。这个检查的实现在worker.php里是检查一个控制文件是否存在存在就break退出循环。所以一个Worker从收到信号到真正退出最多2秒这2秒内它会把当前正在处理的任务执行完。第二层也就是补救措施Worker退出前如果Redis里还有它的未完成任务怎么办项目目前的做法是任务详情留在Hash里不删除。重启后scan进程不会重新扫描这些任务因为它们不在ZSet里了。我在二次开发时加了一个孤儿任务扫描启动时扫描Hash里所有记录把那些不在ZSet里也不在List里的任务重新放回ZSetscore设为当前时间戳让它们马上被重新处理。这个方法救了我不止一次。?php // 孤儿任务恢复启动时扫描Hash找回丢失的任务 class TaskRecover { public function recover($redis) { $hashKey delay:queue:hash; $zsetKey delay:queue:zset; // 获取Hash里所有任务ID $allTaskIds array_keys($redis-hGetAll($hashKey)); foreach ($allTaskIds as $taskId) { // 如果任务既不在ZSet里未到期/未处理也不在List里排队中 if (!$redis-zScore($zsetKey, $taskId) !$redis-lPos(delay:queue:list, $taskId)) { // 重新放回ZSetscore当前时间让扫描进程立刻捞出来 $redis-zAdd($zsetKey, time(), $taskId); error_log(孤儿任务已恢复: {$taskId}); } } } }这个lPos是Redis 6.2才有的命令如果你的Redis版本比较老可以用lRem的返回值来判断不过那样会误删任务。这里如果版本不支持我一般会换一种方式把List里的所有任务ID取出来放到一个临时Set里再和Hash里的全量比对。虽然性能差点但在启动时跑一次完全能接受。4.3 消费能力的动态扩展增加Worker数量项目默认的Worker数量在config.php里配置是静态的。但实际运行中延时任务的量往往有波峰波谷。比如每天的整点会有大量定时任务触发这时候4个Worker的消费能力可能跟不上List里堆积大量任务延时越积越久。我在这个项目上做的第一个改造就是支持动态扩容Worker。思路不复杂server.php提供一个reload命令读取当前配置的worker_count比较当前存活的Worker数量如果配置变大了就补启动多余的Worker变小了就杀掉多余的。实现的关键是记录每个Worker的PID和退出状态这个在Swoole里可以通过Process::wait(false)非阻塞方式获取。# 动态扩容示例先把config.php里的worker_count改为8然后执行 php server.php reloadreload逻辑先读取config里的worker_count再统计server.php里维护的alivePids数组长度比较两者。如果配置变大for循环补启动差额的Worker如果配置变小遍历alivePids杀掉多余的Worker只保留前worker_count个。这个机制让我在活动大促时不用重启整个服务就能提升消费能力很实用。5. 避坑指南跑起来之后最常踩的五个坑5.1 任务重复执行Redis连接超时导致业务重复操作现象订单关闭任务偶尔对同一订单执行两次关闭操作下游系统收到重复请求。原因Worker执行完业务逻辑后删除Hash记录的这一步因为Redis连接超时失败了。下次扫描时任务已经不在ZSet里zrem已经执行过但Hash里的详情还在恢复脚本会把任务重新放回ZSet。解决删除操作失败时做一个补偿标记比如在业务处理结果里增加一个去重字段或者把删除操作放在try/finally里失败后立即重试3次。我在实践中发现重试3次基本能覆盖绝大多数连接抖动。5.2 扫描进程阻塞Lua脚本批量太大拖垮Redis现象某次活动大量定时任务同时到期Redis响应突然变慢所有操作延迟升高。原因扫描进程Limit设置为200但Lua脚本内部每次hget一个任务详情200个hget在一个脚本里执行阻塞Redis约几十毫秒。任务量再大时这个阻塞时间会线性增长。解决把批量大小调低到50同时把Lua脚本里的hget改成用hmget一次性取出所有任务的详情减少脚本内部的命令数量。改造后即使每秒处理500个任务Redis的阻塞时间也低于10ms。5.3 Worker进程僵死任务回调里出现死循环现象某个Worker的CPU占用100%其他Worker正常List里的任务只进不出。原因业务Job类的handle方法里出现了死循环一般是调用外部接口时没有设置超时接口一直不返回代码卡在等待响应。解决这一步在框架层面不好拦截只能从Job编写规范上约束。我在项目的Job基类里加了一个最大执行时间的限制用pcntl_alarm设置定时器超时后抛异常终止执行。这个保护很粗暴但确实有效从那以后Worker僵死的问题再没出现过。5.4 时区问题服务器时区变了导致任务提前或延迟执行现象部署到新服务器后所有延时任务都提前了8小时触发。原因新服务器的PHP时区配置是UTC而代码里time()获取的是UTC时间戳但业务逻辑里计算延迟时用了带时区的本地时间两边差了8小时。ZSet的score使用的是时间戳时间戳本身是时区无关的问题出在投递方把本地时间字符串转时间戳时用错了时区。解决统一在配置里显式设置date_default_timezone_set(Asia/Shanghai)所有时间计算都使用同一个时区基准。这个坑排查起来很玄学因为本地测试一切正常一上服务器就偏。5.5 消费端重复拉起没有锁保护导致多套进程同时消费现象运维手动执行了两次php server.php start结果消费Worker翻倍同一个任务被两个不同Worker同时处理。原因没有做进程单例保护重复启动时没有检测到已有实例在运行。解决启动时先检查PID文件如果PID文件存在且对应的进程还活着直接拒绝启动。?php // 启动时互斥检查 $pidFile /tmp/delay_queue_server.pid; if (file_exists($pidFile)) { $pid intval(file_get_contents($pidFile)); if (posix_kill($pid, 0)) { exit(服务已在运行中PID: {$pid}\n); } } file_put_contents($pidFile, posix_getpid());posix_kill($pid, 0)是检查进程是否存在的常用手法发给进程的信号是0不会真的执行任何操作但会返回进程是否存活。这个方法在PHP里做进程单例非常方便。6. 二次开发把延时队列改造成通用调度组件改造方向有三个我按推荐优先级来说。第一个是增加任务延迟策略现在的实现是fixed delay——投递后固定延时执行。但实际业务里经常需要按业务条件动态计算延迟比如支付回调没到第一次30秒后检查第二次2分钟后检查第三次10分钟后再查。我在Job类里增加了一个nextDelay()方法Job的handle方法执行完后如果返回了一个新的延迟秒数Worker就把这个任务重新投递到延时队列否则任务结束。改造后的handle流程从执行完就删变成执行完看返回值决定是否续延代码只改了十几行。第二个方向是支持At-Least-Once语义的任务重试现在的实现是失败就丢弃保留Hash记录但不再执行。可以加一个重试次数字段存在Hash详情里失败重试时先判断重试次数是否超过上限没有就重新投递到ZSetscore为当前时间加退避秒数。这样就能实现带退避的可靠任务适合对接第三方支付这种需要重试的场景每次重试间隔为2的指数幂最多重试5次。第三个方向是把任务类型注册改成自动发现。现在的jobMap是配置文件里手写的数组新增一个任务类型要改配置文件。改成扫描jobs目录下的所有PHP文件约定类名以Job结尾自动加载并抽取出类的静态属性$type作为任务类型标识。这样再新增业务任务时只需要在jobs目录下新增一个文件不用改任何配置二次开发的门槛进一步降低。这套队列我前前后后折腾了两个迭代从最早的单进程轮询到现在的Swoole多进程消费中间踩过连接共享的坑也踩过时区错乱的坑。到现在我每次新部署一个PHP常驻进程服务都会强制走一遍进程单例、优雅退出、孤儿任务恢复这三件套写代码之前先想清楚任务失败时的现场怎么保留。这个项目的代码量不大但涉及的知识点很集中Redis ZSet的排序特性、Lua脚本的原子性、Swoole的进程生命周期管理。把这些点吃透了你在自己的项目里做延时任务调度时会多很多底气。希望这篇拆解对你有帮助。本文还有配套的精品资源点击获取