Parallel 扩展

概述

Parallel 扩展是 PHP 官方提供的并行处理扩展，允许在 PHP 中创建真正的并行执行线程。与 Swoole 的协程（单线程异步）不同，Parallel 扩展利用操作系统的线程来实现真正的多线程并行计算。它提供了 Runtime（运行时）、Future（异步结果）、Channel（通道）等抽象，使得 PHP 多线程编程更加安全和便捷。

PHP 版本要求

Parallel 扩展自 PHP 8.0 引入。本文基于 PHP 8.1+ 编写。Parallel 需要 ZTS（Zend Thread Safety）版本的 PHP。

PHP 版本要求

Parallel 扩展不支持 Windows 和 macOS。仅支持 Linux 和 FreeBSD。此外，需要使用 ZTS 版本的 PHP。

基础概念

Parallel vs 多进程 vs 协程

特性	Parallel（多线程）	pcntl（多进程）	Swoole（协程）
执行方式	真正并行	真正并行	协作式多任务
通信方式	共享内存（直接）	管道/共享内存	Channel
内存开销	小（共享地址空间）	大（独立地址空间）	极小
创建开销	中	大	极小
数据共享	直接读取	需序列化	直接读取
安全性	需同步控制	进程隔离安全	单线程安全

线程安全

在 Parallel 扩展中，PHP 对象不能直接在线程间共享。只能在 parallel\Runtime 的闭包中访问序列化后的数据副本。

安装与配置

# 通过 PECL 安装
pecl install parallel

# 验证
php -m | grep parallel
php --ri parallel

ZTS 版本要求

Parallel 需要 ZTS 版本的 PHP。在 Ubuntu 上：

# 安装 ZTS 版本
apt-get install php8.1-zts

详细说明

parallel\Runtime

Runtime 是 Parallel 的核心类，代表一个独立的 PHP 运行时线程。

<?php
declare(strict_types=1);

use parallel\Runtime;

// 创建运行时（默认自动检测线程数）
$runtime = new Runtime();

// 创建指定线程数的运行时
$runtime = new Runtime(Runtime::DEFAULT);

// 使用闭包执行任务
$future = $runtime->run(function (): string {
    return "线程 " . Thread::getCurrentThreadId() . ": " . date('H:i:s');
});

// 获取结果（阻塞等待）
echo $future->value() . PHP_EOL;

传递参数和引用

<?php
declare(strict_types=1);

use parallel\Runtime;

$runtime = new Runtime(4);

// 传递参数（自动序列化）
$future = $runtime->run(function (string $name, int $count): string {
    return "处理 {$name}: 完成 {$count} 项任务";
}, ['order_processor', 42]);

echo $future->value() . PHP_EOL;

加载文件

<?php
declare(strict_types=1);

use parallel\Runtime;

$runtime = new Runtime(4, [
    'bootstrap' => __DIR__ . '/bootstrap.php', // 在新线程中预加载文件
]);

// bootstrap.php 中可以加载 Composer autoloader

parallel\Future

Future 代表异步执行的结果，支持阻塞和非阻塞的结果获取。

<?php
declare(strict_types=1);

use parallel\Runtime;
use parallel\Future;

$runtime = new Runtime(4);

// 提交多个任务
$futures = [];

for ($i = 0; $i < 10; $i++) {
    $num = $i;
    $futures[$i] = $runtime->run(function () use ($num): int {
        usleep(mt_rand(100000, 500000));
        return $num * $num;
    });
}

// 阻塞等待所有结果
$results = [];
foreach ($futures as $i => $future) {
    $results[$i] = $future->value();
    echo "任务 {$i}: 结果 = {$results[$i]}" . PHP_EOL;
}

echo "所有任务完成" . PHP_EOL;

Future 状态检查

<?php
declare(strict_types=1);

use parallel\Runtime;

$runtime = new Runtime(4);

$future = $runtime->run(function (): string {
    sleep(3);
    return "完成";
});

// 非阻塞检查状态
while (!$future->done()) {
    echo "等待中..." . PHP_EOL;
    usleep(500000);
}

// 获取结果
echo "结果: " . $future->value() . PHP_EOL;

