本文目录导读：
我来为您详细介绍PHP分布式定时任务调度的完整方案。
1 核心问题
- 单机定时任务存在单点故障
- 任务重复执行问题
- 任务分发与负载均衡
- 任务状态监控与管理
2 主流解决方案
graph TB
A[分布式定时任务方案] --> B[基于消息队列]
A --> C[基于Redis]
A --> D[开源框架]
A --> E[云服务]
B --> B1[RabbitMQ延迟队列]
B --> B2[Kafka定时消息]
C --> C1[Redis ZSet]
C --> C2[Redis Keyspace]
D --> D1[XXL-JOB]
D --> D2[Elastic-Job]
D --> D3[Swoole框架]
E --> E1[阿里云SchedulerX]
E --> E2[腾讯云TCT]
完整实现方案
1 基于Redis的分布式调度器
<?php
/**
* Redis分布式任务调度器
*/
class RedisScheduler
{
private $redis;
private $taskKey = 'scheduler:tasks';
private $taskPrefix = 'scheduler:task:';
private $lockPrefix = 'scheduler:lock:';
private $nodeId;
public function __construct($redisConfig, $nodeId)
{
$this->redis = new Redis();
$this->redis->connect($redisConfig['host'], $redisConfig['port']);
$this->redis->auth($redisConfig['password'] ?? '');
$this->redis->select($redisConfig['database'] ?? 0);
$this->nodeId = $nodeId;
}
/**
* 添加定时任务
*/
public function addTask($taskName, $cronExpr, $callback, $params = [])
{
$taskId = md5($taskName . uniqid());
$task = [
'id' => $taskId,
'name' => $taskName,
'cron' => $cronExpr,
'callback' => $callback,
'params' => $params,
'status' => 'active',
'created_at' => time()
];
// 存储任务详情
$this->redis->hMset($this->taskPrefix . $taskId, $task);
// 计算下次执行时间
$nextRunTime = $this->getNextRunTime($cronExpr);
// 加入有序集合，score为下次执行时间
$this->redis->zAdd($this->taskKey, $nextRunTime, $taskId);
return $taskId;
}
/**
* 调度器主循环
*/
public function run()
{
echo "调度器节点 {$this->nodeId} 启动\n";
while (true) {
$this->processDueTasks();
usleep(100000); // 100ms间隔
}
}
/**
* 处理到期任务
*/
private function processDueTasks()
{
$now = microtime(true) * 1000;
// 获取所有到期的任务ID
$dueTasks = $this->redis->zRangeByScore($this->taskKey, 0, $now);
foreach ($dueTasks as $taskId) {
// 尝试获取分布式锁，避免重复执行
$lockKey = $this->lockPrefix . $taskId;
if ($this->acquireLock($lockKey, $taskId)) {
try {
// 从有序集合移除
$this->redis->zRem($this->taskKey, $taskId);
// 执行任务
$this->executeTask($taskId);
// 计算下次执行时间
$taskInfo = $this->redis->hGetAll($this->taskPrefix . $taskId);
if ($taskInfo && $taskInfo['status'] === 'active') {
$nextRunTime = $this->getNextRunTime($taskInfo['cron']);
$this->redis->zAdd($this->taskKey, $nextRunTime, $taskId);
}
} finally {
// 释放锁
$this->releaseLock($lockKey, $taskId);
}
}
}
}
/**
* 获取分布式锁
*/
private function acquireLock($lockKey, $taskId, $timeout = 10)
{
$token = $this->nodeId . ':' . uniqid();
$result = $this->redis->set($lockKey, $token, ['NX', 'EX' => $timeout]);
if ($result) {
// 设置锁的值，用于确认
$this->redis->set($lockKey . ':token', $token, ['EX' => $timeout]);
return true;
}
return false;
}
/**
* 释放锁
*/
private function releaseLock($lockKey, $taskId)
{
$token = $this->redis->get($lockKey . ':token');
$luaScript = <<<LUA
if redis.call("get", KEYS[1]) == ARGV[1] then
return redis.call("del", KEYS[1])
else
return 0
end
LUA;
$this->redis->eval($luaScript, [$lockKey, $token], 1);
$this->redis->del($lockKey . ':token');
}
/**
* 计算Cron表达式下次执行时间
*/
private function getNextRunTime($cronExpr)
{
$cron = CronExpression::factory($cronExpr);
$nextRun = $cron->getNextRunDate();
return $nextRun->getTimestamp() * 1000;
}
/**
* 执行任务（异步）
*/
private function executeTask($taskId)
{
$taskInfo = $this->redis->hGetAll($this->taskPrefix . $taskId);
// 发送到消息队列异步执行
$this->sendToQueue($taskId, $taskInfo);
}
private function sendToQueue($taskId, $taskInfo)
{
// 这里可以对接 RabbitMQ/Kafka 等
$event = new TaskEvent([
'task_id' => $taskId,
'params' => $taskInfo['params'] ?? [],
'callback' => $taskInfo['callback'] ?? ''
]);
// 使用异步进程或消息队列
event($event);
}
}
2 Cron表达式解析器
<?php
/**
* Cron表达式解析器
*/
class CronExpression
{
private $minutes = [];
private $hours = [];
private $days = [];
private $months = [];
private $weekdays = [];
public static function factory($expression)
{
$instance = new self($expression);
return $instance;
}
public function __construct($expression)
{
$parts = preg_split('/\s+/', trim($expression));
if (count($parts) !== 5) {
throw new InvalidArgumentException('Cron表达式格式错误');
}
list($minute, $hour, $day, $month, $weekday) = $parts;
$this->minutes = $this->parseField($minute, 0, 59);
$this->hours = $this->parseField($hour, 0, 23);
$this->days = $this->parseField($day, 1, 31);
$this->months = $this->parseField($month, 1, 12);
$this->weekdays = $this->parseField($weekday, 0, 6);
}
private function parseField($field, $min, $max)
{
$values = [];
if ($field === '*') {
return range($min, $max);
}
// 处理逗号分隔
foreach (explode(',', $field) as $part) {
if (strpos($part, '/') !== false) {
list($range, $step) = explode('/', $part);
$start = $range === '*' ? $min : intval($range);
for ($i = $start; $i <= $max; $i += intval($step)) {
$values[] = $i;
}
} elseif (strpos($part, '-') !== false) {
list($start, $end) = explode('-', $part);
$values = array_merge($values, range(intval($start), intval($end)));
} else {
$values[] = intval($part);
}
}
sort($values);
return array_unique($values);
}
public function getNextRunDate()
{
$current = new DateTime();
$current->setTime($current->format('H'), $current->format('i'), 0);
for ($i = 0; $i < 1440; $i++) { // 24小时搜索
$date = clone $current;
$date->modify("+{$i} minutes");
if ($this->matches($date)) {
return $date;
}
}
throw new RuntimeException('找不到下次执行时间');
}
private function matches(DateTime $date)
{
return in_array(intval($date->format('i')), $this->minutes) &&
in_array(intval($date->format('H')), $this->hours) &&
in_array(intval($date->format('d')), $this->days) &&
in_array(intval($date->format('m')), $this->months) &&
in_array(intval($date->format('w')), $this->weekdays);
}
}
3 分布式任务执行器
<?php
/**
* 分布式任务Worker
*/
class TaskWorker
{
private $queue;
private $processor;
private $logger;
private $maxRetries = 3;
public function __construct($queue, $processor, $logger)
{
$this->queue = $queue;
$this->processor = $processor;
$this->logger = $logger;
}
/**
* 启动Worker
*/
public function start($workerCount = 4)
{
echo "启动 {$workerCount} 个Worker进程\n";
for ($i = 0; $i < $workerCount; $i++) {
$pid = pcntl_fork();
if ($pid === -1) {
throw new RuntimeException('进程创建失败');
} elseif ($pid === 0) {
// 子进程处理任务
$this->processLoop($i);
exit(0);
}
}
// 主进程等待
while (true) {
$status = 0;
pcntl_wait($status);
}
}
/**
* 任务处理循环
*/
private function processLoop($workerId)
{
echo "Worker {$workerId} 启动\n";
while (true) {
try {
// 从队列获取任务
$task = $this->queue->pop('task_queue', 5);
if ($task) {
$this->processTask($task, $workerId);
}
usleep(100000); // 100ms
} catch (Exception $e) {
$this->logger->error('任务处理异常: ' . $e->getMessage());
usleep(1000000); // 1s
}
}
}
/**
* 处理单个任务
*/
private function processTask($task, $workerId)
{
$taskId = $task['task_id'];
$callback = $task['callback'];
$params = $task['params'];
echo "Worker {$workerId} 处理任务: {$taskId}\n";
for ($attempt = 1; $attempt <= $this->maxRetries; $attempt++) {
try {
// 执行任务
$result = $this->executeCallback($callback, $params);
// 记录执行日志
$this->logger->info("任务 {$taskId} 执行成功", [
'worker' => $workerId,
'attempt' => $attempt,
'result' => $result
]);
return true;
} catch (Exception $e) {
$this->logger->error("任务 {$taskId} 执行失败", [
'worker' => $workerId,
'attempt' => $attempt,
'error' => $e->getMessage()
]);
if ($attempt < $this->maxRetries) {
// 退避重试
sleep($attempt * 5);
}
}
}
// 任务最终失败，发送告警
$this->sendAlert($taskId, "任务执行失败超过最大重试次数");
return false;
}
/**
* 执行回调函数
*/
private function executeCallback($callback, $params)
{
if (is_callable($callback)) {
return call_user_func($callback, $params);
}
if (strpos($callback, '@') !== false) {
list($class, $method) = explode('@', $callback);
$instance = new $class();
return $instance->$method($params);
}
throw new RuntimeException("无效的回调方式");
}
}
4 使用XXL-JOB方案
<?php
/**
* XXL-JOB PHP客户端
*/
class XxlJobHandler
{
private $token;
private $adminAddress;
private $executorId;
/**
* 配置处理器
*/
public function registerHandler($jobId, $handler)
{
$handlerMap[$jobId] = $handler;
return $this;
}
/**
* 执行Job处理逻辑
*/
public function execute($jobId, $jobParam)
{
$jobInfo = $this->getJobInfo($jobId);
switch ($jobInfo['type']) {
case 'bean':
return $this->executeBeanJob($jobInfo, $jobParam);
case 'script':
return $this->executeScriptJob($jobInfo, $jobParam);
case 'http':
return $this->executeHttpJob($jobInfo, $jobParam);
default:
throw new RuntimeException("未知任务类型");
}
}
/**
* Bean类型任务
*/
private function executeBeanJob($jobInfo, $jobParam)
{
$handler = $this->handlerMap[$jobInfo['handler']] ?? null;
if (!$handler) {
throw new RuntimeException("处理器未注册");
}
return $handler($jobParam);
}
/**
* HTTP回调方式
*/
public function httpCallback($url, $params, $timeout = 30)
{
$ch = curl_init();
curl_setopt_array($ch, [
CURLOPT_URL => $url,
CURLOPT_POST => true,
CURLOPT_POSTFIELDS => json_encode($params),
CURLOPT_RETURNTRANSFER => true,
CURLOPT_TIMEOUT => $timeout,
CURLOPT_HTTPHEADER => [
'Content-Type: application/json'
]
]);
$result = curl_exec($ch);
$error = curl_error($ch);
curl_close($ch);
if ($error) {
throw new RuntimeException("请求失败：{$error}");
}
return json_decode($result, true);
}
}
5 监控与管理
<?php
/**
* 任务监控与管理
*/
class TaskMonitor
{
private $redis;
private $logger;
/**
* 记录任务执行统计
*/
public function recordExecution($taskId, $status, $data = [])
{
$key = "task:stats:{$taskId}";
$this->redis->hIncrBy($key, 'total', 1);
$this->redis->hIncrBy($key, $status, 1);
$this->redis->hSet($key, 'last_run', time());
if ($status == 'success') {
$this->redis->hSet($key, 'last_success', time());
} else {
$this->redis->hSet($key, 'last_error', json_encode($data));
}
// 更新每日统计
$dailyKey = "task:daily:{$taskId}:" . date('Y-m-d');
$this->redis->hIncrBy($dailyKey, 'total', 1);
$this->redis->hIncrBy($dailyKey, $status, 1);
}
/**
* 发送告警
*/
public function sendAlert($taskId, $message, $level = 'warning')
{
$alert = [
'task_id' => $taskId,
'message' => $message,
'level' => $level,
'time' => date('Y-m-d H:i:s')
];
// 优先使用钉钉/企业微信
if (class_exists('DingTalk')) {
DingTalk::send($alert);
}
// 邮件告警备用
$this->logger->alert(json_encode($alert));
}
/**
* 获取任务健康状态
*/
public function getTaskHealth()
{
$tasks = $this->redis->keys('task:stats:*');
$health = [];
foreach ($tasks as $task) {
$id = str_replace('task:stats:', '', $task);
$stats = $this->redis->hGetAll($task);
$lastRun = $stats['last_run'] ?? 0;
$age = time() - $lastRun;
$health[$id] = [
'total' => $stats['total'] ?? 0,
'success' => $stats['success'] ?? 0,
'failed' => $stats['failed'] ?? 0,
'last_run' => $lastRun,
'last_success' => $stats['last_success'] ?? 0,
'status' => $age > 3600 ? 'warning' : 'ok'
];
}
return $health;
}
}
6 启动脚本
<?php
/**
* 分布式调度系统启动入口
*/
class SchedulerApplication
{
private $config;
public function __construct($config)
{
$this->config = $config;
$this->init();
}
private function init()
{
date_default_timezone_set('Asia/Shanghai');
require_once 'vendor/autoload.php';
}
/**
* 启动调度器
*/
public function runScheduler()
{
$nodeId = gethostname() . ':' . getmypid();
$scheduler = new RedisScheduler($this->config['redis'], $nodeId);
// 注册任务
$this->registerTasks($scheduler);
// 启动调度
$scheduler->run();
}
/**
* 启动Worker
*/
public function runWorker()
{
$worker = new TaskWorker(
new QueueClient($this->config['queue']),
new TaskProcessor(),
new Logger($this->config['log'])
);
$worker->start($this->config['worker_count'] ?? 4);
}
/**
* 注册系统任务
*/
private function registerTasks($scheduler)
{
// 数据清理任务
$scheduler->addTask(
'cleanup_logs',
'0 2 * * *',
function() {
LogCleaner::cleanup(7);
}
);
// 报表生成任务
$scheduler->addTask(
'generate_report',
'0 6 * * *',
'ReportGenerator@generateDaily',
['format' => 'pdf', 'send_email' => true]
);
// 缓存刷新任务
$scheduler->addTask(
'refresh_cache',
'*/10 * * * *',
'CacheRefresher@refresh',
[]
);
}
}
部署架构
# Process Manager 配置 ; /etc/supervisor/conf.d/scheduler.conf [program:php_scheduler] command=php /var/www/scheduler/bin/scheduler.php directory=/var/www/scheduler user=www-data numprocs=1 autostart=true autorestart=true startsecs=10 stdout_logfile=/var/log/scheduler/scheduler.log [program:php_worker] command=php /var/www/scheduler/bin/worker.php directory=/var/www/scheduler user=www-data numprocs=4 autostart=true autorestart=true startsecs=10 stdout_logfile=/var/log/scheduler/worker.log
最佳实践
1 任务设计原则
- 幂等性：任务必须支持重复执行
- 超时控制：设置合理的执行超时
- 失败重试：带退避策略的重试机制
- 任务分级：按优先级分配资源
2 性能优化
- 使用消息队列异步处理
- 批量处理提高效率
- 合理设置Worker数量
- 避免长任务阻塞
3 监控告警
- 任务失败实时告警
- 任务执行延迟监控
- 资源使用率监控
- 分钟级心跳检测
其他开源方案推荐
- XXL-JOB: 功能丰富的分布式任务平台
- Elastic-Job: 基于Zookeeper的分布式调度
- Swoole Framework: 高性能PHP框架，适合长进程任务
- Workerman: 高性能PHP Socket框架
- 阿里云SchedulerX: 云端托管方案
这种分布式架构可以支持大规模的任务调度,确保高可用和可扩展性，选择方案时需要根据项目需求、团队技术栈和运维能力综合考虑。
