本文目录导读：
在 PHP 中定制调度（Scheduling）通常有以下几种场景和方法,我来详细说明：
使用 Cron 调度任务
基础 Cron 配置
// crontab 配置示例 * * * * * php /path/to/your/script.php
使用第三方库（推荐）
composer require dragonmantank/cron-expression
<?php
use Cron\CronExpression;
// 解析 Cron 表达式
$cron = CronExpression::factory('*/5 * * * *');
// 检查是否该执行
if ($cron->isDue()) {
// 执行任务
echo "任务执行中...";
}
// 获取下次运行时间
$nextRun = $cron->getNextRunDate();
echo "下次运行时间: " . $nextRun->format('Y-m-d H:i:s');
使用 Laravel 任务调度
定义调度任务
// app/Console/Kernel.php
protected function schedule(Schedule $schedule)
{
// 基础调度
$schedule->command('emails:send')
->daily()
->at('08:00')
->timezone('Asia/Shanghai');
// 复杂调度
$schedule->call(function () {
// 执行任务代码
DB::table('users')->delete();
})->everyMinute();
// 带条件的调度
$schedule->command('backup:run')
->weekly()
->when(function () {
return config('app.backup_enabled');
});
// 任务重叠保护
$schedule->command('report:generate')
->everyFiveMinutes()
->withoutOverlapping(30);
}
自定义调度器实现
简单调度类
<?php
class TaskScheduler {
private $tasks = [];
public function addTask($name, callable $callback, $schedule) {
$this->tasks[] = [
'name' => $name,
'callback' => $callback,
'schedule' => $schedule
];
}
public function run() {
foreach ($this->tasks as $task) {
if ($this->shouldRun($task['schedule'])) {
try {
$start = microtime(true);
call_user_func($task['callback']);
$this->logExecution($task['name'], microtime(true) - $start);
} catch (Exception $e) {
$this->logError($task['name'], $e->getMessage());
}
}
}
}
private function shouldRun($schedule) {
// 检查时间是否匹配
$now = new DateTime();
if (isset($schedule['time'])) {
$expectedTime = new DateTime($schedule['time']);
if ($now->format('H:i') !== $expectedTime->format('H:i')) {
return false;
}
}
if (isset($schedule['interval'])) {
$interval = $schedule['interval'];
$lastRun = $this->getLastRun(); // 从数据库获取上次运行时间
if ($lastRun && ($now->getTimestamp() - $lastRun) < $interval) {
return false;
}
}
return true;
}
private function logExecution($taskName, $duration) {
// 记录日志
error_log("[{$taskName}] 执行完成，耗时: {$duration}秒");
}
}
// 使用示例
$scheduler = new TaskScheduler();
$scheduler->addTask('清理缓存', function() {
cache()->flush();
}, ['interval' => 3600]);
$scheduler->addTask('发送邮件', function() {
// 发送邮件逻辑
}, ['time' => '09:00']);
使用 RabbitMQ / Redis 队列调度
Redis 队列调度
<?php
class RedisScheduler {
private $redis;
private $scheduleKey;
public function __construct(Redis $redis, $scheduleKey) {
$this->redis = $redis;
$this->scheduleKey = $scheduleKey;
}
public function schedule($delay, $callback) {
$task = [
'callback' => serialize($callback),
'execute_at' => time() + $delay
];
$this->redis->zAdd($this->scheduleKey, $task['execute_at'], json_encode($task));
}
public function process() {
while (true) {
$tasks = $this->redis->zRangeByScore($this->scheduleKey, 0, time());
foreach ($tasks as $taskData) {
$task = json_decode($taskData, true);
// 执行任务
$callback = unserialize($task['callback']);
$callback();
// 移除已执行的任务
$this->redis->zRem($this->scheduleKey, $taskData);
}
sleep(1);
}
}
}
完整的企业级调度系统
<?php
interface ScheduleInterface {
public function getNextRun(\DateTime $currentTime): \DateTime;
public function isDue(\DateTime $time): bool;
}
class CronSchedule implements ScheduleInterface {
private $expression;
private $parser;
public function __construct($expression) {
$this->parser = new \Cron\CronExpression($expression);
$this->expression = $expression;
}
public function getNextRun(\DateTime $currentTime): \DateTime {
return $this->parser->getNextRunDate($currentTime);
}
public function isDue(\DateTime $time): bool {
return $this->parser->isDue($time);
}
}
class TaskManager {
private $tasks = [];
private $storage;
private $logger;
public function __construct($storage, $logger) {
$this->storage = $storage;
$this->logger = $logger;
}
public function register($name, $task, ScheduleInterface $schedule, array $options = []) {
$this->tasks[$name] = [
'task' => $task,
'schedule' => $schedule,
'options' => $options
];
}
public function run() {
$currentTime = new DateTime();
foreach ($this->tasks as $name => $config) {
try {
if (!$this->shouldRun($name, $config, $currentTime)) {
continue;
}
$this->executeTask($name, $config);
} catch (Exception $e) {
$this->logger->error("任务 {$name} 执行失败: {$e->getMessage()}");
}
}
}
private function shouldRun($name, $config, $currentTime): bool {
$schedule = $config['schedule'];
$lastRun = $this->storage->getLastRun($name);
if (!$lastRun) {
return true;
}
// 检查是否满足调度条件
$nextRun = $schedule->getNextRun($lastRun);
return $currentTime >= $nextRun;
}
private function executeTask($name, $config) {
$startTime = microtime(true);
// 添加任务锁防止重复执行
if (isset($config['options']['lock']) && $config['options']['lock']) {
$lockKey = "task_lock_{$name}";
if (!$this->acquireLock($lockKey)) {
return;
}
}
try {
$result = call_user_func($config['task']);
// 记录执行状态
$this->storage->setLastRun($name, new DateTime());
$this->storage->setLastResult($name, $result);
$duration = microtime(true) - $startTime;
$this->logger->info(
"任务 {$name} 执行成功",
[
'duration' => $duration,
'result' => $result
]
);
} finally {
// 释放锁
if (isset($config['options']['lock']) && $config['options']['lock']) {
$this->releaseLock($lockKey);
}
}
}
private function acquireLock($key): bool {
// 使用 Redis 实现分布式锁
return $this->redis->set($key, 1, ['NX', 'EX' => 60]);
}
private function releaseLock($key) {
$this->redis->del($key);
}
}
使用环境配置
// config/scheduler.php
return [
'default_timezone' => 'Asia/Shanghai',
'tasks' => [
[
'name' => 'send_notifications',
'schedule' => '*/5 * * * *',
'callback' => [NotificationService::class, 'sendPending'],
'params' => ['limit' => 100]
],
[
'name' => 'cleanup_logs',
'schedule' => '0 3 * * *',
'callback' => [LogService::class, 'cleanup'],
'params' => ['days_old' => 30]
]
],
'locking' => [
'enabled' => true,
'driver' => 'redis'
],
'monitoring' => [
'enabled' => true,
'metrics' => ['execution_time', 'memory_usage'],
'alerts' => ['email' => 'admin@example.com']
]
];
最佳实践建议
// 1. 任务状态记录
class TaskStatusMiddleware {
public function __invoke($task, callable $next) {
$startTime = microtime(true);
$taskId = uniqid('task_');
try {
$result = $next();
$this->recordSuccess($taskId, $result, microtime(true) - $startTime);
return $result;
} catch (\Exception $e) {
$this->recordFailure($taskId, $e, microtime(true) - $startTime);
throw $e;
}
}
}
// 2. 动态任务注册
class DynamicScheduler {
public function loadTasksFromConfig() {
$configs = config('scheduler.tasks');
foreach ($configs as $config) {
if ($this->shouldRegister($config)) {
$this->registerTask(
$config['name'],
$config['callback'],
$config['schedule']
);
}
}
}
}
选择合适的调度方案取决于：
- 项目复杂度
- 部署环境（是否有 Cron）
- 是否需要实时调度
- 可用基础设施（Redis、DB 等）
- 监控和调试需求
推荐从简单的 Cron 或 Laravel 调度开始,随着需求增长再升级到复杂的队列调度系统。