parallel\Channel

Channel 是线程间的通信通道，支持数据的发送和接收。

<?php
declare(strict_types=1);

use parallel\Channel;

// 创建缓冲通道（缓冲区大小为 4）
$channel = new Channel(4);

// 发送数据
$channel->send('hello');
$channel->send(42);
$channel->send(['key' => 'value']);

// 接收数据
echo $channel->recv() . PHP_EOL;    // 'hello'
echo $channel->recv() . PHP_EOL;    // 42
print_r($channel->recv());          // ['key' => 'value']

// 关闭通道
$channel->close();

Channel 同步

<?php
declare(strict_types=1);

use parallel\Runtime;
use parallel\Channel;

$runtime = new Runtime(1);

$channel = new Channel();

// 生产者-消费者模式
$runtime->run(function (): void {
    for ($i = 0; $i < 5; $i++) {
        $channel->send("消息 {$i}");
        usleep(500000);
    }
    $channel->send(null); // 发送结束标记
});

// 消费者
while (true) {
    $message = $channel->recv();
    if ($message === null) {
        echo "接收完成" . PHP_EOL;
        break;
    }
    echo "收到: {$message}" . PHP_EOL;
}

$channel->close();

并行数据处理

<?php
declare(strict_types=1);

use parallel\Runtime;
use parallel\Future;

/**
 * 并行数组处理
 */
class ParallelProcessor
{
    private Runtime $runtime;

    public function __construct(int $threads = 4)
    {
        $this->runtime = new Runtime($threads);
    }

    /**
     * 并行 map
     */
    public function map(array $items, callable $fn): array
    {
        $futures = [];

        foreach ($items as $i => $item) {
            $futures[$i] = $this->runtime->run(function () use ($fn, $item): mixed {
                return $fn($item);
            });
        }

        $results = [];
        foreach ($futures as $i => $future) {
            $results[$i] = $future->value();
        }

        return $results;
    }

    /**
     * 并行 foreach
     */
    public function each(array $items, callable $fn): void
    {
        $futures = [];
        foreach ($items as $i => $item) {
            $futures[$i] = $this->runtime->run(function () use ($fn, $item): void {
                $fn($item);
            });
        }

        // 等待所有完成
        foreach ($futures as $future) {
            $future->value();
        }
    }

    /**
     * 并行 filter
     */
    public function filter(array $items, callable $fn): array
    {
        $result = $this->map($items, fn($item) => [
            'item' => $item,
            'keep' => $fn($item),
        ]);

        return array_column(
            array_filter($result, fn(array $r) => $r['keep']),
            'item'
        );
    }
}

// 使用示例
$processor = new ParallelProcessor(4);

// 并行计算平方
$numbers = range(1, 20);
$squares = $processor->map($numbers, fn(int $n): int => $n * $n);
print_r($squares);

// 并行过滤偶数
$evens = $processor->filter($numbers, fn(int $n): bool => $n % 2 === 0);
print_r($evens);

实战示例

并行 HTTP 请求

<?php
declare(strict_types=1);

use parallel\Runtime;
use parallel\Future;

/**
 * 多线程并发 HTTP 请求
 */
function parallelHttpRequests(array $urls): array
{
    $runtime = new Runtime(4);
    $futures = [];

    foreach ($urls as $i => $url) {
        $futures[$i] = $runtime->run(function () use ($url): array {
            $start = microtime(true);
            $context = stream_context_create(['http' => ['timeout' => 10]]);
            $content = @file_get_contents($url, false, $context);
            $time = (microtime(true) - $start) * 1000;

            return [
                'url' => $url,
                'success' => $content !== false,
                'time_ms' => round($time, 2),
                'size' => strlen((string) $content),
            ];
        });
    }

    $results = [];
    foreach ($futures as $i => $future) {
        $results[$i] = $future->value();
    }

    return $results;
}

$urls = array_fill(0, 10, 'https://httpbin.org/get');
$startTime = microtime(true);
$results = parallelHttpRequests($urls);
$totalTime = round((microtime(true) - $startTime) * 1000, 2);

