PHP项目如何实现多线程处理？

wen PHP项目 2026-06-18 4

本文目录导读：

PHP项目如何实现多线程处理？

pthreads 扩展（PHP 7.x 及以下）
pcntl_fork（进程方式）
使用 proc_open 或 shell_exec
使用并行库（推荐）
使用消息队列（适合复杂场景）
选择建议
注意事项

在PHP中实现多线程处理主要有以下几种常见方案：

pthreads 扩展（PHP 7.x 及以下）

最直接的多线程方案，但仅适用于 PHP 7.x 及以下版本，且需要在 CLI 模式下运行。

// 安装：pecl install pthreads
class MyThread extends Thread {
    private $data;
    public function __construct($data) {
        $this->data = $data;
    }
    public function run() {
        // 线程执行代码
        echo "Processing: " . $this->data . "\n";
        sleep(1);
    }
}
// 创建多个线程
$threads = [];
for ($i = 0; $i < 5; $i++) {
    $threads[$i] = new MyThread("Task " . $i);
    $threads[$i]->start();
}
// 等待所有线程完成
foreach ($threads as $thread) {
    $thread->join();
}

pcntl_fork（进程方式）

使用进程创建子进程，PHP 8.x 中兼容性更好。

function doWork($taskId) {
    echo "Task {$taskId} started\n";
    sleep(2);
    echo "Task {$taskId} completed\n";
}
for ($i = 1; $i <= 5; $i++) {
    $pid = pcntl_fork();
    if ($pid == -1) {
        die("Could not fork");
    } elseif ($pid == 0) {
        // 子进程
        doWork($i);
        exit(0);
    }
}
// 等待所有子进程结束
while (pcntl_waitpid(0, $status) != -1) {
    // 等待所有子进程
}

使用 proc_open 或 shell_exec

通过命令行启动 PHP 脚本实现并行处理：

function parallelProcess($scripts) {
    $processes = [];
    foreach ($scripts as $index => $script) {
        $processes[$index] = popen("php {$script}", "r");
    }
    // 读取输出结果
    foreach ($processes as $index => $process) {
        $output = stream_get_contents($process);
        echo "Process {$index}: {$output}\n";
        pclose($process);
    }
}
// 使用示例
$scripts = [
    'worker1.php',
    'worker2.php', 
    'worker3.php'
];
parallelProcess($scripts);

使用并行库（推荐）

Amp 异步框架

// 安装：composer require amphp/parallel
use Amp\Parallel\Worker;
use Amp\Promise;
$functions = [
    function() { return "Task 1"; },
    function() { return "Task 2"; },
    function() { return "Task 3"; }
];
$promises = [];
foreach ($functions as $index => $fn) {
    $promises[$index] = Worker\enqueueCallable($fn);
}
$results = Promise\wait(Promise\all($promises));
print_r($results);

Swoole 协程（生产环境推荐）

// 安装：pecl install swoole
use Swoole\Coroutine;
Co\run(function() {
    $tasks = ['Task1', 'Task2', 'Task3'];
    $coroutines = [];
    foreach ($tasks as $task) {
        $coroutines[] = Coroutine::create(function() use ($task) {
            echo "{$task} started\n";
            Coroutine::sleep(1);
            echo "{$task} completed\n";
        });
    }
});

使用消息队列（适合复杂场景）

// 使用 RabbitMQ 或 Redis 队列实现
class MultiThreadProcessor {
    private $redis;
    public function __construct() {
        $this->redis = new Redis();
        $this->redis->connect('127.0.0.1', 6379);
    }
    public function dispatchJobs($jobs, $workerCount = 3) {
        // 分发作业到队列
        foreach ($jobs as $job) {
            $this->redis->lPush('job_queue', json_encode($job));
        }
        // 启动多个 worker
        for ($i = 0; $i < $workerCount; $i++) {
            $this->startWorker("worker_{$i}.php");
        }
    }
    private function startWorker($workerScript) {
        $pid = pcntl_fork();
        if ($pid == 0) {
            // worker.php 会从队列中消费作业
            exec("php {$workerScript}");
            exit(0);
        }
    }
}