PHP 怎么定制调度

wen PHP项目 2026-08-04 205

本文目录导读：

PHP 怎么定制调度

使用 Cron 调度任务
使用 Laravel 任务调度
自定义调度器实现
使用 RabbitMQ / Redis 队列调度
完整的企业级调度系统
使用环境配置
最佳实践建议

在 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']
                );
            }
        }
    }
}

选择合适的调度方案取决于：