本文目录导读：
我来为您详细介绍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: 云端托管方案
这种分布式架构可以支持大规模的任务调度,确保高可用和可扩展性，选择方案时需要根据项目需求、团队技术栈和运维能力综合考虑。
版权声明：NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
