PHP项目离线任务如何定时批量处理海量数据

本文目录导读：

1. 架构设计核心原则
2. 主流方案对比与选型
3. 具体实现步骤（以方案1为例：Cron + 批处理）
4. 更优方案：消息队列方式（生产推荐）
5. 关键优化与注意事项

针对PHP项目中离线任务定时批量处理海量数据的需求,核心挑战在于避免内存溢出、防止脚本超时以及确保任务不重复执行，以下是系统性的解决方案，从架构设计到代码实现。

架构设计核心原则

1. 解耦任务调度与任务执行：使用独立的调度器（如Cron）触发任务，任务本身是一个负责“分发”或“执行”的脚本。
2. 分而治之（Chunking）：永远不要尝试一次加载所有数据，必须分批次处理。
3. 避免阻塞：单次任务运行时间应可控，避免影响其他在线服务。

主流方案对比与选型

对于大多数“海量数据定时处理”场景，推荐方案2（消息队列）作为生产级选择，方案1作为快速原型选择。

具体实现步骤（以方案1为例：Cron + 批处理）

假设你需要每天凌晨3点处理昨天产生的 500万 条用户行为日志。

创建批处理脚本 (batch_process.php)

<?php
/**
 * 批处理脚本：按ID范围分批处理
 * 调用方式：php batch_process.php --start-id=0 --end-id=50000000 --batch=1000
 */
// 1. 超时与内存控制
set_time_limit(0); // 不限制执行时间（由Cron或进程管理器控制）
ini_set('memory_limit', '512M'); // 根据单批数据量调整，但不要过大
// 2. 解析命令行参数
$options = getopt('', ['start-id:', 'end-id:', 'batch::']);  // 读取当前批次信息
$startId = (int)($options['start-id'] ?? 0);
$endId = (int)($options['end-id'] ?? 0);
$batchSize = (int)($options['batch'] ?? 1000);
if ($endId <= 0 || $startId >= $endId) {
    die("Invalid ID range.\n");
}
// 3. 连接数据库（使用长连接或PDO连接池）
$pdo = new PDO('mysql:host=127.0.0.1;dbname=your_db;charset=utf8', 'user', 'pass', [
    PDO::ATTR_PERSISTENT => true, // 长连接，避免每次创建连接
]);
// 4. 循环分批处理
$currentStart = $startId;
while ($currentStart <= $endId) {
    $currentEnd = min($currentStart + $batchSize, $endId);
    // 4.1 查询当前批次数据
    $stmt = $pdo->prepare("SELECT * FROM user_logs WHERE id BETWEEN ? AND ?");
    $stmt->execute([$currentStart, $currentEnd]);
    $rows = $stmt->fetchAll(PDO::FETCH_ASSOC);
    if (empty($rows)) {
        $currentStart = $currentEnd + 1;
        continue;
    }
    // 4.2 处理每条数据（占用内存小）
    foreach ($rows as $row) {
        // 具体业务逻辑：例如计算、聚合、发送通知等
        // 注意：不要在循环内执行耗时IO操作（如单独查库），应批量处理
        processRow($row);
        // 可选：释放内存
        unset($row);
    }
    // 4.3 记录进度日志（用于断点续传）
    file_put_contents('/var/log/process_progress.log', "Completed IDs: {$currentStart} to {$currentEnd}\n", FILE_APPEND);
    // 4.4 释放内存
    $rows = null;
    $currentStart = $currentEnd + 1;
    // 可选：短暂休眠，避免压垮数据库
    usleep(100); // 0.1毫秒
}
echo "Batch processing completed from $startId to $endId.\n";

创建主调度脚本 (dispatch_process.php)

这个脚本由Cron调用,负责决定本次要处理的范围，并调用上面的批处理脚本。

