PHP CURL多线程与单线程结合实现混合任务执行求助

看起来你已经有了不错的基础——并行CURL处理和单串行处理，但现在需要把两者结合，让串行任务链的首个任务也能参与全局并行，后续任务则依赖前一个完成再启动。这个需求在任务管理系统里很常见，我来给你梳理思路并提供实现方案：

核心思路拆解 #

我们需要把任务分成两类来管理：

• 独立任务：无依赖，直接加入并行池执行
• 串行任务链：属于同一个主任务的明细序列，比如[URL-D, URL-E, URL-F]，这类任务的规则是：
  1. 链的第一个任务先加入并行池，和其他独立任务一起跑
  2. 当链的当前任务完成后，自动启动链的下一个任务，加入并行池（不阻塞其他任务）

这样既保证了并行效率，又满足了串行依赖的要求。

代码实现方案 #

我们可以封装一个任务管理器，结合你现有的CURL函数，改造为支持混合任务的版本。以下是完整的实现：

1. 任务结构定义 #

首先，我们需要统一任务的格式，方便管理：

// 独立任务格式：['type' => 'single', 'url' => 'xxx']
// 串行任务链格式：['type' => 'chain', 'urls' => ['url1', 'url2', ...], 'current_index' => 0]

2. 改造后的混合任务处理函数 #

function runMixedRequests($tasks, $thread_width = 10) {
    $master = curl_multi_init();
    $curl_opts = [
        CURLOPT_RETURNTRANSFER => true,
        CURLOPT_FOLLOWLOCATION => true,
        CURLOPT_MAXREDIRS => 5,
        CURLOPT_CONNECTTIMEOUT => 15,
        CURLOPT_TIMEOUT => 15
    ];
    $active_handles = []; // 记录当前活跃的CURL句柄和对应的任务
    $results = [];

    // 初始化：把所有独立任务和串行链的第一个任务加入队列
    foreach ($tasks as $task_id => $task) {
        if ($task['type'] === 'single') {
            $ch = curl_init($task['url']);
            curl_setopt_array($ch, $curl_opts);
            curl_multi_add_handle($master, $ch);
            $active_handles[(int)$ch] = [
                'task_id' => $task_id,
                'task_type' => 'single',
                'url' => $task['url']
            ];
        } elseif ($task['type'] === 'chain' && $task['current_index'] === 0) {
            $first_url = $task['urls'][0];
            $ch = curl_init($first_url);
            curl_setopt_array($ch, $curl_opts);
            curl_multi_add_handle($master, $ch);
            $active_handles[(int)$ch] = [
                'task_id' => $task_id,
                'task_type' => 'chain',
                'url' => $first_url,
                'chain_index' => 0
            ];
        }
    }

    // 处理任务循环
    do {
        // 执行CURL多线程
        while(($execrun = curl_multi_exec($master, $running)) === -1) {}
        curl_multi_select($master);

        // 处理完成的任务
        while ($done = curl_multi_info_read($master)) {
            $ch = $done['handle'];
            $handle_key = (int)$ch;
            $task_info = $active_handles[$handle_key];
            $result_content = curl_multi_getcontent($ch);

            // 记录结果
            $results[] = [
                'task_id' => $task_info['task_id'],
                'url' => $task_info['url'],
                'result' => $result_content
            ];

            // 如果是串行链任务，检查是否有下一个任务
            if ($task_info['task_type'] === 'chain') {
                $current_chain = &$tasks[$task_info['task_id']];
                $next_index = $task_info['chain_index'] + 1;
                if ($next_index < count($current_chain['urls'])) {
                    // 启动链的下一个任务，加入并行池
                    $next_url = $current_chain['urls'][$next_index];
                    $new_ch = curl_init($next_url);
                    curl_setopt_array($new_ch, $curl_opts);
                    curl_multi_add_handle($master, $new_ch);
                    $active_handles[(int)$new_ch] = [
                        'task_id' => $task_info['task_id'],
                        'task_type' => 'chain',
                        'url' => $next_url,
                        'chain_index' => $next_index
                    ];
                    // 更新链的当前索引
                    $current_chain['current_index'] = $next_index;
                }
            }

            // 清理完成的句柄
            curl_multi_remove_handle($master, $ch);
            curl_close($ch);
            unset($active_handles[$handle_key]);
        }
    } while ($running > 0);

    curl_multi_close($master);
    return $results;
}

3. 使用示例 #

假设你从数据库拿到的任务数据：

// 构造任务列表
$tasks = [
    // 独立任务（单主单明细）
    1 => ['type' => 'single', 'url' => 'https://example.com/task-A'],
    2 => ['type' => 'single', 'url' => 'https://example.com/task-B'],
    // 串行任务链（单主多明细：D→E→F）
    3 => ['type' => 'chain', 'urls' => ['https://example.com/task-D', 'https://example.com/task-E', 'https://example.com/task-F'], 'current_index' => 0],
    // 另一个串行任务链（G→H）
    4 => ['type' => 'chain', 'urls' => ['https://example.com/task-G', 'https://example.com/task-H'], 'current_index' => 0]
];

// 执行混合任务
$results = runMixedRequests($tasks, 10);

// 处理结果
foreach ($results as $res) {
    echo "任务ID {$res['task_id']} 处理URL {$res['url']} 完成\n";
}

关键点说明 #

1. 任务跟踪：通过active_handles数组关联CURL句柄和对应的任务信息，方便任务完成后判断是否需要启动下一个串行任务
2. 串行链触发：当串行链的当前任务完成后，立即创建新的CURL句柄加入多线程池，这样不会阻塞其他并行任务，同时保证了链内的顺序
3. 复用现有逻辑：保留了你原有的CURL配置和多线程执行逻辑，只是增加了任务类型的判断和串行链的调度

这个方案既满足了你“串行任务首个参与并行”的需求，又最大化了多线程的效率，同时清晰地管理了任务的依赖关系。

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