本文目录导读：
我来为您提供一个完整的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模式
根据实际需求选择合适的调度方式,并根据系统负载调整配置参数。
版权声明：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直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
