从凌晨两点的报警电话说起聊聊我用纯PHP搭建分布式任务调度的全过程如果你维护过带有大量异步任务的PHP系统大概率经历过这种场景凌晨两点监控群突然开始刷屏某个跑批任务堆积了十万条数据没消费而你的crontab里躺着的那条*/1 * * * * php worker.php完全无动于衷。更头疼的是这台唯一的调度服务器一旦宕机所有任务全部停摆没有任何节点能接管。这个坑我踩过不止一次后来终于下决心用纯PHP方案重新实现了一套分布式任务调度把调度、执行、监控彻底拆开才从无休止的半夜爬起来手动跑脚本里解脱出来。这篇文章不聊Swoole常驻内存那套高级玩法当然会用上一点也不扯Kafka、Flink这些重型武器就聚焦在用PHP本身配合Redis和MySQL怎么搭建一套可靠的分布式任务调度系统。适合那些业务量在中等规模、不想为了一个调度系统引入整套Java微服务体系、但又受够了单机crontab的团队。我会把选型原因、核心代码、部署踩坑、故障排查全部分享出来都是实际生产环境验证过的方案。1. 为什么单机crontab撑不住分布式任务调度的三个核心痛点先花点时间说清楚我们到底在解决什么问题。很多人觉得分布式任务调度就是多弄几台服务器跑cron这是个常见的误区。单机crontab的最大问题不是性能而是可靠性、扩展性和可观测性这三个维度全面缺失。1.1 可靠性单点故障意味着所有定时任务一起陪葬crontab跑在系统层的crond守护进程里它本身非常稳定但你部署的PHP脚本就不是了。我碰到过一次最典型的故障某台服务器上一条任务执行到一半因为内存溢出进程被杀但在crontab里它没有加锁下一个分钟周期又启动了新的进程两个进程同时处理同一批数据结果就是大量重复写入直接拖垮了数据库。这种问题在没有分布式调度框架时只能靠自己在脚本里写flock文件锁对付单机勉强够用一旦任务迁移到多台服务器文件锁就失效了——服务器A加锁的文件服务器B根本感知不到。另一个更致命的是宕机场景。假设你的调度机是一台2核4G的小服务器某天半夜磁盘被日志写满crond直接起不来。所有关键任务都没执行但没有任何告警因为crond本身不会通知你我没跑成。等你发现的时候已经错过了一整个批处理窗口。分布式调度要做的基础能力之一就是任务调度的高可用——至少有多个节点同时盯着任务表一个节点挂了另一个节点要能接过调度权。1.2 扩展性单机并发能力有一个隐形的天花板crontab每分钟只能触发一次任务这是crond的设计限制。如果你的任务队列需要每秒消费上百条消息这样的吞吐crontab根本做不到。更麻烦的是crontab的触发粒度和执行能力没有解耦。举个例子你的worker.php脚本需要处理积压的邮件队列处理完后需要重新排队等下一分钟。如果你给这个任务在crontab里设置了并发进程数量你会发现脚本里有大量重复的避免自己同时跑多个实例的判断逻辑。而在分布式架构里调度逻辑和执行逻辑是分离的调度器只负责把任务标记为可执行真正干活的Worker节点独立运行有多少个Worker实例就能扩展到多少并发两者各司其职扩展只是加机器的事。1.3 可观测性跑完没跑完、跑成没跑成你根本一无所知单机crontab时代任务的执行状态只能靠任务自己记录。如果脚本里没有完善的日志任务失败了、超时了、重复执行了你只能去服务器上看/var/log/cron而且默认只记录了启动命令没有输出和状态。生产环境里我见过太多团队排查这个任务跑没跑的方式是——登录服务器ps aux | grep一下没有就直接手动执行一次。这个痛点其实是最影响日常体验的。分布式调度框架天然会把任务状态、执行日志、消费进度统一收敛到一个地方你在一个管理页面上就看得到所有任务的心跳、成功/失败次数、最近一次执行时间。这套可观测能力才是分布式调度相对于单机crontab的代差级优势。2. 架构先行任务中心、调度器、执行器三层如何分工搞清楚了痛点很多人第一反应是那我上Gearman吧或者用RabbitMQ。但我的建议是先想清楚架构分层再决定怎么实现。分布式任务调度本质上是三件事任务的定义与存储、任务的触发决策、任务的实际执行。把这三件事拆开整个系统就清晰了。2.1 三层架构的核心职责划分我把这套系统划分成了三个角色任务中心Task Center负责所有任务的注册、状态管理、结果记录。它是一张MySQL表承载任务的定义——任务类型、优先级、超时时间、重试次数、调度规则、最近执行状态、最近执行时间。所有节点都只跟任务中心交互不互相直接通信。调度器Scheduler常驻进程每隔固定周期扫描任务中心找出到达调度时间且状态为待执行的任务把它们标记为已派发写入一个待消费的队列我这里用的Redis List。调度器本身是多个节点同时跑的通过分布式锁保证同一时刻只有一个节点在对某个任务做派发决策。执行器Worker也是常驻进程从Redis队列里取任务真正执行任务逻辑。Worker可以按业务拆成多个进程组每个进程组消费一个专属队列互不干扰。执行完成后向任务中心回写执行结果。这个分层的好处在于调度器不要直接执行任务执行器也不要在任务中心里抢任务标记。调度器一旦只负责决策它的负载极低可以铺很多节点做高可用执行器只负责执行可以独立扩展数量按业务量随意横向加。两者之间用Redis队列做解耦还能顺带实现流量削峰。2.2 任务怎么被定义和流转我用一个比较精简的任务表来说明流转流程。任务投递时往MySQL里insert一条记录status0待调度、next_run_at是下一次计划执行时间。调度器每次扫描WHERE status0 AND next_run_atNOW()命中后进入派发流程先把任务的status更新为1已派发这一步要带一个条件判断——只有当status0时才能更新成功避免多个调度器节点重复派发。把任务标识比如任务ID和类型推入对应业务的Redis队列。Worker从队列里消费到任务后把status更新为2执行中开始执行任务。执行完毕后按结果把status3成功或status4失败写回任务中心并更新next_run_at和last_run_at。这个流转过程里最难的是多个调度器节点如何避免重复派发。如果两个调度器同时扫到同一条任务同时执行UPDATE就出现了竞争条件。我试过两种解法效果都不错后面会详细说。2.3 为什么Redis比数据库轮询更适合做执行队列很多人会说我直接用MySQL做队列行不行行但我不推荐作为主力方案。MySQL的SELECT ... FOR UPDATE SKIP LOCKED可以做队列但有两个明显短板数据库连接数压力大而且如果你在任务中心表上频繁加行锁业务表也会被拖累。Redis的List数据结构天生适合做轻量级队列LPUSH投递、RPOP消费配合BRPOP阻塞读取Worker进程不会空转。加上Redis本身有持久化和主从切换队列数据不会轻易丢失。对于绝大多数PHP团队来说Redis已经是标配不需要额外引入消息队列组件就能获得不错的队列语义。当然如果你有严格的消息不丢要求、需要多副本消费那就应该上RabbitMQ或者Kafka这个后面选型对比里会详细分析。3. 选型对比Redis直连、Gearman与外部消息队列哪种方案更适合你的团队这一节聊选型因为我发现很多团队在分布式任务调度这个需求面前容易陷入两个极端要么觉得MySQL就能搞定要么觉得得上一套微服务全家桶。我的观点是要根据团队基础设施和业务体量来选这里给出三个可落地的方案和它们的适用边界。3.1 方案ARedis直连队列本文主线方案完全依赖MySQL Redis不引入任何新的服务组件。调度器定时扫描任务中心把到期的任务推进Redis队列Worker从队列消费执行完回写状态。这套方案的优点非常明显成本极低Redis和MySQL任何PHP团队都有代码量可控核心调度器一个进程文件几百行就够运维简单没有任何需要额外维护的中间件。短板在于Redis队列本身不提供消息确认语义Worker消费后即使任务执行失败消息也已经被RPOP移除了需要自己实现补偿机制。此外复杂的定时触发策略比如跨时区cron、表达式要靠自己解析cron表达式工作量会多一些。这套方案适合中小团队、任务量每秒万级别以内的场景。3.2 方案BPHP-Gearman专业任务分发Gearman是纯C实现的任务分发系统PHP扩展成熟它天然支持多Worker并发、任务优先级和持久化队列。如果团队愿意引入一个新的守护进程组件Gearman的接入成本比大部分人想象的低# Ubuntu/Debian安装 apt install gearman-job-server gearman-tools php-gearmanGearman的核心模型是客户端提交任务 - 服务器派发给空闲的Worker - Worker处理完返回结果。它自带任务确认机制Worker没处理完之前任务不会消失故障后还能重新派发。如果你的场景里大量任务是异步、即时、高并发的比如缩略图生成、短信发送Gearman是很好的选择。但Gearman也有明显短板它的定时调度能力很弱需要你自己写一个常驻进程去按cron规则投递任务而且它是中心化架构任务分发服务器本身是个单点需要自己做主从。所以我通常的建议是如果你需要的是强大的调度触发能力 灵活的队列解耦选方案A如果你的核心诉求是异步任务分发、带确认语义选Gearman。3.3 方案C外部消息队列RabbitMQ/Kafka 独立调度器到了这个层面你已经不是在做任务调度而是在做一个完整的异步消息系统。RabbitMQ提供完善的ACK、死信、延迟队列Kafka提供巨大的吞吐和离线消费能力。这套方案的正确姿势是把调度器做在队列外面——调度器按cron规则往MQ里推消息消费者的逻辑在MQ的消费组里实现任务的执行状态通过MQ的回调机制跟踪管理。这个方案适合任务量大、要求严格不丢失、有明确的消费语义的团队。但代价是运维复杂度陡增RabbitMQ和Kafka都不是改个配置文件就能上线的组件你需要考虑集群部署、监控、客户端连接处理。对中小团队来说我倾向于不建议为了任务调度这个场景去上Kafka——你的业务量很可能远没有到那个级别引入一个需要专业运维的重组件收益很可能抵不上它的维护成本。3.4 我的推荐按团队规模分阶段演进我个人的经验是分三步走第一版用Redis直连队列一个crontab 两个PHP常驻进程解决90%的问题等业务量涨了再把某个特定业务的队列迁移到Gearman或者RabbitMQ调度器本身不动最后如果确实需要复杂的流式处理再考虑Kafka。这三步的过渡成本很低因为调度器、任务中心、执行器是分离的替换任何一层都不需要动整个系统。下面就用方案A为主线带大家把整套系统从零写出来。4. 核心代码全拆解任务中心、调度器、Worker的完整实现链路前面讲了很多理念现在进入代码环节。我会按照任务中心定义 - 投递端 - 调度器 - Worker的顺序把每一部分的职责、实现细节和为什么这么写说清楚。4.1 任务中心一张表 一个状态机搞定所有任务的生命周期管理任务中心的表结构是整个系统的地基。设计这张表的时候我踩过不少坑尤其是重试次数和调度表达式这两块的关联关系一开始没设计好后来不得不做数据迁移。这里直接给出我在生产环境用下来的最终版CREATE TABLE task_center ( id BIGINT UNSIGNED NOT NULL AUTO_INCREMENT, task_code VARCHAR(64) NOT NULL COMMENT 任务编码, 业务唯一标识, task_type TINYINT NOT NULL DEFAULT 0 COMMENT 0一次性任务, 1周期任务, 2定时任务, payload JSON NOT NULL COMMENT 任务参数, 由业务自行定义, cron_expr VARCHAR(100) DEFAULT NULL COMMENT 调度表达式, 周期/定时任务使用, priority TINYINT NOT NULL DEFAULT 5 COMMENT 优先级, 0最高, 9最低, delay INT NOT NULL DEFAULT 0 COMMENT 延迟执行秒数, 0表示不延迟, timeout INT NOT NULL DEFAULT 300 COMMENT 执行超时时间, 单位秒, max_retry TINYINT NOT NULL DEFAULT 3 COMMENT 最大重试次数, retry_count TINYINT NOT NULL DEFAULT 0 COMMENT 已重试次数, status TINYINT NOT NULL DEFAULT 0, plan_next_run_at DATETIME NOT NULL, last_run_at DATETIME DEFAULT NULL, last_result TEXT DEFAULT NULL COMMENT 最近一次执行结果, 失败信息等, created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, PRIMARY KEY (id), KEY idx_status_next (status, plan_next_run_at), KEY idx_task_type (task_type) ) ENGINEInnoDB DEFAULT CHARSETutf8mb4;关于状态机有几个细节容易被忽略status字段我直接用数字不用枚举字符串。MySQL的ENUM在ALTER TABLE加新值的时候非常麻烦数字可以随时扩展。plan_next_run_at这个字段是调度器判断该不该派发的唯一依据不能直接用created_at代替。因为周期任务每次执行完成之后需要更新下一次的计划时间。payload用JSON类型不要拆多个字段。分布式任务调度的任务参数内容是不确定的用JSON最灵活。索引必须覆盖(status, plan_next_run_at)这个组合这个查询是调度器每周期都要执行的过滤掉无关的任务。任务状态机的完整流转status含义流转方向0待调度新任务或等待到期0-11已派发投递队列成功1-2 或 1-0(投递失败回滚)2执行中Worker已领取2-3 / 2-4 / 2-0(超时重排)3成功3-0(周期任务进入下一轮)4失败4-0(重试次数未耗尽) 或 4-5(放弃)5放弃重试次数耗尽终态人工介入这个状态机的重点是不要让状态卡死在某一步。状态0的待调度任务一定要有调度器扫描时发现它过期太久没被消费的保护逻辑后面讲。2的执行中任务要配合执行器的心跳上报机制如果Worker崩溃心跳中断调度器要能找回这个任务重新派发。4.2 投递端业务代码如何把任务交给调度系统任务中心的表只是存储业务代码需要按规则往里面写入任务。这里我封装了一个TaskProducer类业务侧只管调用不用关心底层是MySQL还是Redis。?php declare(strict_types1); namespace App\Task; use PDO; final class TaskProducer { public function __construct(private PDO $pdo) { } /** * 投递一个周期任务按cron表达式触发 */ public function cron(string $taskCode, string $cronExpr, array $payload, int $priority 5, int $timeout 300, int $maxRetry 3): int { // cron表达式转换成下一次触发时间这里用第三方库cron-expression $nextRunAt (new \Cron\CronExpression($cronExpr)) -getNextRunDate() -format(Y-m-d H:i:s); return $this-insertTask([ task_code $taskCode, task_type 1, payload json_encode($payload, JSON_UNESCAPED_UNICODE), cron_expr $cronExpr, priority $priority, timeout $timeout, max_retry $maxRetry, plan_next_run_at $nextRunAt, status 0, ]); } /** * 投递一个一次性延迟任务, 延迟多少秒后执行 */ public function delay(string $taskCode, int $delaySeconds, array $payload, int $priority 5, int $timeout 300, int $maxRetry 3): int { $nextRunAt date(Y-m-d H:i:s, time() $delaySeconds); return $this-insertTask([ task_code $taskCode, task_type 0, payload json_encode($payload, JSON_UNESCAPED_UNICODE), delay $delaySeconds, priority $priority, timeout $timeout, max_retry $maxRetry, plan_next_run_at $nextRunAt, status 0, ]); } private function insertTask(array $data): int { $columns implode(,, array_keys($data)); $placeholders implode(,, array_fill(0, count($data), ?)); $stmt $this-pdo-prepare( INSERT INTO task_center ($columns) VALUES ($placeholders) ); $stmt-execute(array_values($data)); return (int)$this-pdo-lastInsertId(); } }这里有两点值得注意cron_expr只在任务中心存了生产结果下一次触发时间会被plan_next_run_at字段具体化。调度器只看时间字段不要在派发时才去解析cron表达式减少调度器的计算负担。我在早期版本里就是派发时解析cron到了大促时调度器CPU直接跑满排查半天才发现是反复解析表达式导致的。一次性任务和周期任务的差别只在task_type调度器对一次性任务执行完就置为终态对周期任务每次执行成功都要根据cron_expr算下一次执行时间。所以cron_expr字段一次性任务可以留NULL周期任务必须填。4.3 调度器多节点竞争 原子派发杜绝重复投递调度器是整个系统的核心大脑也是最容易写错的部分。我见过不少实现方式是用SQL的SELECT ... FOR UPDATE把任务行锁住再更新状态但这种方法在低并发下还行一旦任务量达到一定规模MySQL的行锁竞争会非常严重。我后来换成了原子性CAS更新的方式完全避免了行锁?php declare(strict_types1); namespace App\Task\Scheduler; use PDO; use Redis; final class Scheduler { // 默认每5秒扫描一次任务中心 private const SCAN_INTERVAL 5; public function __construct( private PDO $pdo, private Redis $redis, private int $nodeId ) { } public function run(): void { $this-log(调度器节点 {$this-nodeId} 启动); while (true) { $this-dispatchDueTasks(); sleep(self::SCAN_INTERVAL); } } private function dispatchDueTasks(): void { $now date(Y-m-d H:i:s); // 1. 找出所有待调度且已到期的任务 $stmt $this-pdo-prepare( SELECT id, task_code, task_type, payload, priority, timeout, max_retry, plan_next_run_at FROM task_center WHERE status 0 AND plan_next_run_at :now ORDER BY priority ASC, plan_next_run_at ASC LIMIT 200 ); $stmt-execute([:now $now]); $tasks $stmt-fetchAll(PDO::FETCH_ASSOC); // 2. 逐条CAS更新状态从0改为1 $casStmt $this-pdo-prepare( UPDATE task_center SET status 1, updated_at :updated WHERE id :id AND status 0 ); foreach ($tasks as $task) { $casStmt-execute([ :updated $now, :id $task[id], ]); // rowCount为1表示CAS成功代表这台调度器抢到了该任务的派发权 if ($casStmt-rowCount() 0) { continue; } // 3. 投递到Redis队列, 按业务类型区分队列名 $queueKey task:queue: . $task[task_code]; $message json_encode([ task_id $task[id], task_code $task[task_code], payload json_decode($task[payload], true), timeout (int)$task[timeout], max_retry (int)$task[max_retry], ], JSON_UNESCAPED_UNICODE); $this-redis-rpush($queueKey, $message); $this-log( sprintf( [派发] 任务#%d code%s 投递到队列 %s, $task[id], $task[task_code], $queueKey ) ); } } private function log(string $message): void { echo [ . date(Y-m-d H:i:s) . ] . $message . PHP_EOL; } }这段代码的精髓在于那个CAS更新UPDATE ... SET status1 WHERE id? AND status0。如果多个调度器节点同时扫描到了同一个任务数据库引擎会在行锁上排队谁先执行成功rowCount()就是1谁就获得了派发权其他节点的更新匹配不到status0已经被改成1了rowCount()就是0直接跳过。这个原子操作天然做到了多个调度器竞争一个任务最多只有一个派发成功不需要分布式锁也不需要事务。为什么不用Redis分布式锁我在第一版实现里用SET NX EX做锁每个调度器先抢锁再扫描结果发现问题不少锁持有时间不好控制扫描派发耗时长Redis故障期间整个调度系统停摆。后来改成CAS更新之后调度器的可用性其实更高——即使某个Redis节点短时间抖动也只是影响消息投递那一环任务中心的调度决策不受影响任务可以等Redis恢复后重新投递。4.4 Worker执行器阻塞队列消费 任务超时强制回收调度器把消息推入了Redis队列Worker要做的事情就是从队列里取消息、执行任务、回写状态。这里有几个细节我特意踩过坑后才调整对?php declare(strict_types1); namespace App\Task\Worker; use PDO; use Redis; use Throwable; abstract class BaseWorker { protected PDO $pdo; protected Redis $redis; private string $workerId; private int $heartbeatInterval 30; public function __construct(PDO $pdo, Redis $redis, string $workerId) { $this-pdo $pdo; $this-redis $redis; $this-workerId $workerId; } public function listen(string $queueKey): void { $this-log(Worker {$this-workerId} 开始监听队列 {$queueKey}); while (true) { // 阻塞读取队列最多等待60秒避免空转 $raw $this-redis-brpop($queueKey, 60); if (!$raw) { $this-sendHeartbeat(); continue; } $task json_decode($raw[1], true); if (!$task) { $this-log(队列数据格式错误: . $raw[1]); continue; } $taskId (int)$task[task_id]; $this-markExecuting($taskId); try { $this-process($task); $this-markSuccess($taskId, ok); } catch (Throwable $e) { $this-markFailed($taskId, $e-getMessage()); $this-log([失败] 任务#{$taskId} . $e-getMessage()); } $this-sendHeartbeat(); } } // 业务子类实现真正任务逻辑 abstract protected function process(array $task): void; private function markExecuting(int $taskId): void { $stmt $this-pdo-prepare( UPDATE task_center SET status 2, last_run_at NOW() WHERE id :id AND status 1 ); $stmt-execute([:id $taskId]); } private function markSuccess(int $taskId, string $result): void { // 周期任务执行成功后计算下一次执行时间 $stmt $this-pdo-prepare(SELECT task_type, cron_expr FROM task_center WHERE id :id); $stmt-execute([:id $taskId]); $taskMeta $stmt-fetch(PDO::FETCH_ASSOC); $nextRunAt null; if ($taskMeta (int)$taskMeta[task_type] 1 $taskMeta[cron_expr]) { $nextRunAt (new \Cron\CronExpression($taskMeta[cron_expr])) -getNextRunDate() -format(Y-m-d H:i:s); } if ($nextRunAt) { // 周期任务回到待调度状态静默等到下一次时间 $this-pdo-prepare( UPDATE task_center SET status 0, retry_count 0, plan_next_run_at :next_run, last_result :result, updated_at NOW() WHERE id :id )-execute([:next_run $nextRunAt, :result $result, :id $taskId]); } else { // 一次性任务终态成功 $this-pdo-prepare( UPDATE task_center SET status 3, last_result :result, updated_at NOW() WHERE id :id )-execute([:result $result, :id $taskId]); } } private function markFailed(int $taskId, string $error): void { // 失败重试处理需要在这里完成, 见5.2节 $this-handleRetry($taskId, $error); } private function sendHeartbeat(): void { $this-redis-set(task:worker:alive: . $this-workerId, time()); } private function log(string $message): void { echo [ . date(Y-m-d H:i:s) . ] . $message . PHP_EOL; } }然后业务侧实现一个具体的Worker逻辑就很简单?php declare(strict_types1); namespace App\Task\Workers; use App\Task\Worker\BaseWorker; use Throwable; final class SendEmailWorker extends BaseWorker { protected function process(array $task): void { $payload $task[payload]; // 模拟耗时的邮件发送流程 $this-sendEmail($payload[to], $payload[subject], $payload[body]); } private function sendEmail(string $to, string $subject, string $body): void { // 这里替换成真实的邮件服务调用 usleep(200000); // 模拟200ms耗时 // 注意: 如果发送失败要抛异常, 不要静默吞掉 if ($to killexample.com) { throw new RuntimeException(模拟的发送失败); } } }Worker的核心要点必须用BRPOP而不是RPOP。BRPOP是阻塞读取队列里没有消息时进程会睡在Redis连接上不消耗CPURPOP需要自己usleep轮询空转会吃掉大量CPU。我第一次写的时候用了RPOP开了20个Worker进程CPU直接占满了一半。队列消费和状态回写是分开的两段操作。消费成功不代表任务成功这个语义必须通过状态机来保证。如果Worker在markSuccess之前进程崩溃任务停留在status2执行中需要调度器旁路扫描后面讲把它捞回来重试。BaseWorker里不要自己catch业务异常后吞掉必须抛出让基类的markFailed统一处理重试逻辑。5. 分布式锁与定时任务的进阶实战如何用Redis锁实现跨节点调度协调任务中心的普通任务一次性/延迟任务用CAS更新就能解决多节点竞争问题但有一类场景绕不过去——周期性的全局任务协调典型例子是所有节点一起扫表但只允许一个节点执行的全局清理任务或者多个节点都要抢某一个全局资源的场景。这类场景需要真正的分布式锁。5.1 不用SET NX EX的Redis分布式锁还有什么讲究网上的Redis分布式锁教程千篇一律教大家用SET key value NX PX 30000但在任务调度场景直接这么用会出事如果任务执行时间超过锁过期时间A节点的锁自动过期了B节点趁机拿到了锁两个节点同时执行任务PX过期时间设置太长节点A崩了锁迟迟不释放B一直拿不到任务被阻塞。我试过一套简单可靠的做法把锁分为持有期和续约期两段?php declare(strict_types1); namespace App\Task\Support; use Redis; final class DistributedLock { private string $lockKey; private string $ownerToken; private int $expireSeconds; private Redis $redis; public function __construct(Redis $redis, string $lockKey, int $expireSeconds 60) { $this-redis $redis; $this-lockKey $lockKey; $this-expireSeconds $expireSeconds; // 每个节点生成一个唯一token用于安全释放锁 $this-ownerToken uniqid(node-, true); } public function acquire(): bool { // NX不存在才设置, EX过期时间(秒) $result $this-redis-set( $this-lockKey, $this-ownerToken, [NX, EX $this-expireSeconds] ); return $result ! false; } public function renew(): bool { // 只有持有者自己才能续期用Lua脚本保证原子性 $lua LUA if redis.call(get, KEYS[1]) ARGV[1] then return redis.call(expire, KEYS[1], ARGV[2]) end return 0 LUA; $result $this-redis-eval($lua, [$this-lockKey, $this-ownerToken, $this-expireSeconds], 1); return $result 1; } public function release(): bool { // 同样用Lua脚本校验持有者身份防止误删其他节点的锁 $lua LUA if redis.call(get, KEYS[1]) ARGV[1] then return redis.call(del, KEYS[1]) end return 0 LUA; $result $this-redis-eval($lua, [$this-lockKey, $this-ownerToken], 1); return $result 1; } }丢失的status1已派发但队列消费慢任务也要捞出来重新投递否则一直占着队列名额不执行执行中的status2Worker心跳超时任务认定Worker进程已死需要重置回0并加一次重试计数。表设计上要加一个dispatched_at字段记录派发时间方便扫描过期任务。这部分逻辑我放在调度器里每30秒执行一次独立于主扫描循环。提示分布式锁不是越多越好。能用CAS更新解决的派发竞争就绝不用锁锁的成本和执行期的不可控性都比CAS要高。只有在全局只有一个节点能执行的强约束场景才去依赖分布式锁。6. 实战踩坑记录重复消费、任务堆积、Worker假死的排查与修复最后这部分我把过去一年在生产环境踩过的坑挑几个典型的分享出来每个都附上排查思路和最终修复方案希望能帮大家少走弯路。6.1 重复消费BRPOP 状态回写之间的真空期事故现象某个联名活动的通知任务用户收到两条一模一样的推送。排查后发现是Worker在BRPOP拿到消息后进程突然被OOM Killer杀掉Redis的消息已经被取走了但markExecuting还没执行、状态还停留在1。调度器旁路扫描发现这个任务派发了10分钟还没执行认为它丢了重新投递了一遍于是重复消费了。最终修复方案分三层Worker消费后立刻执行markExecuting把状态从1改为2把已领取未执行的时间窗口缩到最短调度器旁路扫描时对状态1的任务额外判断dispatched_at是否超过一个合理的消费等待窗口我设置的10分钟没超过就跳过避免误捞业务侧对通知类任务做幂等控制任务投递时携带业务唯一键处理前先查重。调度系统再可靠也不可能保证100%不重业务侧兜底才是最后一道防线。6.2 任务堆积调度器秒级扫描变成了扫不动现象双十一预热期一批营销任务在零点准时释放TaskCenter里几万条任务瞬间全部到期。调度器每5秒扫描一次一次查出200条派发过程中又不断有新任务到期队列积压越来越严重业务方反馈活动开始10分钟了券还没发完。排查链路先看Redis队列长度——LLEN task:queue:coupon发现有20万条积压再看Worker消费速率发现单队列只有5个Worker进程在跑每个Worker处理一条券任务要300毫秒5个进程每秒最多消费16条处理20万条需要3.4小时。瓶颈根本不在调度器而在Worker的消费吞吐。修复思路增加Worker进程数量配合BRPOP的多队列消费特性开了50个进程吞吐提升到150条/秒Redis队列在多进程消费场景下会有热点问题我调整成了每个进程独享一个子队列投递端按取模把任务散到多个子队列彻底消除单队列的RPOP竞争。把调度器的扫描批次从200提升到1000减少空扫描次数派发延迟明显下降。这次故障让我意识到分布式调度系统里调度器的派发能力几乎不会成为瓶颈Worker的消费能力才是。设计时一定要给Worker预留横向扩展的手段我后来把所有业务的队列都改成了分片模式。6.3 Worker假死进程活着但不干活心跳还正常现象某天上午收到告警一个下载任务的队列积压严重但看Worker进程状态所有进程都活着心跳也一直在刷新。登录服务器top一看每个Worker进程CPU占用都很低看起来像是在正常阻塞等待。排查链路先用redis-cli手动推一条测试消息进队列发现消息没有被消费说明Worker虽然连在Redis上但消费循环卡住了查看strace发现进程阻塞在MySQL的select等待上——某个Worker中的markSuccess在执行时被一条未提交事务的SELECT FOR UPDATE锁住了定位到是另一个团队在业务数据库里跑了一个长时间的报表SQL锁住了任务中心表的某一行导致Worker回写状态时被卡死。修复方案给所有Worker的MySQL连接设置wait_timeout和lock_wait_timeout避免无限期等待行锁任务中心的读写 credentials 单独建一个账号INNODB_LOCK_WAIT_TIMEOUT10超过10秒直接报错宁可让任务重试也不要卡住整个Worker进程把这个卡死的任务从Worker的process逻辑中捞出来确认是外部锁引起的后续在业务代码里提醒相关团队大查询不要跑在业务库上。提示Worker进程的活着不等于健康。我后来给每个Worker加了一个内部事务数计数器每分钟处理的任务数结合心跳一起上报监控只要计数器归零而心跳正常就判定为假死Supervisor自动重启该进程组。7. PHP生态的部署与运维经验常驻进程、Supervisor和监控告警分布式任务调度系统里的调度器和Worker都是常驻进程这意味着你不能像普通PHP应用一样丢给FPM跑。这一节讲部署和运维层面的关键经验这些都是代码写完之后真正决定系统稳不稳定的量。7.1 Supervisor守护常驻进程崩溃自动拉起、退出码校验PHP的Worker进程再稳定也难免遇到内存泄漏、未捕获异常导致的退出。生产环境我强烈建议用Supervisor来做进程守护它能保证进程死了立马拉起来绝不静默消失。; /etc/supervisor/conf.d/task-scheduler.conf [program:task-scheduler] commandphp /data/www/app/bin/scheduler.php --node1 autostarttrue autorestarttrue startsecs3 startretries10 stopasgrouptrue killasgrouptrue redirect_stderrtrue stdout_logfile/data/logs/task-scheduler.log stderr_logfile/data/logs/task-scheduler-error.log numprocs1 ; /etc/supervisor/conf.d/task-worker.conf [program:task-worker-email] commandphp /data/www/app/bin/worker.php --typeemail autostarttrue autorestarttrue startsecs3 startretries10 ; 同时启动8个进程实例各自独立消费 numprocs8 process_name%(program_name)s_%(process_num)02d redirect_stderrtrue stdout_logfile/data/logs/task-worker-email.log重点讲几个参数numprocsWorker按业务组配置想扩并发直接改这个数值然后supervisorctl update即可不需要重启整个服务。startsecs3进程启动3秒内退出不算“正常启动”这个参数防止“一拉起来又立刻崩”的死循环拖垮CPU。stopasgrouptrue/killasgrouptrue如果Worker里fork了子进程重启时会一并杀掉否则会留下孤儿进程继续消费队列等你发现的时候已经消费了一堆“不该消费”的任务。7.2 调度器的多节点部署一主一备怎么配调度器正常情况下一台就够但为了高可用必须部署多节点。多个节点同时跑Scheduler进程通过CAS更新天然解决派发重复问题所以你做高可用时不需要任何特殊配置只需要在每个节点上各开一个Scheduler进程即可。有个地方要注意多节点调度器实例之间时钟必须校准。CAS更新只解决“多个节点抢同一个任务”但如果两个节点的系统时钟偏差超过几秒任务可能被错派——A节点认为任务已经到期更新了状态B节点因为时钟慢还没看到这个任务没问题但如果B节点时钟快它可能在任务到期的前一分钟就派发了这个偏差是你排查“任务提前执行”时要考虑的方向。生产环境我直接用NTP同步偏差控制在一秒内问题基本消除。7.3 监控告警队列积压、执行成功率、心跳三件事必须盯最后是监控。没有监控的分布式调度系统等于在悬崖边开车不看仪表盘。我重点盯三个指标指标获取方式告警阈值队列积压数LLEN task:queue:*定时轮询超过1万持续5分钟任务执行成功率任务中心表按last_result统计成功率低于99%持续10分钟Worker心跳GET task:worker:alive:*过期超过1分钟队列积压用Redis自带的MONITOR轮询就行不用额外引入Prometheus。执行成功率直接在MySQL里跑个汇总查询。心跳检测用一个独立的PHP脚本每30秒扫一遍所有Worker的心跳键。这里有个小技巧监控脚本本身也要被Supervisor守护否则监控挂了你还不知道调度系统已经出事。我通常把监控脚本也挂到Supervisor底下用cron周期跑一旦某个指标触发阈值直接调企业微信机器人接口发告警。8. 最后说点实际体会整套系统从设计到落地我最大的感受是——分布式任务调度不是引入一个多复杂的框架而是把调度、执行、存储三个职责拆开在每个环节选最务实的方案。用MySQL做任务中心是因为它稳定、可靠、可审计用Redis做执行队列是因为它够快、够简单、够轻量调度器用CAS更新代替分布式锁是因为它少一个故障点、少一套需要维护的分布式锁客户端。我建议刚开始做这件事的团队从最小的闭环开始——一张任务表、一个调度进程、一个Worker进程先跑通投递-调度-执行-回写这四个环节再逐步加旁路扫描、分片队列、Supervisor守护、监控告警。不要一开始就追求各种开源框架的完整功能很多时候你要的根本不是那些功能而是一个能随着业务渐进演进的系统骨架。如果你现在也在被单机crontab的可靠性问题折磨不妨按这篇文章的流程试一版把调度器、Worker跑起来先把半夜爬起来手动跑脚本这个痛点解决掉后面的事情等遇到再说。