PHP 利用 Socket 實現多執行緒模擬技術

PHP 利用 Socket 實現多執行緒模擬技術

PHP 本身不支援多執行緒，但 Web 伺服器（Apache/Nginx+PHP-FPM）本質上是多執行緒/多進程架構。利用此特性，透過 fsockopen() 發起非阻塞 HTTP 請求，可模擬多執行緒並行處理。本文詳細介紹原理與實作。

核心原理

1. 主腳本透過 fsockopen() 連接本機 Web 伺服器
2. 發送 HTTP GET 請求觸發子腳本執行（帶參數標識）
3. 立即關閉連接，不等待回應（非阻塞）
4. Web 伺服器並行處理多個子請求

基礎單執行緒程式碼

$fp = fsockopen($_SERVER['HTTP_HOST'], 80, $errno, $errstr, 5);
if (!$fp) {
    echo "$errstr ($errno)
\n";
} else {
    fputs($fp, "GET {$_SERVER['PHP_SELF']}?flag=1\r\n");
    fclose($fp);
}

封裝多執行緒類

<?php
/**
 * PHP 多執行緒模擬類
 * @author axgle
 */
class Thread {
    private $count;

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

    // 提交多個執行緒
    private function _submit() {
        for ($i = 1; $i <= $this->count; $i++) {
            $this->_thread();
        }
        return true;
    }

    // 單個執行緒請求
    private function _thread() {
        $fp = fsockopen($_SERVER['HTTP_HOST'], 80, $errno, $errstr, 5);
        if (!$fp) {
            echo "$errstr ($errno)
\n";
            return;
        }
        // 發送請求，帶 flag 參數區分主/子執行緒
        fputs($fp, "GET {$_SERVER['PHP_SELF']}?flag=1\r\n");
        fclose($fp);
    }

    // 執行入口
    public function exec($func) {
        // 若有 flag 參數，執行業務函數（子執行緒）
        // 否則提交多執行緒（主執行緒）
        isset($_GET['flag']) ? call_user_func($func) : $this->_submit();
    }
}

// 應用範例
$th = new Thread(10); // 模擬 10 個執行緒
$th->exec('demo');    // 執行 demo 函數

function demo() {
    // 子執行緒業務邏輯：建立以微秒命名的檔案
    fopen('data/' . microtime(), 'w');
}
?>

完整可測試範例

<?php
// thread_demo.php
class Thread {
    private $count;
    public function __construct($count = 1) { $this->count = $count; }
    private function _submit() {
        for ($i = 1; $i <= $this->count; $i++) $this->_thread();
        return true;
    }
    private function _thread() {
        $host = $_SERVER['HTTP_HOST'];
        $uri  = $_SERVER['PHP_SELF'];
        $fp = fsockopen($host, 80, $errno, $errstr, 5);
        if (!$fp) { echo "Connect failed: $errstr ($errno)\n"; return; }
        fputs($fp, "GET $uri?thread=1\r\n");
        fclose($fp);
    }
    public function exec($func) {
        if (isset($_GET['thread'])) {
            call_user_func($func);
        } else {
            $this->_submit();
            echo "Main thread: submitted {$this->count} threads\n";
        }
    }
}

// 使用
$th = new Thread(5);
$th->exec('worker');

function worker() {
    $id = uniqid();
    file_put_contents("logs/worker_$id.log", "Worker $id started at " . microtime(true) . "\n");
    sleep(rand(1, 3)); // 模擬耗時任務
    file_put_contents("logs/worker_$id.log", "Worker $id finished at " . microtime(true) . "\n", FILE_APPEND);
}
?>

進階改良：非阻塞 + 超時控制

class AsyncThread {
    private $threads = [];
    private $timeout = 30;

    public function add(callable $func, array $args = []) {
        $this->threads[] = [$func, $args];
    }

    public function execute() {
        $host = $_SERVER['HTTP_HOST'];
        $uri  = $_SERVER['PHP_SELF'];
        $sockets = [];

        // 發起所有請求
        foreach ($this->threads as $i => [$func, $args]) {
            $params = http_build_query(array_merge(['async_id' => $i], $args));
            $fp = @fsockopen($host, 80, $errno, $errstr, 2);
            if ($fp) {
                stream_set_blocking($fp, 0); // 非阻塞
                stream_set_timeout($fp, $this->timeout);
                fwrite($fp, "POST $uri HTTP/1.1\r\n");
                fwrite($fp, "Host: $host\r\n");
                fwrite($fp, "Content-Type: application/x-www-form-urlencoded\r\n");
                fwrite($fp, "Content-Length: " . strlen($params) . "\r\n");
                fwrite($fp, "Connection: Close\r\n\r\n");
                fwrite($fp, $params);
                $sockets[$i] = $fp;
            }
        }

        // 等待完成（簡易輪詢）
        $start = time();
        while ($sockets && (time() - $start < $this->timeout)) {
            $read = $write = $except = $sockets;
            if (stream_select($read, $write, $except, 0, 500000) > 0) {
                foreach ($read as $fp) {
                    $meta = stream_get_meta_data($fp);
                    if ($meta['eof'] || $meta['timed_out']) {
                        fclose($fp);
                        unset($sockets[(int)$fp]);
                    }
                }
            }
        }
        // 強制關閉剩餘
        foreach ($sockets as $fp) fclose($fp);
    }
}

// 使用
$async = new AsyncThread();
for ($i = 0; $i < 10; $i++) {
    $async->add('heavy_task', ['id' => $i]);
}
$async->execute();

方案比較與選擇建議

注意事項與限制

• 回傳值難獲取：fsockopen 發送後即關閉，需透過資料庫、檔案、Redis 等外部儲存交換資料
• 連接數限制：Web 伺服器最大連接數、PHP-FPM 進程數限制並發上限
• 安全性：避免 SSRF 攻擊，驗證請求來源，生產環境建議用內網 IP 或 Unix Socket
• Session 衝突：子請求會啟動新 Session，避免寫入衝突（session_write_close()）
• 超時處理：設定合理的連接/讀寫超時，避免主進程卡死

現代替代方案推薦

新專案強烈建議使用 Swoole 或 OpenSwoole 擴展，提供原生協程、Channel、Process Pool 等高階並行原語，效能遠超 Socket 模擬。若無法安裝擴展，可考慮 ReactPHP 或 Amp 等純 PHP 非阻塞框架。

Socket 模擬多執行緒是 PHP 早期的權宜之計，理解其原理有助於掌握 PHP 並行處理的演進脈絡，但現代專案應優先選擇成熟的擴展或框架方案。