PHP curl_multi_add_handle函数

示例 4：动态添加句柄与进度监控

@php
class DynamicMultiRequest {
    private $multiHandle;
    private $activeHandles = [];
    private $completedHandles = [];
    private $maxConcurrent;

    public function __construct($maxConcurrent = 5) {
        $this->multiHandle = curl_multi_init();
        $this->maxConcurrent = $maxConcurrent;
    }

    /**
     * 动态添加请求
     */
    public function addRequest($url, $options = [], $id = null) {
        if ($id === null) {
            $id = uniqid('req_');
        }

        $ch = curl_init();

        $defaultOptions = [
            CURLOPT_URL => $url,
            CURLOPT_RETURNTRANSFER => true,
            CURLOPT_TIMEOUT => 30,
            CURLOPT_NOPROGRESS => false,
            CURLOPT_PROGRESSFUNCTION => function($resource, $download_size, $downloaded, $upload_size, $uploaded) use ($id) {
                if ($download_size > 0) {
                    $progress = round(($downloaded / $download_size) * 100);
                    // 在实际应用中，这里可以更新进度显示
                    echo "进度 [{$id}]: {$progress}%\r";
                }
                return 0;
            }
        ];

        $finalOptions = $options + $defaultOptions;
        curl_setopt_array($ch, $finalOptions);

        // 添加到多句柄
        $result = curl_multi_add_handle($this->multiHandle, $ch);

        if ($result === CURLM_OK) {
            $this->activeHandles[$id] = [
                'handle' => $ch,
                'url' => $url,
                'added_time' => microtime(true)
            ];
            return $id;
        } else {
            curl_close($ch);
            return false;
        }
    }

    /**
     * 执行所有请求（支持动态添加）
     */
    public function execute($pendingUrls = []) {
        $allUrls = $pendingUrls;
        $processedCount = 0;

        echo "开始执行动态并行请求 (最大并发: {$this->maxConcurrent})...\n\n";

        $startTime = microtime(true);

        while (!empty($this->activeHandles) || !empty($allUrls)) {
            // 动态添加新请求（不超过最大并发数）
            while (count($this->activeHandles) < $this->maxConcurrent && !empty($allUrls)) {
                $url = array_shift($allUrls);
                $id = $this->addRequest($url);
                if ($id) {
                    $processedCount++;
                    echo "已添加请求: {$url} [{$processedCount}]\n";
                }
            }

            if (empty($this->activeHandles)) {
                break;
            }

            // 执行当前活动的请求
            $this->executeActiveRequests();

            // 处理完成的请求
            $this->processCompletedRequests();
        }

        $totalTime = microtime(true) - $startTime;

        echo "\n所有请求完成!\n";
        echo "总请求数: {$processedCount}\n";
        echo "总耗时: " . round($totalTime, 2) . " 秒\n";

        return $this->completedHandles;
    }

    private function executeActiveRequests() {
        $running = null;
        do {
            $status = curl_multi_exec($this->multiHandle, $running);

            if ($running) {
                curl_multi_select($this->multiHandle, 0.1);
            }
        } while ($running > 0 && $status === CURLM_OK);
    }

    private function processCompletedRequests() {
        while ($info = curl_multi_info_read($this->multiHandle)) {
            $handle = $info['handle'];
            $id = $this->findHandleId($handle);

            if ($id !== null) {
                $this->completeRequest($id, $handle, $info);
            }
        }
    }

    private function findHandleId($handle) {
        foreach ($this->activeHandles as $id => $data) {
            if ($data['handle'] === $handle) {
                return $id;
            }
        }
        return null;
    }

    private function completeRequest($id, $handle, $info) {
        $content = curl_multi_getcontent($handle);
        $curlInfo = curl_getinfo($handle);
        $error = curl_error($handle);

        $requestData = $this->activeHandles[$id];
        $processingTime = microtime(true) - $requestData['added_time'];

        $this->completedHandles[$id] = [
            'url' => $requestData['url'],
            'success' => ($info['result'] === CURLE_OK),
            'http_code' => $curlInfo['http_code'],
            'processing_time' => $processingTime,
            'content_length' => strlen($content),
            'error' => $error
        ];

        // 从多句柄中移除
        curl_multi_remove_handle($this->multiHandle, $handle);
        curl_close($handle);

        unset($this->activeHandles[$id]);

        $status = $info['result'] === CURLE_OK ? '完成' : '失败';
        echo "请求完成 [{$id}]: {$status} (" . round($processingTime, 2) . "秒)\n";
    }

    public function close() {
        foreach ($this->activeHandles as $data) {
            if (is_resource($data['handle'])) {
                curl_multi_remove_handle($this->multiHandle, $data['handle']);
                curl_close($data['handle']);
            }
        }

        if (is_resource($this->multiHandle)) {
            curl_multi_close($this->multiHandle);
        }

        $this->activeHandles = [];
        $this->completedHandles = [];
    }

    public function __destruct() {
        $this->close();
    }
}

// 使用动态并行请求
$dynamic = new DynamicMultiRequest(3); // 最大并发3个

// 准备URL列表（模拟20个请求）
$urls = [];
for ($i = 1; $i <= 20; $i++) {
    $delay = ($i % 4) + 1; // 延迟1-4秒
    $urls[] = "https://httpbin.org/delay/{$delay}";
}

echo "准备处理 " . count($urls) . " 个URL...\n";

$results = $dynamic->execute($urls);

// 统计结果
$successful = array_filter($results, function($r) {
    return $r['success'];
});

echo "\n最终统计:\n";
echo "成功: " . count($successful) . " 个\n";
echo "失败: " . (count($results) - count($successful)) . " 个\n";

// 显示最快的3个请求
usort($results, function($a, $b) {
    return $a['processing_time'] <=> $b['processing_time'];
});

echo "\n最快的3个请求:\n";
for ($i = 0; $i < min(3, count($results)); $i++) {
    $result = $results[$i];
    echo ($i + 1) . ". " . $result['url'] . " (" . round($result['processing_time'], 2) . "秒)\n";
}

$dynamic->close();
@endphp