一、分布式任务调度（类XXL-Job） composerrequireswoole/ide-helper predis/predis 架构：[调度中心 Scheduler]→ Redis →[执行器节点 Worker xN]↑ ↓ 定时触发 执行任务+上报结果 调度中心：<?php// scheduler.php - 单实例，负责触发任务require'vendor/autoload.php';useSwoole\Timer;usePredis\Client;$redis=newClient();// 注册任务定义$jobs=['order.timeout'=>['cron'=>'*/5 * * * *','route'=>'round_robin'],'report.daily'=>['cron'=>'0 2 * * *','route'=>'random'],'cache.warmup'=>['cron'=>'*/1 * * * *','route'=>'broadcast'],// 所有节点都执行];// 抢分布式锁（只有一个调度中心实例工作）functionacquireLock(Client$redis,string$key,int$ttl=30):bool{return(bool)$redis->set($key,gethostname(),'EX',$ttl,'NX');}functionmatchCron(string$cron):bool{[$min,$hour,$dom,$mon,$dow]=explode(' ',$cron);$t=getdate();$match=fn($expr,$val)=>$expr==='*'?true:(str_starts_with($expr,'*/')?$val%(int)substr($expr,2)===0:in_array($val,array_map('intval',explode(',',$expr))));return$match($min,$t['minutes'])&&$match($hour,$t['hours'])&&$match($dom,$t['mday'])&&$match($mon,$t['mon'])&&$match($dow,$t['wday']);}// 每分钟检查一次（生产用每秒检查）Timer::tick(60000,function()use($redis,$jobs){if(!acquireLock($redis,'scheduler:lock'))return;// 没抢到锁就跳过foreach($jobsas$jobName=>$job){if(!matchCron($job['cron']))continue;$taskId=uniqid($jobName.':',true);$task=json_encode(['id'=>$taskId,'job'=>$jobName,'route'=>$job['route'],'trigger'=>time(),]);match($job['route']){'broadcast'=>broadcastTask($redis,$jobName,$task),'round_robin'=>$redis->rpush("queue:worker:".nextWorker($redis),$task),default=>$redis->rpush("queue:worker:".randomWorker($redis),$task),};// 记录调度日志$redis->hset("job:log:$taskId",'status','dispatched','time',time());echo"[$taskId] 已调度$jobName\n";}});functionbroadcastTask(Client$redis,string$job,string$task):void{foreach($redis->smembers('workers:online')as$worker){$redis->rpush("queue:worker:$worker",$task);}}functionnextWorker(Client$redis):string{$workers=$redis->smembers('workers:online');$idx=(int)$redis->incr('scheduler:rr_index')%count($workers);return$workers[$idx];}functionrandomWorker(Client$redis):string{return$redis->srandmember('workers:online');}echo"调度中心启动\n";\Swoole\Event::wait();执行器节点（多实例）：<?php// worker.php - 多实例部署require'vendor/autoload.php';useSwoole\Process;usePredis\Client;$redis=newClient();$workerId=gethostname().':'.getmypid();// 注册自己上线$redis->sadd('workers:online',$workerId);$redis->expire('workers:online',30);// 注册任务处理器$handlers=['order.timeout'=>function(array$task):array{// 实际业务逻辑echo"处理订单超时任务\n";return['processed'=>42,'status'=>'ok'];},'report.daily'=>function(array$task):array{echo"生成日报\n";return['report_id'=>uniqid(),'status'=>'ok'];},'cache.warmup'=>function(array$task):array{echo"预热缓存\n";return['keys'=>1024,'status'=>'ok'];},];// 心跳（保持在线状态）\Swoole\Timer::tick(10000,fn()=>$redis->sadd('workers:online',$workerId)&&$redis->expire('workers:online',30));// 主循环：阻塞拉取任务echo"Worker$workerId启动\n";while(true){// blpop 阻塞等待，超时2秒重试$item=$redis->blpop("queue:worker:$workerId",2);if(!$item)continue;$task=json_decode($item[1],true);$job=$task['job'];echo"[{$task['id']}] 开始执行$job\n";$redis->hset("job:log:{$task['id']}",'status','running','worker',$workerId);try{$result=isset($handlers[$job])?($handlers[$job])($task):thrownew\RuntimeException("未知任务:$job");$redis->hset("job:log:{$task['id']}",'status','success','result',json_encode($result));echo"[{$task['id']}] 完成\n";}catch(\Throwable$e){$redis->hset("job:log:{$task['id']}",'status','failed','error',$e->getMessage());// 失败重试（最多3次）$retries=(int)$redis->hget("job:log:{$task['id']}",'retries');if($retries<3){$redis->hset("job:log:{$task['id']}",'retries',$retries+1);$redis->rpush("queue:worker:$workerId",json_encode($task));}echo"[{$task['id']}] 失败:{$e->getMessage()}\n";}}查询任务状态：<?php// 查日志$log=$redis->hgetall("job:log:$taskId");// ['status' => 'success', 'result' => '...', 'worker' => 'host:pid']大白话解释 分布式任务调度： 单机定时任务（crontab）： 一台服务器挂了 → 任务不跑了 任务太多 → 一台跑不完 分布式任务调度： 调度中心：只管"什么时候触发什么任务"执行器：只管"怎么跑任务"Redis：中间传话的 流程： 调度中心 → 到点了，order.timeout该跑了 → 塞进Redis队列 Worker1 → 从队列取出来跑 → 跑完上报结果 Worker挂了 → 其他Worker继续跑，不影响
