本文目录导读：
我来为您提供一个完整的PHP爬虫任务调度解决方案,包含多种调度方式和实战示例。
基础架构设计
1 任务调度器核心类
<?php
// TaskScheduler.php
namespace Crawler\Scheduler;
use Exception;
use Redis;
use PDO;
class TaskScheduler
{
private $redis;
private $pdo;
private $config;
public function __construct($config)
{
$this->config = $config;
$this->initRedis();
$this->initDatabase();
}
/**
* 初始化Redis连接
*/
private function initRedis()
{
$this->redis = new Redis();
$this->redis->connect(
$this->config['redis']['host'],
$this->config['redis']['port']
);
if (!empty($this->config['redis']['password'])) {
$this->redis->auth($this->config['redis']['password']);
}
$this->redis->select($this->config['redis']['db']);
}
/**
* 初始化数据库连接
*/
private function initDatabase()
{
$dsn = sprintf(
"mysql:host=%s;dbname=%s;charset=utf8mb4",
$this->config['database']['host'],
$this->config['database']['dbname']
);
$this->pdo = new PDO($dsn, $this->config['database']['user'], $this->config['database']['password']);
$this->pdo->setAttribute(PDO::ATTR_ERRMODE, PDO::ERRMODE_EXCEPTION);
}
/**
* 添加任务
*/
public function addTask($taskName, $callback, $interval = 60, $priority = 5)
{
$taskId = uniqid('task_', true);
$taskData = [
'id' => $taskId,
'name' => $taskName,
'callback' => $callback,
'interval' => $interval,
'priority' => $priority,
'last_run' => 0,
'status' => 'pending'
];
// 保存到Redis队列
$this->redis->hSet('crawler:tasks', $taskId, json_encode($taskData));
return $taskId;
}
/**
* 获取待执行的任务
*/
public function getPendingTasks()
{
$tasks = $this->redis->hGetAll('crawler:tasks');
$pendingTasks = [];
foreach ($tasks as $taskId => $taskJson) {
$task = json_decode($taskJson, true);
if ($task['status'] === 'pending' ||
(time() - $task['last_run']) >= $task['interval']) {
$pendingTasks[] = $task;
}
}
// 按优先级排序
usort($pendingTasks, function($a, $b) {
return $a['priority'] <=> $b['priority'];
});
return $pendingTasks;
}
/**
* 运行任务
*/
public function runTask($task)
{
$taskId = $task['id'];
// 标记任务为运行中
$this->markTaskRunning($taskId);
try {
// 执行任务回调
$result = call_user_func($task['callback']);
// 更新任务状态
$this->markTaskCompleted($taskId, $result);
return ['success' => true, 'data' => $result];
} catch (Exception $e) {
// 记录错误
$this->markTaskFailed($taskId, $e->getMessage());
return ['success' => false, 'error' => $e->getMessage()];
}
}
private function markTaskRunning($taskId)
{
$task = $this->getTask($taskId);
$task['status'] = 'running';
$this->redis->hSet('crawler:tasks', $taskId, json_encode($task));
}
private function markTaskCompleted($taskId, $result)
{
$task = $this->getTask($taskId);
$task['status'] = 'completed';
$task['last_run'] = time();
$task['last_result'] = $result;
$this->redis->hSet('crawler:tasks', $taskId, json_encode($task));
// 保存执行日志
$this->saveTaskLog($taskId, 'success', $result);
}
private function markTaskFailed($taskId, $error)
{
$task = $this->getTask($taskId);
$task['status'] = 'failed';
$task['last_error'] = $error;
$this->redis->hSet('crawler:tasks', $taskId, json_encode($task));
// 保存错误日志
$this->saveTaskLog($taskId, 'error', $error);
}
public function getTask($taskId)
{
$taskJson = $this->redis->hGet('crawler:tasks', $taskId);
return $taskJson ? json_decode($taskJson, true) : null;
}
private function saveTaskLog($taskId, $status, $message)
{
$sql = "INSERT INTO task_logs (task_id, status, message, created_at)
VALUES (?, ?, ?, NOW())";
$stmt = $this->pdo->prepare($sql);
$stmt->execute([$taskId, $status, $message]);
}
}
定时调度实现
1 Cron定时调度
<?php
// cron_scheduler.php
require_once 'TaskScheduler.php';
class CronScheduler
{
private $scheduler;
private $taskConfigs = [];
public function __construct($config)
{
$this->scheduler = new TaskScheduler($config);
}
/**
* 注册定时任务配置
*/
public function registerTask($taskName, $cronExpression, $callback)
{
$this->taskConfigs[] = [
'name' => $taskName,
'cron' => $cronExpression,
'callback' => $callback
];
}
/**
* 检查并运行到期任务
*/
public function checkAndRun()
{
foreach ($this->taskConfigs as $taskConfig) {
if ($this->isDue($taskConfig['cron'])) {
$this->scheduler->runTask($taskConfig);
}
}
}
/**
* 解析Cron表达式（简化版）
*/
private function isDue($cronExpression)
{
$parts = explode(' ', trim($cronExpression));
if (count($parts) != 5) {
return false;
}
list($minute, $hour, $day, $month, $weekday) = $parts;
$now = time();
return $this->matchField($minute, date('i', $now)) &&
$this->matchField($hour, date('H', $now)) &&
$this->matchField($day, date('j', $now)) &&
$this->matchField($month, date('n', $now)) &&
$this->matchField($weekday, date('w', $now));
}
private function matchField($pattern, $value)
{
if ($pattern == '*') {
return true;
}
if (strpos($pattern, '/') !== false) {
list($range, $step) = explode('/', $pattern);
if ($range == '*') {
$range = '0-59';
}
list($start, $end) = explode('-', $range);
return ($value >= $start && $value <= $end && ($value - $start) % $step == 0);
}
if (strpos($pattern, '-') !== false) {
list($start, $end) = explode('-', $pattern);
return ($value >= $start && $value <= $end);
}
if (strpos($pattern, ',') !== false) {
$values = explode(',', $pattern);
return in_array($value, $values);
}
return $value == $pattern;
}
}
// 使用示例
$config = require 'config.php';
$scheduler = new CronScheduler($config);
// 注册定时爬虫任务
$scheduler->registerTask('crawl_news', '*/30 * * * *', function() {
$crawler = new NewsCrawler();
return $crawler->crawl();
});
$scheduler->registerTask('crawl_products', '0 */2 * * *', function() {
$crawler = new ProductCrawler();
return $crawler->crawl();
});
// 执行调度检查
$scheduler->checkAndRun();
2 PHP脚本无限循环调度
<?php
// loop_scheduler.php
class LoopScheduler
{
private $scheduler;
private $maxExecutionTime;
private $delayBetweenTasks;
public function __construct($config)
{
$this->scheduler = new TaskScheduler($config);
$this->maxExecutionTime = $config['scheduler']['max_execution_time'] ?? 60;
$this->delayBetweenTasks = $config['scheduler']['delay_between_tasks'] ?? 5;
}
/**
* 无限循环执行
*/
public function runForever()
{
$startTime = time();
while (true) {
// 检查最大执行时间
if ((time() - $startTime) > $this->maxExecutionTime) {
log_message("达到最大执行时间，退出循环");
break;
}
// 获取待执行任务
$tasks = $this->scheduler->getPendingTasks();
foreach ($tasks as $task) {
$this->scheduler->runTask($task);
sleep($this->delayBetweenTasks);
}
// 防止占用过多CPU
usleep(100000); // 0.1秒
}
}
/**
* 带退避重试的无限循环
*/
public function runWithBackoff()
{
$retryCount = 0;
$maxRetries = 5;
while (true) {
try {
$this->runForever();
$retryCount = 0; // 重置重试计数
} catch (Exception $e) {
$retryCount++;
if ($retryCount > $maxRetries) {
log_error("连续失败{$maxRetries}次，停止调度器");
break;
}
// 指数退避
$backoffTime = pow(2, $retryCount) * 5;
log_warning("调度器异常，{$backoffTime}秒后重试");
sleep($backoffTime);
}
}
}
}
多进程调度
1 基于PCNTL的多进程调度
<?php
// multi_process_scheduler.php
class MultiProcessScheduler
{
private $scheduler;
private $processLimit;
private $childProcesses = [];
public function __construct($config)
{
$this->scheduler = new TaskScheduler($config);
$this->processLimit = $config['scheduler']['process_limit'] ?? 4;
}
/**
* 多进程执行任务
*/
public function runConcurrently($tasks)
{
if (function_exists('pcntl_fork') === false) {
throw new Exception('PCNTL扩展未安装');
}
$taskChunks = array_chunk($tasks, $this->processLimit);
foreach ($taskChunks as $chunk) {
$this->forkAndExecute($chunk);
$this->waitForChildren();
}
}
private function forkAndExecute($tasks)
{
$pid = pcntl_fork();
if ($pid == -1) {
throw new Exception('无法创建子进程');
}
if ($pid) {
// 父进程
$this->childProcesses[] = $pid;
} else {
// 子进程
foreach ($tasks as $task) {
$result = $this->scheduler->runTask($task);
echo "Task {$task['name']} result: " . json_encode($result) . "\n";
}
exit(0);
}
}
private function waitForChildren()
{
while (count($this->childProcesses) > 0) {
$pid = pcntl_waitpid(-1, $status, WNOHANG);
if ($pid > 0) {
if (($key = array_search($pid, $this->childProcesses)) !== false) {
unset($this->childProcesses[$key]);
}
}
usleep(100000); // 0.1秒检查一次
}
}
}
Redis队列调度
1 基于Redis的队列调度器
<?php
// queue_scheduler.php
class QueueScheduler
{
private $redis;
private $queueKey = 'crawler:queue';
private $retryKey = 'crawler:retry';
public function __construct($config)
{
$this->redis = new Redis();
$this->redis->connect($config['redis']['host'], $config['redis']['port']);
}
/**
* 添加任务到队列
*/
public function enqueue($task, $priority = 0, $delay = 0)
{
$taskData = [
'data' => $task,
'priority' => $priority,
'available_at' => time() + $delay
];
// 使用有序集合实现优先级队列
$this->redis->zAdd($this->queueKey, $priority + ($taskData['available_at'] / 1000000), json_encode($taskData));
}
/**
* 从队列获取任务
*/
public function dequeue()
{
$options = [
'count' => 1,
'withscores' => true
];
$tasks = $this->redis->zRangeByScore($this->queueKey, 0, time(), $options);
if (empty($tasks)) {
return null;
}
$taskKey = key($tasks);
$this->redis->zRem($this->queueKey, $taskKey);
return json_decode($taskKey, true);
}
/**
* 任务重试
*/
public function retryTask($task, $retryCount = 0)
{
$this->enqueue($task, 5, 2^$retryCount * 5);
if ($retryCount > 3) {
$this->redis->hSet($this->retryKey, uniqid(), json_encode($task));
}
}
/**
* 消费者工作进程
*/
public function worker()
{
while (true) {
$task = $this->dequeue();
if ($task) {
try {
$this->processTask($task);
} catch (Exception $e) {
$this->handleFailure($task, $e);
}
} else {
usleep(1000000); // 1秒后重试
}
}
}
private function processTask($task)
{
echo "Processing task: " . json_encode($task) . "\n";
// 实际处理逻辑
}
private function handleFailure($task, Exception $e)
{
$task['error'] = $e->getMessage();
$this->retryTask($task);
}
}
数据库调度
1 基于数据库的任务调度
<?php
// database_scheduler.php
class DatabaseScheduler
{
private $pdo;
public function __construct($config)
{
$dsn = sprintf(
"mysql:host=%s;dbname=%s;charset=utf8mb4",
$config['database']['host'],
$config['database']['dbname']
);
$this->pdo = new PDO($dsn, $config['database']['user'], $config['database']['password']);
$this->pdo->setAttribute(PDO::ATTR_ERRMODE, PDO::ERRMODE_EXCEPTION);
$this->createTables();
}
/**
* 创建必要的表
*/
private function createTables()
{
$sql = "
CREATE TABLE IF NOT EXISTS scheduled_tasks (
id INT AUTO_INCREMENT PRIMARY KEY,
task_name VARCHAR(100) NOT NULL,
task_type VARCHAR(50) NOT NULL,
task_params TEXT,
schedule_expression VARCHAR(100),
is_active BOOLEAN DEFAULT TRUE,
last_run_at DATETIME,
next_run_at DATETIME,
status VARCHAR(20) DEFAULT 'pending',
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
INDEX idx_next_run (next_run_at)
) ENGINE=InnoDB;
CREATE TABLE IF NOT EXISTS task_executions (
id INT AUTO_INCREMENT PRIMARY KEY,
task_id INT NOT NULL,
execution_time DATETIME NOT NULL,
status VARCHAR(20) NOT NULL,
output TEXT,
error TEXT,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
FOREIGN KEY (task_id) REFERENCES scheduled_tasks(id)
) ENGINE=InnoDB;
";
$this->pdo->exec($sql);
}
/**
* 添加定时任务
*/
public function createScheduledTask($name, $type, $params, $schedule)
{
$sql = "INSERT INTO scheduled_tasks (task_name, task_type, task_params, schedule_expression, next_run_at)
VALUES (?, ?, ?, ?, NOW())";
$stmt = $this->pdo->prepare($sql);
$stmt->execute([$name, $type, json_encode($params), $schedule]);
return $this->pdo->lastInsertId();
}
/**
* 获取到期任务
*/
public function getDueTasks()
{
$sql = "SELECT * FROM scheduled_tasks
WHERE is_active = TRUE
AND next_run_at <= NOW()
AND status = 'pending'
ORDER BY next_run_at ASC
LIMIT 10";
$stmt = $this->pdo->query($sql);
return $stmt->fetchAll(PDO::FETCH_ASSOC);
}
/**
* 更新任务状态和下一次执行时间
*/
public function updateTaskAfterRun($taskId, $status, $output = null, $error = null)
{
try {
// 记录执行历史
$sql = "INSERT INTO task_executions (task_id, execution_time, status, output, error)
VALUES (?, NOW(), ?, ?, ?)";
$stmt = $this->pdo->prepare($sql);
$stmt->execute([$taskId, $status, $output, $error]);
// 更新任务信息
if ($status == 'success') {
$sql = "UPDATE scheduled_tasks
SET last_run_at = NOW(),
next_run_at = DATE_ADD(NOW(), INTERVAL 1 DAY),
status = 'pending'
WHERE id = ?";
} else {
$sql = "UPDATE scheduled_tasks
SET last_run_at = NOW(),
next_run_at = DATE_ADD(NOW(), INTERVAL 10 MINUTE),
status = 'retry'
WHERE id = ?";
}
$stmt = $this->pdo->prepare($sql);
$stmt->execute([$taskId]);
} catch (Exception $e) {
error_log("Error updating task: " . $e->getMessage());
}
}
/**
* 运行调度器
*/
public function runScheduler()
{
$tasks = $this->getDueTasks();
foreach ($tasks as $task) {
try {
$result = $this->executeTask($task);
$this->updateTaskAfterRun($task['id'], 'success', json_encode($result));
} catch (Exception $e) {
$this->updateTaskAfterRun($task['id'], 'error', null, $e->getMessage());
}
}
}
private function executeTask($task)
{
$params = json_decode($task['task_params'], true);
switch ($task['task_type']) {
case 'crawler':
return $this->executeCrawler($params);
case 'parser':
return $this->executeParser($params);
case 'notification':
return $this->executeNotification($params);
default:
throw new Exception("Unknown task type: {$task['task_type']}");
}
}
private function executeCrawler($params)
{
$url = $params['url'];
$html = file_get_contents($url);
// 示例：使用Goutte或Symfony DomCrawler
$crawler = new GoutteClient();
$crawler->request('GET', $url);
return [
'url' => $url,
'title' => $crawler->filter('title')->text(),
'content_length' => strlen($html)
];
}
private function executeParser($params)
{
// 解析逻辑
return ['status' => 'parsed'];
}
private function executeNotification($params)
{
// 通知逻辑
return ['status' => 'notified'];
}
}
完整使用示例
1 配置文件
<?php
// config.php
return [
'redis' => [
'host' => '127.0.0.1',
'port' => 6379,
'password' => '',
'db' => 0
],
'database' => [
'host' => '127.0.0.1',
'dbname' => 'crawler',
'user' => 'root',
'password' => 'password'
],
'scheduler' => [
'max_execution_time' => 3600,
'delay_between_tasks' => 5,
'process_limit' => 4,
'retry_count' => 3,
'retry_delay' => 60
],
'crawler' => [
'user_agent' => 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36',
'timeout' => 30,
'max_depth' => 3,
'respect_robots' => true
]
];
2 主调度入口
<?php
// main_scheduler.php
require_once 'TaskScheduler.php';
require_once 'CronScheduler.php';
require_once 'LoopScheduler.php';
require_once 'MultiProcessScheduler.php';
require_once 'QueueScheduler.php';
class MainScheduler
{
private $config;
private $scheduler;
public function __construct($config)
{
$this->config = $config;
$this->scheduler = new TaskScheduler($config);
}
/**
* 启动调度系统
*/
public function start()
{
$mode = $this->config['scheduler']['mode'] ?? 'cron';
switch ($mode) {
case 'cron':
$this->runCronMode();
break;
case 'loop':
$this->runLoopMode();
break;
case 'multiprocess':
$this->runMultiProcessMode();
break;
case 'queue':
$this->runQueueMode();
break;
default:
throw new Exception("Unsupported scheduler mode: $mode");
}
}
private function runCronMode()
{
$scheduler = new CronScheduler($this->config);
// 注册任务
$scheduler->registerTask('crawl_news', '*/5 * * * *', function() {
$crawler = new NewsCrawler();
return $crawler->crawlLatestNews();
});
$scheduler->registerTask('crawl_products', '0 */1 * * *', function() {
$crawler = new ProductCrawler();
return $crawler->crawlProducts();
});
$scheduler->checkAndRun();
}
private function runLoopMode()
{
$scheduler = new LoopScheduler($this->config);
$scheduler->runWithBackoff();
}
private function runMultiProcessMode()
{
$scheduler = new MultiProcessScheduler($this->config);
while (true) {
$tasks = $this->scheduler->getPendingTasks();
if (!empty($tasks)) {
$scheduler->runConcurrently($tasks);
}
sleep(10);
}
}
private function runQueueMode()
{
$scheduler = new QueueScheduler($this->config);
$scheduler->worker();
}
}
// 启动调度器
$config = require 'config.php';
$mainScheduler = new MainScheduler($config);
$mainScheduler->start();
3 监控脚本
<?php
// monitor.php
class SchedulerMonitor
{
private $redis;
private $pdo;
public function __construct($config)
{
$this->redis = new Redis();
$this->redis->connect($config['redis']['host'], $config['redis']['port']);
$dsn = "mysql:host=" . $config['database']['host'] . ";dbname=" . $config['database']['dbname'];
$this->pdo = new PDO($dsn, $config['database']['user'], $config['database']['password']);
}
/**
* 获取系统状态
*/
public function getStatus()
{
return [
'queue_size' => $this->redis->lLen('crawler:queue'),
'active_tasks' => $this->redis->hLen('crawler:tasks'),
'last_executions' => $this->getLastExecutions(),
'error_rates' => $this->getErrorRate()
];
}
private function getLastExecutions($limit = 10)
{
$sql = "SELECT * FROM task_executions ORDER BY created_at DESC LIMIT ?";
$stmt = $this->pdo->prepare($sql);
$stmt->execute([$limit]);
return $stmt->fetchAll(PDO::FETCH_ASSOC);
}
private function getErrorRate()
{
$sql = "SELECT
COUNT(*) as total,
SUM(CASE WHEN status = 'error' THEN 1 ELSE 0 END) as errors
FROM task_executions
WHERE created_at >= DATE_SUB(NOW(), INTERVAL 1 HOUR)";
$stmt = $this->pdo->query($sql);
$result = $stmt->fetch(PDO::FETCH_ASSOC);
return $result['total'] > 0 ?
($result['errors'] / $result['total']) * 100 : 0;
}
/**
* 发送警报
*/
public function sendAlert($message, $severity = 'warning')
{
// 实现邮件、短信或Webhook通知
// ...
}
}
// 定期执行监控
$monitor = new SchedulerMonitor($config);
$status = $monitor->getStatus();
if ($status['error_rates'] > 20) {
$monitor->sendAlert("错误率过高: " . $status['error_rates'] . "%");
}
这套PHP爬虫任务调度系统包含：
- 多种调度模式：Cron定时、无限循环、多进程、Redis队列
- 任务管理：添加、删除、查询、暂停/恢复
- 错误处理：重试机制、错误日志、告警通知
- 高可用性：多进程并发、队列处理、数据库持久化
- 监控支持：执行日志、状态查询、错误率统计
使用建议：
- 简单任务使用Cron调度
- 需要快速响应用Loop模式
- 大规模爬取使用多进程或队列模式
- 生产环境推荐使用Redis队列+ Workers模式
根据实际需求选择合适的调度方式,并根据系统负载调整配置参数。