<?php
// 主调度脚本：由Cron每分钟或每5分钟调用一次
// 检查是否有未完成的批次，并启动
$progressFile = '/var/log/process_progress.log';
$lastProcessedId = 0;
// 读取上次处理到的位置
if (file_exists($progressFile)) {
    $lines = file($progressFile, FILE_IGNORE_NEW_LINES | FILE_SKIP_EMPTY_LINES);
    if (!empty($lines)) {
        $lastLine = end($lines);
        preg_match('/Completed IDs: (\d+) to (\d+)/', $lastLine, $matches);
        $lastProcessedId = (int)($matches[2] ?? 0);
    }
}
// 获取当前最大ID（确保不会重复处理新生成的数据）
$pdo = new PDO('mysql:host=127.0.0.1;dbname=your_db', 'user', 'pass');
$stmt = $pdo->query("SELECT MAX(id) FROM user_logs WHERE created_at < '2023-10-27 00:00:00'");
$maxId = (int)$stmt->fetchColumn();
if ($lastProcessedId >= $maxId) {
    echo "No new data to process.\n";
    exit(0);
}
// 执行批处理（使用 proc_open 或 shell_exec 避免阻塞主进程）
$cmd = sprintf(
    "php /path/to/batch_process.php --start-id=%d --end-id=%d --batch=1000",
    $lastProcessedId + 1,
    $maxId
);
// 后台执行，不等待输出
exec("{$cmd} > /dev/null 2>&1 &");
echo "Dispatched batch process: {$cmd}\n";

配置Cron任务

# 每5分钟检查并调度一次
*/5 * * * * /usr/bin/php /path/to/distpatch_process.php >> /var/log/cron_dispatcher.log 2>&1

更优方案：消息队列方式（生产推荐）

如果数据量达到亿级,或需要处理失败重试，建议使用消息队列。

架构流程：

1. 生产者：Cron触发时，将“需要处理的批次”作为消息发送到Redis队列，每个消息包含 { "start_id": 10000, "end_id": 10999 }。
2. 消费者：多个PHP进程（Workers）从队列中消费消息，独立执行批处理脚本。
3. 失败重试：队列提供失败重试机制（如RabbitMQ的dead letter）。
4. 进度管理：使用Redis记录已处理的最大ID。

示例代码（Redis作为队列）：

// 生产者：dispatch_to_queue.php
$redis = new Redis();
$redis->connect('127.0.0.1', 6379);
$total = 50000000;
$batch = 1000;
for ($i = 0; $i < $total; $i += $batch) {
    $msg = json_encode(['start' => $i, 'end' => min($i + $batch, $total)]);
    $redis->lPush('task_queue', $msg);
}
// 消费者：worker.php (常驻运行)
while (true) {
    $msg = $redis->brPop('task_queue', 10); // 阻塞读取
    if ($msg) {
        $data = json_decode($msg[1], true);
        // 执行 batch_process.php 类似的逻辑
        processBatch($data['start'], $data['end']);
    }
}

关键优化与注意事项

1. 数据库层面：

   • 使用索引：确保 WHERE id BETWEEN ... AND ... 走索引。
   • 读写分离：批处理任务读取从库，避免影响主库写入性能。
   • 避免锁表：处理时尽量使用 SELECT ... FOR UPDATE 或乐观锁。

2. 内存泄漏预防：

   • 使用 unset() 显式释放变量。
   • 避免在循环内创建全局对象（如 $this->db）。
   • 批量处理框架可以选择 Laravel Chunk 或 ThinkPHP chunk 方法，它们会自动控制逻辑与内存。

3. 断点续传 & 幂等性：

   • 记录处理进度（如文件、数据库表、Redis）。
   • 保证每条数据被处理的结果唯一,即使重复执行也不会出错。

4. 监控与报警：

   • 记录每次批处理的耗时、处理条数、错误数。
   • 当失败次数超过阈值时发送邮件或短信。

最终建议：如果是遗留PHP项目，先使用Cron + 批处理（方案一）实现，监控一段时间，当出现性能瓶颈时（如脚本运行超过15分钟），再逐步迁移到消息队列方案。