本文目录导读：
我将为您设计一个完整的PHP定时任务调度系统,这个系统将包含任务管理、调度执行、日志记录等核心功能。
系统架构
├── cron/
│ ├── config/
│ │ └── tasks.php # 任务配置文件
│ ├── lib/
│ │ ├── Logger.php # 日志系统
│ │ ├── TaskManager.php # 任务管理器
│ │ ├── Scheduler.php # 调度器
│ │ └── Task.php # 任务类
│ ├── logs/ # 日志目录
│ └── run.php # 入口脚本核心代码实现
1 任务类 (Task.php)
<?php
namespace CronSystem;
/**
* 任务实体类
*/
class Task
{
private $name;
private $command;
private $schedule;
private $timeout;
private $status;
private $lastRunTime;
private $lastRunResult;
private $retryCount;
private $failCount;
private $maxRetries;
public function __construct(array $config)
{
$this->name = $config['name'] ?? '';
$this->command = $config['command'] ?? '';
$this->schedule = $config['schedule'] ?? '* * * * *';
$this->timeout = $config['timeout'] ?? 300;
$this->status = 'pending';
$this->lastRunTime = null;
$this->lastRunResult = null;
$this->retryCount = 0;
$this->failCount = 0;
$this->maxRetries = $config['max_retries'] ?? 3;
}
// 检查任务是否应该在当前时间执行
public function isDue(\DateTime $currentTime): bool
{
return $this->cronExpressionMatches($currentTime);
}
// 解析cron表达式并检查是否匹配
private function cronExpressionMatches(\DateTime $time): bool
{
$parts = explode(' ', $this->schedule);
if (count($parts) !== 5) {
return false;
}
$minute = $time->format('i');
$hour = $time->format('G');
$day = $time->format('j');
$month = $time->format('n');
$weekday = $time->format('w');
return $this->expressionMatch($parts[0], (int)$minute) &&
$this->expressionMatch($parts[1], (int)$hour) &&
$this->expressionMatch($parts[2], (int)$day) &&
$this->expressionMatch($parts[3], (int)$month) &&
$this->expressionMatch($parts[4], (int)$weekday);
}
// 匹配单个cron字段
private function expressionMatch(string $expression, int $value): bool
{
if ($expression === '*') return true;
// 处理逗号分隔的多个值
if (strpos($expression, ',') !== false) {
$values = explode(',', $expression);
foreach ($values as $val) {
if ($this->expressionMatch(trim($val), $value)) return true;
}
return false;
}
// 处理范围值
if (strpos($expression, '-') !== false) {
list($start, $end) = explode('-', $expression);
return $value >= (int)$start && $value <= (int)$end;
}
// 处理步长值
if (strpos($expression, '/') !== false) {
list($base, $step) = explode('/', $expression);
if ($base === '*') $base = 0;
return ($value >= (int)$base) && (($value - (int)$base) % (int)$step === 0);
}
// 精确匹配
return (int)$expression === $value;
}
// Getters and Setters
public function getName() { return $this->name; }
public function getCommand() { return $this->command; }
public function getSchedule() { return $this->schedule; }
public function getTimeout() { return $this->timeout; }
public function getStatus() { return $this->status; }
public function setStatus($status) { $this->status = $status; }
public function setLastRunTime($time) { $this->lastRunTime = $time; }
public function setLastRunResult($result) { $this->lastRunResult = $result; }
public function incrementRetry() { $this->retryCount++; }
public function getRetryCount() { return $this->retryCount; }
public function getMaxRetries() { return $this->maxRetries; }
public function incrementFailCount() { $this->failCount++; }
public function getFailCount() { return $this->failCount; }
}
2 任务管理器 (TaskManager.php)
<?php
namespace CronSystem;
/**
* 任务管理器 - 负责加载和保存任务配置
*/
class TaskManager
{
private $tasks = [];
private $configFile;
private $stateFile;
public function __construct(string $configDir)
{
$this->configFile = $configDir . '/tasks.php';
$this->stateFile = $configDir . '/runtime_state.json';
$this->loadTasks();
$this->loadState();
}
// 加载任务配置
private function loadTasks(): void
{
if (file_exists($this->configFile)) {
$configs = require $this->configFile;
foreach ($configs as $config) {
$this->tasks[$config['name']] = new Task($config);
}
}
}
// 加载运行状态
private function loadState(): void
{
if (file_exists($this->stateFile)) {
$state = json_decode(file_get_contents($this->stateFile), true);
if (is_array($state)) {
foreach ($state as $name => $taskState) {
if (isset($this->tasks[$name])) {
$this->tasks[$name]->setStatus($taskState['status'] ?? 'pending');
$this->tasks[$name]->setLastRunTime($taskState['last_run_time'] ?? null);
if (isset($taskState['fail_count'])) {
// 这里简化为记录失败次数，完整实现需要更多字段
}
}
}
}
}
}
// 保存运行状态
public function saveState(): void
{
$state = [];
foreach ($this->tasks as $name => $task) {
$state[$name] = [
'status' => $task->getStatus(),
'last_run_time' => $task->getLastRunTime(),
];
}
file_put_contents($this->stateFile, json_encode($state, JSON_PRETTY_PRINT));
}
// 获取所有任务
public function getTasks(): array
{
return $this->tasks;
}
// 获取特定任务
public function getTask(string $name): ?Task
{
return $this->tasks[$name] ?? null;
}
// 查找需要执行的任务
public function findDueTasks(\DateTime $time): array
{
$dueTasks = [];
foreach ($this->tasks as $task) {
if ($task->isDue($time)) {
$dueTasks[] = $task;
}
}
return $dueTasks;
}
// 添加新任务
public function addTask(array $config): bool
{
if (isset($this->tasks[$config['name']])) {
return false;
}
$task = new Task($config);
$this->tasks[$config['name']] = $task;
$this->persistTasks();
return true;
}
// 更新任务
public function updateTask(string $name, array $config): bool
{
if (!isset($this->tasks[$name])) {
return false;
}
$this->tasks[$name] = new Task(array_merge(
['name' => $name],
$config
));
$this->persistTasks();
return true;
}
// 删除任务
public function deleteTask(string $name): bool
{
if (!isset($this->tasks[$name])) {
return false;
}
unset($this->tasks[$name]);
$this->persistTasks();
return true;
}
// 持久化任务配置
private function persistTasks(): void
{
$taskConfigs = [];
foreach ($this->tasks as $task) {
$taskConfigs[] = [
'name' => $task->getName(),
'command' => $task->getCommand(),
'schedule' => $task->getSchedule(),
'timeout' => $task->getTimeout(),
];
}
$content = "<?php return " . var_export($taskConfigs, true) . ";";
file_put_contents($this->configFile, $content);
}
}
3 调度器 (Scheduler.php)
<?php
namespace CronSystem;
/**
* 调度器 - 核心调度逻辑
*/
class Scheduler
{
private $taskManager;
private $logger;
private $isRunning = false;
private $processes = [];
public function __construct(TaskManager $taskManager, Logger $logger)
{
$this->taskManager = $taskManager;
$this->logger = $logger;
}
// 执行一次调度检查
public function run(): void
{
$currentTime = new \DateTime();
$this->logger->info("Scheduler check at: " . $currentTime->format('Y-m-d H:i:s'));
$dueTasks = $this->taskManager->findDueTasks($currentTime);
foreach ($dueTasks as $task) {
$this->executeTask($task);
}
$this->taskManager->saveState();
$this->cleanupFinishedProcesses();
}
// 执行单个任务
private function executeTask(Task $task): void
{
// 防止重复执行
if ($task->getStatus() === 'running') {
$this->logger->warning("Task {$task->getName()} already running, skipping");
return;
}
$this->logger->info("Starting task: {$task->getName()}");
// 设置任务状态
$task->setStatus('running');
$task->setLastRunTime(date('Y-m-d H:i:s'));
// 创建子进程执行任务
$processId = pcntl_fork();
if ($processId === -1) {
// fork失败
$this->logger->error("Failed to fork process for task: {$task->getName()}");
$task->setStatus('failed');
$task->incrementFailCount();
return;
} elseif ($processId === 0) {
// 子进程执行
try {
$this->executeCommand($task);
exit(0);
} catch (\Exception $e) {
$this->logger->error("Task execution failed: " . $e->getMessage());
exit(1);
}
} else {
// 父进程记录子进程信息
$this->processes[$processId] = [
'task' => $task,
'started_at' => time()
];
}
}
// 实际执行命令
private function executeCommand(Task $task): void
{
// 检查并清理子进程资源
$this->cleanupProcesses();
// 设置超时
$process = proc_open(
$task->getCommand(),
[
0 => ['file', '/dev/null', 'r'],
1 => ['file', '/tmp/task_output.txt', 'w'],
2 => ['file', '/tmp/task_error.txt', 'a']
],
$pipes
);
if (is_resource($process)) {
$status = proc_get_status($process);
$startTime = time();
// 等待进程完成或超时
while ($status['running']) {
usleep(100000); // 100ms
$status = proc_get_status($process);
// 检查超时
if ((time() - $startTime) > $task->getTimeout()) {
$this->logger->error("Task {$task->getName()} timed out");
proc_terminate($process);
$task->setStatus('timeout');
$task->incrementFailCount();
return;
}
}
// 进程完成
$exitCode = $status['exitcode'];
if ($exitCode === 0) {
$task->setStatus('success');
$task->setLastRunResult('Success');
} else {
$task->setStatus('failed');
$task->incrementFailCount();
$task->setLastRunResult('Failed with exit code: ' . $exitCode);
// 重试机制
if ($task->getRetryCount() < $task->getMaxRetries()) {
$this->retryTask($task);
}
}
proc_close($process);
// 记录完成时间
$this->logger->info(
"Task {$task->getName()} completed with status: {$task->getStatus()} , 耗时: " . (time() - $startTime) . "s"
);
}
}
// 重试机制
private function retryTask(Task $task): void
{
$this->logger->info("Retrying task: {$task->getName()} ({$task->getRetryCount()}/{$task->getMaxRetries()})");
$retryDelay = pow(2, $task->getRetryCount()) * 5; // 指数退避
sleep($retryDelay);
if ($task->isDue(new \DateTime())) {
$task->incrementRetry();
$this->executeTask($task);
}
}
// 清理已完成进程
private function cleanupProcesses(): void
{
foreach ($this->processes as $pid => $processInfo) {
$res = pcntl_waitpid($pid, $status, WNOHANG);
if ($res > 0 || $res === -1) {
unset($this->processes[$pid]);
}
}
}
// 清理所有资源
public function shutdown(): void
{
foreach ($this->processes as $pid => $processInfo) {
// 强制终止还在运行的进程
posix_kill($pid, SIGKILL);
}
$this->logger->info("Scheduler shutdown complete");
}
// 获取所有任务状态
public function getTaskStatuses(): array
{
$statuses = [];
$tasks = $this->taskManager->getTasks();
foreach ($tasks as $task) {
$statuses[$task->getName()] = [
'status' => $task->getStatus(),
'last_run' => $task->getLastRunTime(),
'last_result' => $task->getLastRunResult(),
'fail_count' => $task->getFailCount(),
];
}
return $statuses;
}
}
4 日志系统 (Logger.php)
<?php
namespace CronSystem;
/**
* 日志系统 - 支持分级日志和文件输出
*/
class Logger
{
private $logFile;
private $logLevel;
private $maxFileSize = 10485760; // 10MB
const LEVELS = [
'debug' => 100,
'info' => 200,
'warning' => 300,
'error' => 400,
'critical' => 500
];
public function __construct(string $logDir, string $level = 'info')
{
$this->logFile = $logDir . '/cron_' . date('Y-m-d') . '.log';
$this->logLevel = $level;
// 确保日志目录存在
if (!is_dir($logDir)) {
mkdir($logDir, 0755, true);
}
// 日志文件轮转
$this->rotateLogFile();
}
public function debug(string $message): void
{
$this->log('debug', $message);
}
public function info(string $message): void
{
$this->log('info', $message);
}
public function warning(string $message): void
{
$this->log('warning', $message);
}
public function error(string $message): void
{
$this->log('error', $message);
}
public function critical(string $message): void
{
$this->log('critical', $message);
}
private function log(string $level, string $message): void
{
// 检查日志级别
if (self::LEVELS[$level] < self::LEVELS[$this->logLevel]) {
return;
}
$timestamp = date('Y-m-d H:i:s');
$logMessage = sprintf(
"[%s] [%s] [PID:%d] %s\n",
$timestamp,
strtoupper($level),
getmypid(),
$message
);
file_put_contents($this->logFile, $logMessage, FILE_APPEND | LOCK_EX);
}
// 日志文件轮转
private function rotateLogFile(): void
{
if (file_exists($this->logFile) && filesize($this->logFile) >= $this->maxFileSize) {
$backupFile = $this->logFile . '.' . date('His');
rename($this->logFile, $backupFile);
}
}
// 获取日志内容
public function getLogs(int $lines = 100): array
{
if (!file_exists($this->logFile)) {
return [];
}
$content = file($this->logFile);
return array_slice($content, -$lines);
}
}
5 入口脚本 (run.php)
<?php
require_once __DIR__ . '/lib/Logger.php';
require_once __DIR__ . '/lib/Task.php';
require_once __DIR__ . '/lib/TaskManager.php';
require_once __DIR__ . '/lib/Scheduler.php';
use CronSystem\Logger;
use CronSystem\TaskManager;
use CronSystem\Scheduler;
// 基础配置
$configDir = __DIR__ . '/config';
$logDir = __DIR__ . '/logs';
// 初始化系统
$logger = new Logger($logDir, 'info');
$taskManager = new TaskManager($configDir);
$scheduler = new Scheduler($taskManager, $logger);
// 处理命令行参数
$options = getopt('h', ['help', 'run', 'list', 'status', 'add:', 'delete:', 'update:']);
if (isset($options['h']) || isset($options['help'])) {
echo <<<HELP
PHP Task Scheduler
Usage: php run.php [options]
Options:
--run Run scheduler once
--list List all tasks
--status Show task statuses
--add=JSON Add new task (synchronous)
--delete=NAME Delete task
--update=NAME=JSON Update task
--help Show this help
HELP;
exit(0);
}
// 主调度循环
if (isset($options['run'])) {
$logger->info("Starting scheduler in run-once mode");
// 信号处理
pcntl_async_signals(true);
pcntl_signal(SIGTERM, function() use ($scheduler, $logger) {
$logger->info("Received SIGTERM, shutting down...");
$scheduler->shutdown();
exit(0);
});
pcntl_signal(SIGINT, function() use ($scheduler, $logger) {
$logger->info("Received SIGINT, shutting down...");
$scheduler->shutdown();
exit(0);
});
// 执行调度
$scheduler->run();
// 可选：持续运行模式
$continuous = false;
if ($continuous) {
while (true) {
sleep(60);
$scheduler->run();
}
}
} elseif (isset($options['list'])) {
// 列出任务
$tasks = $taskManager->getTasks();
echo "Task Listing:\n";
echo "=============\n";
foreach ($tasks as $task) {
echo sprintf(
"Name: %-20s Schedule: %-15s Status: %-10s\n",
$task->getName(),
$task->getSchedule(),
$task->getStatus()
);
}
} elseif (isset($options['status'])) {
// 显示任务状态
echo "Task Status Report\n";
echo "==================\n";
$statuses = $scheduler->getTaskStatuses();
foreach ($statuses as $name => $status) {
echo "{$name}:\n";
echo " Status: {$status['status']}\n";
echo " Last Run: {$status['last_run']}\n";
echo " Result: {$status['last_result']}\n";
echo " Fail Count: {$status['fail_count']}\n";
echo " --------------------\n";
}
} elseif (isset($options['add'])) {
// 添加新任务
$config = json_decode($options['add'], true);
if ($taskManager->addTask($config)) {
echo "Task added successfully\n";
} else {
echo "Failed to add task\n";
}
} elseif (isset($options['delete'])) {
// 删除任务
$taskName = $options['delete'];
if ($taskManager->deleteTask($taskName)) {
echo "Task deleted successfully\n";
} else {
echo "Task not found or deletion failed\n";
}
} elseif (isset($options['update'])) {
// 更新任务
$parts = explode('=', $options['update'], 2);
$taskName = $parts[0];
$config = json_decode($parts[1], true);
if ($taskManager->updateTask($taskName, $config)) {
echo "Task updated successfully\n";
} else {
echo "Task update failed\n";
}
} else {
echo "Invalid usage. Use --help for more information.\n";
exit(1);
}
6 配置文件示例 (tasks.php)
<?php
return [
[
'name' => 'backup_database',
'command' => 'php /path/to/backup_script.php',
'schedule' => '0 2 * * *', // 每天凌晨2点
'timeout' => 600,
'max_retries' => 3,
],
[
'name' => 'clean_temp_files',
'command' => 'rm -rf /tmp/temp_*',
'schedule' => '*/30 * * * *', // 每30分钟
'timeout' => 60,
'max_retries' => 2,
],
[
'name' => 'update_indexes',
'command' => 'php /path/to/update_scripts.php --type=index',
'schedule' => '0 */12 * * *', // 每12小时
'timeout' => 1200,
'max_retries' => 5,
],
[
'name' => 'send_email_digest',
'command' => 'php /path/to/send_digest.php',
'schedule' => '0 8 * * 1-5', // 工作日早上8点
'timeout' => 300,
'max_retries' => 3,
],
[
'name' => 'process_webhooks',
'command' => 'php /path/to/process_webhooks.php',
'schedule' => '*/5 * * * *', // 每5分钟
'timeout' => 120,
'max_retries' => 5,
],
];
使用说明
1 安装和配置
- 将上述代码保存到对应的文件路径中
- 确保PHP环境支持
pcntl和posix扩展 - 在
config/tasks.php中配置你的定时任务
2 常用命令
# 运行一次调度检查
php run.php --run
# 查看任务列表
php run.php --list
# 添加新任务
php run.php --add='{"name":"test_task","command":"php script.php","schedule":"* * * * *"}'
# 删除任务
php run.php --delete=test_task
# 更新任务
php run.php --update='test_task={"schedule":"*/5 * * * *"}'
# 查看任务状态
php run.php --status
# 显示帮助
php run.php --help
3 使用 Cron 集成
在系统的 crontab 中添加以下配置：
# 每5分钟运行一次调度器 */5 * * * * /usr/bin/php /path/to/cron/run.php --run >> /var/log/cron.log 2>&1
监控和通知
1 添加邮件通知功能
<?php
namespace CronSystem;
class Notification
{
private $emailConfig;
public function __construct(array $config)
{
$this->emailConfig = $config;
}
public function sendTaskFailureNotification(Task $task, string $error): void
{
$subject = "Task Failed: {$task->getName()}";
$message = "Task '{$task->getName()}' failed at " . date('Y-m-d H:i:s');
$message .= "\nError: $error";
$message .= "\nCommand: " . $task->getCommand();
$this->sendEmail($subject, $message);
}
private function sendEmail(string $subject, string $body): void
{
$headers = "From: " . $this->emailConfig['from'] . "\r\n";
$headers .= "Content-type: text/html; charset=UTF-8\r\n";
mail($this->emailConfig['to'], $subject, $body, $headers);
}
}
性能优化建议
- 使用消息队列：对于大量任务，可以考虑使用 Redis 或 RabbitMQ 来管理任务队列
- 分布式执行：如果需要跨服务器执行，可以集成 Redis 作为分布式锁
- 监控警报：添加 Prometheus + Grafana 监控
- 数据库存储：将任务状态存储到 MySQL 以支持历史查询
- Web管理界面：开发一个简单的 PHP Web 界面来管理任务
安全考虑
- 权限控制：任务以最小权限运行
- 命令白名单：只允许执行预先定义的命令
- 超时处理：所有任务必须有明确的超时
- 资源限制：设置每个任务的内存和执行时间限制
- 审计日志：记录所有操作和更改
这个系统提供了完整的任务调度功能,包括并发执行、超时处理、重试机制、日志记录等核心功能，可以根据实际需求进行扩展。