echo "10 个并发请求，总耗时: {$totalTime}ms" . PHP_EOL;
foreach ($results as $result) {
    echo "  {$result['url']} → " . ($result['success'] ? 'OK' : 'FAIL')
        . " ({$result['time_ms']}ms)" . PHP_EOL;
}

并行图像处理

<?php
declare(strict_types=1);

use parallel\Runtime;

/**
 * 并行图片缩略图生成
 */
class ParallelImageProcessor
{
    private Runtime $runtime;

    public function __construct(int $threads = 4)
    {
        $this->runtime = new Runtime($threads, [
            'bootstrap' => __DIR__ . '/vendor/autoload.php',
        ]);
    }

    /**
     * 并行生成缩略图
     */
    public function generateThumbnails(
        array $images,
        string $outputDir,
        int $thumbWidth = 200,
        int $thumbHeight = 200,
    ): array {
        if (!is_dir($outputDir)) {
            mkdir($outputDir, 0755, true);
        }

        $futures = [];
        foreach ($images as $i => $imagePath) {
            $futures[$i] = $this->runtime->run(
                function (string $src, string $dst, int $w, int $h): array {
                    if (!file_exists($src)) {
                        return ['file' => $src, 'success' => false, 'error' => '文件不存在'];
                    }

                    [$width, $height] = getimagesize($src);
                    $thumb = imagecreatetruecolor($w, $h);

                    switch (strtolower(pathinfo($src, PATHINFO_EXTENSION))) {
                        case 'jpg':
                        case 'jpeg':
                            $source = imagecreatefromjpeg($src);
                            break;
                        case 'png':
                            $source = imagecreatefrompng($src);
                            break;
                        default:
                            return ['file' => $src, 'success' => false, 'error' => '不支持的格式'];
                    }

                    imagecopyresampled($thumb, $source, 0, 0, 0, 0, $w, $h, $width, $height);
                    $filename = basename($src);
                    imagejpeg($thumb, $dst . '/' . $filename, 85);
                    imagedestroy($thumb);
                    imagedestroy($source);

                    return ['file' => $filename, 'success' => true];
                },
                [$imagePath, $outputDir, $thumbWidth, $thumbHeight]
            );
        }

        $results = [];
        foreach ($futures as $i => $future) {
            $results[$i] = $future->value();
        }

        return $results;
    }
}

注意事项

限制和约束

1. 不可共享对象：PHP 对象不能直接在线程间共享，必须序列化
2. 不可共享资源：数据库连接、文件句柄等不能跨线程共享
3. 无共享状态：每个线程有独立的 PHP 解释器状态
4. 安全限制：某些 PHP 扩展可能不是线程安全的

线程安全扩展

线程安全扩展列表

以下扩展在 Parallel 中通常是安全的：

• SPL（Standard PHP Library）
• JSON
• PCRE（正则表达式）
• Hash

以下扩展可能不安全：

• MySQLi（使用 mysqlnd 通常是安全的）
• GD
• OpenSSL

最佳实践

1. 合理控制线程数

<?php
// 通常设置为 CPU 核心数
$threads = (int) (`nproc` ?? 4);

// CPU 密集型任务
$runtime = new Runtime($threads);

// I/O 密集型任务（可以多一些线程）
$runtime = new Runtime($threads * 2);

2. 使用 Channel 进行协调

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

$channel = new Channel();
$runtime = new Runtime(4);

// Worker 线程
$runtime->run(function () use ($channel): void {
    while (($task = $channel->recv()) !== null) {
        // 处理任务
        $result = process($task);
        $channel->send($result);
    }
});

3. 错误处理

<?php
use parallel\Runtime;

$runtime = new Runtime();
$future = $runtime->run(function (): int {
    throw new RuntimeException("测试错误");
});

try {
    $future->value();
} catch (\parallel\Error\Error $e) {
    echo "线程错误: " . $e->getMessage() . PHP_EOL;
} catch (\parallel\Future\Error\Killed $e) {
    echo "线程被终止" . PHP_EOL;
}

下一节

继续学习：PHP 扩展概览

参考链接

• PHP 官方文档 - Parallel
• parallel - GitHub
• PHP Internals - Thread Safe