PHP 多執行緒實作方法詳解：非阻塞 I/O、流程控制與實務應用

PHP 本身不支援原生多執行緒，但可透過非阻塞 I/O (Non-blocking I/O)、多進程、擴充套件 實現並發處理。本文深入解析 stream_select 非阻塞網路 I/O、pcntl_fork 多進程、pthreads/parallel 擴充套件三大方案。

方案一：非阻塞 I/O (stream_select) - 純 PHP、無需擴充

適用於並發網路請求：同時檢查多個 API、爬蟲、健康檢查、微服務呼叫。

核心原理

1. 建立多個非阻塞 socket 連線
2. 使用 stream_select() 監控可讀/可寫/異常事件
3. 事件觸發時讀寫資料，達到單進程並發處理

完整實作：並發 HTTP 請求

<?php
/**
 * 非阻塞並發 HTTP 客戶端
 */
class AsyncHttpClient {
    private $sockets = [];
    private $urls = [];
    private $responses = [];
    private $timeouts = [];

    public function addRequest($url, $timeout = 10) {
        $this->urls[] = $url;
        $this->timeouts[] = $timeout;
    }

    public function execute() {
        $this->initConnections();
        $this->eventLoop();
        return $this->responses;
    }

    private function initConnections() {
        foreach ($this->urls as $i => $url) {
            $parsed = parse_url($url);
            $host = $parsed['host'];
            $port = $parsed['port'] ?? ($parsed['scheme'] === 'https' ? 443 : 80);
            $path = ($parsed['path'] ?? '/') . ($parsed['query'] ? '?' . $parsed['query'] : '');
            
            // 建立非阻塞連線
            $flags = STREAM_CLIENT_ASYNC_CONNECT | STREAM_CLIENT_CONNECT;
            if ($parsed['scheme'] === 'https') $flags |= STREAM_CLIENT_SSL;
            
            $socket = stream_socket_client(
                "{$parsed['scheme']}://{$host}:{$port}",
                $errno, $errstr,
                $this->timeouts[$i],
                $flags
            );
            
            if ($socket) {
                stream_set_blocking($socket, false);
                stream_set_timeout($socket, $this->timeouts[$i]);
                $this->sockets[$i] = $socket;
                
                // 準備 HTTP 請求
                $request = "GET {$path} HTTP/1.1\r\n";
                $request .= "Host: {$host}\r\n";
                $request .= "Connection: close\r\n\r\n";
                $this->pending_write[$i] = $request;
            } else {
                $this->responses[$i] = ['error' => "Connect failed: $errstr ($errno)"];
            }
        }
    }

    private function eventLoop() {
        $pending = $this->sockets;
        
        while (!empty($pending)) {
            $read = $write = $pending;
            $except = null;
            
            // 等待事件 (非阻塞)
            $n = @stream_select($read, $write, $except, 30);  // 30秒總逾時
            
            if ($n === false) break;  // 錯誤
            if ($n === 0) {  // 總逾時
                foreach ($pending as $i => $sock) {
                    if (!isset($this->responses[$i])) {
                        $this->responses[$i] = ['error' => 'Total timeout'];
                    }
                }
                break;
            }
            
            // 處理可寫：發送 HTTP 請求
            foreach ($write as $sock) {
                $i = array_search($sock, $this->sockets);
                if ($i !== false && isset($this->pending_write[$i])) {
                    fwrite($sock, $this->pending_write[$i]);
                    unset($this->pending_write[$i]);
                    $this->response_buffer[$i] = '';
                }
            }
            
            // 處理可讀：接收回應
            foreach ($read as $sock) {
                $i = array_search($sock, $this->sockets);
                if ($i === false) continue;
                
                $data = fread($sock, 8192);
                if ($data === '' || $data === false) {
                    // 連線關閉
                    fclose($sock);
                    unset($pending[$i]);
                    // 解析 HTTP 回應
                    $this->parseResponse($i);
                } else {
                    $this->response_buffer[$i] .= $data;
                }
            }
            
            // 處理異常
            if ($except) {
                foreach ($except as $sock) {
                    $i = array_search($sock, $this->sockets);
                    fclose($sock);
                    unset($pending[$i]);
                    $this->responses[$i] = ['error' => 'Stream exception'];
                }
            }
        }
    }

    private function parseResponse($i) {
        $buffer = $this->response_buffer[$i] ?? '';
        // 簡易解析：分離 header 和 body
        $parts = explode("\r\n\r\n", $buffer, 2);
        $header = $parts[0] ?? '';
        $body = $parts[1] ?? '';
        
        // 解析狀態碼
        preg_match('/HTTP\/\d\.\d\s+(\d+)/', $header, $m);
        $status = $m[1] ?? 0;
        
        $this->responses[$i] = [
            'status' => (int)$status,
            'header' => $header,
            'body' => $body,
            'url' => $this->urls[$i],
        ];
    }
}

// 使用範例
$client = new AsyncHttpClient();
$urls = [
    'https://httpbin.org/delay/1',
    'https://httpbin.org/delay/2',
    'https://httpbin.org/delay/3',
    'https://httpbin.org/get',
    'https://httpbin.org/uuid',
];
foreach ($urls as $url) $client->addRequest($url, 10);

$start = microtime(true);
$results = $client->execute();
$elapsed = microtime(true) - $start;

