本文目录导读：
我将为您设计一个完整的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 界面来管理任务
安全考虑
- 权限控制：任务以最小权限运行
- 命令白名单：只允许执行预先定义的命令
- 超时处理：所有任务必须有明确的超时
- 资源限制：设置每个任务的内存和执行时间限制
- 审计日志：记录所有操作和更改
这个系统提供了完整的任务调度功能,包括并发执行、超时处理、重试机制、日志记录等核心功能，可以根据实际需求进行扩展。
版权声明：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直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
