多任务分时段分批执行方案咨询（基于cron与PHP环境）

嘿，我碰到过几乎一模一样的场景，给你几个实际能用的解决方案，都是在项目里跑通的：

方案1：用队列系统（最推荐） #

这是处理动态任务并发最稳妥的方式，天然适配你这种任务数不固定的场景，核心思路是把任务丢进队列，然后用和处理器核心数一致的worker进程持续消费任务：

• 第一步：给你的任务加个状态字段（比如status：0=待处理，1=执行中，2=完成），方便追踪进度
• 第二步：cron定时脚本只负责从SQL拉取待处理任务，把任务参数（比如要执行的command、任务ID）序列化后塞进队列（推荐用Redis，轻量又好用）
• 第三步：启动固定数量的worker进程（数量等于你的处理器核心数），这些worker会一直后台运行，阻塞等待队列里的任务，拿到就执行

代码示例 #

塞任务的cron脚本

// 连接Redis
$redis = new Redis();
$redis->connect('127.0.0.1', 6379);

// 从SQL拉取待处理任务
$pdo = new PDO('mysql:host=localhost;dbname=your_db', 'user', 'pass');
$stmt = $pdo->query("SELECT id, command FROM tasks WHERE status = 0");
$tasks = $stmt->fetchAll(PDO::FETCH_ASSOC);

// 把任务塞进队列
foreach ($tasks as $task) {
    $redis->lpush('task_queue', json_encode($task));
    // 标记任务为已入队，避免重复拉取
    $pdo->exec("UPDATE tasks SET status = 1 WHERE id = {$task['id']}");
}

Worker脚本（常驻进程）

$redis = new Redis();
$redis->connect('127.0.0.1', 6379);
$pdo = new PDO('mysql:host=localhost;dbname=your_db', 'user', 'pass');

// 获取处理器核心数
$coreCount = (int)shell_exec('nproc');

// 创建对应数量的子进程作为worker
for ($i = 0; $i < $coreCount; $i++) {
    $pid = pcntl_fork();
    if ($pid === 0) {
        // 子进程：持续消费队列
        while (true) {
            // 阻塞等待队列任务，0表示一直等
            $taskData = $redis->brpop('task_queue', 0);
            if ($taskData) {
                $task = json_decode($taskData[1], true);
                // 执行任务，把输出丢去/dev/null避免占内存
                exec($task['command'] . ' > /dev/null 2>&1', $output, $returnCode);
                // 更新任务状态：成功为2，失败为3
                $status = $returnCode === 0 ? 2 : 3;
                $pdo->exec("UPDATE tasks SET status = {$status} WHERE id = {$task['id']}");
            }
        }
        exit();
    }
}

// 父进程等待所有子进程（用Supervisor管理的话可以去掉这部分）
while (pcntl_wait($status) !== -1);

部署小技巧

用supervisor管理worker进程，配置它自动重启挂掉的worker，确保一直有足够的进程处理任务，不用手动维护。

方案2：手动分批+定时监控（无队列轻量化方案） #

如果不想引入队列，也可以用“分批启动+定时检查”的方式：

• 第一步：同样给任务加状态字段和批次标记batch_id
• 第二步：cron脚本拉取所有待处理任务，按核心数分成多个批次，先启动第一批（把任务状态设为执行中，用&让进程后台运行），剩下的批次存在Redis或者数据库里
• 第三步：写一个监控脚本（用cron每分钟执行一次），检查当前正在运行的任务数，一旦少于核心数，就启动下一批任务

代码示例（启动第一批） #

$coreCount = (int)shell_exec('nproc');
$pdo = new PDO('mysql:host=localhost;dbname=your_db', 'user', 'pass');
$stmt = $pdo->query("SELECT id, command FROM tasks WHERE status = 0");
$tasks = $stmt->fetchAll(PDO::FETCH_ASSOC);

// 拆分批次
$batches = array_chunk($tasks, $coreCount);
if (empty($batches)) exit;

// 处理第一批
$firstBatch = $batches[0];
foreach ($firstBatch as $task) {
    $pdo->exec("UPDATE tasks SET status = 1, batch_id = 1 WHERE id = {$task['id']}");
    // 后台执行任务
    exec($task['command'] . ' > /dev/null 2>&1 &');
}

// 把剩余批次存到Redis
$redis = new Redis();
$redis->connect('127.0.0.1', 6379);
$remainingBatches = array_slice($batches, 1);
$redis->set('remaining_task_batches', json_encode($remainingBatches));

监控脚本（每分钟cron执行） #

$redis = new Redis();
$redis->connect('127.0.0.1', 6379);
$remainingBatches = json_decode($redis->get('remaining_task_batches'), true);
if (empty($remainingBatches)) exit;

$pdo = new PDO('mysql:host=localhost;dbname=your_db', 'user', 'pass');
$coreCount = (int)shell_exec('nproc');

// 检查当前运行的任务数
$runningCount = $pdo->query("SELECT COUNT(*) FROM tasks WHERE status = 1")->fetchColumn();

if ($runningCount < $coreCount) {
    // 启动下一批
    $nextBatch = array_shift($remainingBatches);
    $batchId = time(); // 用时间戳当批次ID
    foreach ($nextBatch as $task) {
        $pdo->exec("UPDATE tasks SET status = 1, batch_id = {$batchId} WHERE id = {$task['id']}");
        exec($task['command'] . ' > /dev/null 2>&1 &');
    }
    // 更新剩余批次
    $redis->set('remaining_task_batches', json_encode($remainingBatches));
    // 如果没有剩余批次，删除key
    if (empty($remainingBatches)) $redis->del('remaining_task_batches');
}

方案3：进程控制直接管理并发（极简方案） #

如果任务体量不大，直接用PHP的pcntl扩展在一个脚本里控制并发数，不用额外组件：

$pdo = new PDO('mysql:host=localhost;dbname=your_db', 'user', 'pass');
$stmt = $pdo->query("SELECT id, command FROM tasks WHERE status = 0");
$tasks = $stmt->fetchAll(PDO::FETCH_ASSOC);
$coreCount = (int)shell_exec('nproc');
$runningPids = [];

foreach ($tasks as $task) {
    // 等待当前运行进程数降到核心数以下
    while (count($runningPids) >= $coreCount) {
        // 非阻塞等待子进程结束
        $pid = pcntl_wait($status, WNOHANG);
        if ($pid > 0) {
            unset($runningPids[array_search($pid, $runningPids)]);
        }
        usleep(100000); // 每隔100ms检查一次
    }

    // 创建子进程执行任务
    $pid = pcntl_fork();
    if ($pid === 0) {
        // 子进程执行任务
        exec($task['command'] . ' > /dev/null 2>&1', $output, $returnCode);
        $status = $returnCode === 0 ? 2 : 3;
        $pdo->exec("UPDATE tasks SET status = {$status} WHERE id = {$task['id']}");
        exit($returnCode);
    } else {
        $runningPids[] = $pid;
    }
}

// 等待所有子进程完成
foreach ($runningPids as $pid) {
    pcntl_waitpid($pid, $status);
}

注意：这个脚本会一直运行到所有任务完成，要给cron设置合理的执行间隔，或者加个文件锁防止重复触发。

内容的提问来源于stack exchange，提问作者Dany M