echo "總耗時: {$elapsed} 秒 (循序約 7 秒)\n";
foreach ($results as $i => $r) {
    echo "[$i] {$r['url']} => HTTP {$r['status']} (" . strlen($r['body']) . " bytes)\n";
}
?>

方案二：pcntl_fork 多進程 - 適用 CPU 密集任務

需啟用 --enable-pcntl 編譯 PHP (Windows 不支援)。

基本 Fork 模式

<?php
$children = [];
$task_count = 10;

for ($i = 0; $i < $task_count; $i++) {
    $pid = pcntl_fork();
    
    if ($pid === -1) {
        die('Fork 失敗');
    } elseif ($pid === 0) {
        // 子進程
        $result = doHeavyTask($i);
        echo "Child $i 完成: $result\n";
        exit(0);
    } else {
        // 父進程
        $children[$pid] = $i;
    }
}

// 父進程等待所有子進程
while (!empty($children)) {
    $pid = pcntl_wait($status);
    if ($pid > 0) {
        unset($children[$pid]);
        echo "Child $pid 結束, 狀態: $status\n";
    }
}

function doHeavyTask($id) {
    // 模擬 CPU 密集運算
    $sum = 0;
    for ($i = 0; $i < 1000000; $i++) $sum += $i;
    return $sum;
}
?>

進階：進程池模式 (固定進程數)

<?php
class ProcessPool {
    private $max_processes;
    private $tasks = [];
    private $running = [];
    private $results = [];

    public function __construct($max = 4) {
        $this->max_processes = $max;
    }

    public function addTask(callable $task, $data) {
        $this->tasks[] = [$task, $data];
    }

    public function run() {
        while (!empty($this->tasks) || !empty($this->running)) {
            // 啟動新進程直到達到上限
            while (count($this->running) < $this->max_processes && !empty($this->tasks)) {
                $this->forkNext();
            }
            
            // 等待任一子進程結束
            $pid = pcntl_wait($status, WUNTRACED);
            if ($pid > 0) {
                $this->handleChildExit($pid, $status);
            }
        }
        return $this->results;
    }

    private function forkNext() {
        $task_data = array_shift($this->tasks);
        $pid = pcntl_fork();
        
        if ($pid === 0) {
            // 子進程執行任務
            [$task, $data] = $task_data;
            $result = $task($data);
            echo json_encode(['result' => $result]);
            exit(0);
        } elseif ($pid > 0) {
            $this->running[$pid] = $task_data;
        }
    }

    private function handleChildExit($pid, $status) {
        if (isset($this->running[$pid])) {
            // 讀取子進程輸出 (需配合 pipe 或檔案)
            // 這裡簡化處理
            unset($this->running[$pid]);
        }
    }
}

// 使用
$pool = new ProcessPool(4);
for ($i = 0; $i < 20; $i++) {
    $pool->addTask(function($id) {
        sleep(rand(1,3));
        return "Task $id done";
    }, $i);
}
$results = $pool->run();
?>

方案三：pthreads / parallel 擴充套件 - 真正的多執行緒

pthreads (PHP 7.x, 需 ZTS 編譯)

# 安裝需求：PHP 編譯需 --enable-maintainer-zts
pecl install pthreads

<?php
class WorkerThread extends Thread {
    public $data;
    public $result;

    public function __construct($data) {
        $this->data = $data;
    }

    public function run() {
        // 執行緒內執行的程式碼
        $this->result = $this->heavyCompute($this->data);
    }

    private function heavyCompute($n) {
        $sum = 0;
        for ($i = 0; $i < $n; $i++) $sum += $i;
        return $sum;
    }
}

// 建立並啟動多執行緒
$threads = [];
for ($i = 1; $i <= 4; $i++) {
    $threads[$i] = new WorkerThread(1000000 * $i);
    $threads[$i]->start();
}

// 等待所有完成
foreach ($threads as $t) {
    $t->join();
    echo "Thread 結果: {$t->result}\n";
}
?>

parallel (PHP 7.2+, 現代替代方案)

pecl install parallel

<?php
use parallel\Runtime;
use parallel\Future;

$runtimes = [];
$futures = [];

// 建立 4 個執行緒執行環境
for ($i = 0; $i < 4; $i++) {
    $runtimes[$i] = new Runtime();
    $futures[$i] = $runtimes[$i]->run(function($id) {
        $sum = 0;
        for ($j = 0; $j < 1000000 * ($id+1); $j++) $sum += $j;
        return "Runtime $id: $sum";
    }, [$i]);
}

// 取得結果
foreach ($futures as $i => $future) {
    echo $future->value() . "\n";
}
?>

方案四：外部程式/訊息佇列 (生產環境推薦)

選擇指南

關鍵提醒

• PHP-FPM 已是多進程模型：每個請求獨立進程，單請求內若需並發才需上述技術
• 避免在 Web 請求中 fork：會佔用 FPM 槽位、導致資源洩漏、難以除錯
• 背景任務請用 CLI 模式：php artisan queue:work、php worker.php
• 共享資料需同步：檔案鎖、Redis 鎖、Semaphore、資料庫樂觀鎖

PHP 並發處理的核心哲學：「進程隔離、訊息傳遞、共享儲存」。根據任務性質選擇合適方案，不要強行在單一請求中實現複雜多執行緒。