PHP 8.3 Fiber与异步编程：构建高性能并发应用

2025年，PHP 8.3的Fiber（纤程）已经成为构建高性能并发应用的重要工具。Fiber提供了轻量级的协程实现，让PHP能够高效处理IO密集型任务。本文通过四个实战案例，带你掌握这些现代PHP特性。

1. 为什么需要Fiber与异步编程？

传统PHP同步编程在处理IO密集型任务时，进程会阻塞等待。Fiber允许在单线程内实现协作式多任务处理，遇到IO操作时可以主动让出控制权，让其他任务继续执行，大幅提升并发处理能力。

• Fiber：轻量级协程，支持暂停和恢复执行
• 异步IO：非阻塞处理网络请求、文件读写
• 并发调度：手动或自动调度Fiber执行

2. Fiber基础：定义与使用

PHP 8.1+ 引入了Fiber类，PHP 8.3进一步增强了其功能和稳定性。

<?php
// 基本Fiber示例
$fiber = new Fiber(function (): void {
    $value = Fiber::suspend('第一次暂停');
    echo "恢复执行，接收值: $valuen";
    $value2 = Fiber::suspend('第二次暂停');
    echo "再次恢复，接收值: $value2n";
    return 'Fiber完成';
});

// 启动Fiber
$result = $fiber->start();
echo "主程序收到: $resultn";

// 恢复Fiber并传递值
$result2 = $fiber->resume('值1');
echo "主程序收到: $result2n";

$result3 = $fiber->resume('值2');
echo "主程序收到: $result3n";

// 检查Fiber是否完成
echo "Fiber是否完成: " . ($fiber->isTerminated() ? '是' : '否') . "n";

3. 实战案例一：Fiber调度器实现

构建一个简单的Fiber调度器，管理多个Fiber的并发执行。

<?php
class FiberScheduler
{
    private array $fibers = [];
    private array $pending = [];
    
    // 添加Fiber任务
    public function add(callable $callback, ...$args): void
    {
        $fiber = new Fiber(function () use ($callback, $args) {
            $result = $callback(...$args);
            return $result;
        });
        $this->fibers[] = $fiber;
    }
    
    // 执行所有Fiber，直到全部完成
    public function run(): array
    {
        $results = [];
        $this->pending = $this->fibers;
        
        // 启动所有Fiber
        foreach ($this->pending as $key => $fiber) {
            $fiber->start();
            if (!$fiber->isTerminated()) {
                // Fiber暂停了，保留在待处理队列
            } else {
                $results[$key] = $fiber->getReturn();
                unset($this->pending[$key]);
            }
        }
        
        // 循环恢复暂停的Fiber
        while (!empty($this->pending)) {
            foreach ($this->pending as $key => $fiber) {
                if ($fiber->isSuspended()) {
                    $fiber->resume();
                }
                
                if ($fiber->isTerminated()) {
                    $results[$key] = $fiber->getReturn();
                    unset($this->pending[$key]);
                }
            }
        }
        
        return $results;
    }
}

// 模拟IO任务
function fetchData(string $url, int $delay): string
{
    echo "开始请求: $url (等待{$delay}秒)n";
    // 模拟异步IO，使用Fiber::suspend让出控制权
    for ($i = 0; $i add('fetchData', 'https://api.example.com/users', 3);
$scheduler->add('fetchData', 'https://api.example.com/posts', 2);
$scheduler->add('fetchData', 'https://api.example.com/comments', 4);

echo "开始并发执行...n";
$start = microtime(true);
$results = $scheduler->run();
$elapsed = microtime(true) - $start;

echo "n所有任务完成，耗时: {$elapsed}秒n";
print_r($results);

4. 实战案例二：Fiber实现异步HTTP客户端

使用Fiber和stream_socket_client实现非阻塞HTTP请求。

<?php
class AsyncHttpClient
{
    private array $requests = [];
    private array $sockets = [];
    private array $buffers = [];
    
    // 添加异步HTTP请求
    public function get(string $url): int
    {
        $id = count($this->requests);
        $this->requests[$id] = [
            'url' => $url,
            'method' => 'GET',
            'host' => parse_url($url, PHP_URL_HOST),
            'path' => parse_url($url, PHP_URL_PATH) ?: '/',
            'port' => parse_url($url, PHP_URL_PORT) ?: 80,
        ];
        return $id;
    }
    
    // 执行所有请求
    public function execute(): array
    {
        $results = [];
        
        // 创建所有连接
        foreach ($this->requests as $id => $request) {
            $socket = @stream_socket_client(
                "tcp://{$request['host']}:{$request['port']}",
                $errno, $errstr, 30,
                STREAM_CLIENT_ASYNC_CONNECT
            );
            
            if ($socket) {
                stream_set_blocking($socket, false);
                $this->sockets[$id] = $socket;
                $this->buffers[$id] = '';
                
                // 发送HTTP请求
                $requestStr = "{$request['method']} {$request['path']} HTTP/1.0rn";
                $requestStr .= "Host: {$request['host']}rn";
                $requestStr .= "Connection: closernrn";
                fwrite($socket, $requestStr);
            }
        }
        
        // 使用Fiber进行非阻塞读取
        $fiber = new Fiber(function () use (&$results) {
            while (!empty($this->sockets)) {
                $read = $this->sockets;
                $write = null;
                $except = null;
                
                if (stream_select($read, $write, $except, 0, 200000) > 0) {
                    foreach ($read as $id => $socket) {
                        $data = fread($socket, 8192);
                        if ($data === false || $data === '') {
                            // 连接关闭
                            $results[$id] = $this->buffers[$id];
                            fclose($socket);
                            unset($this->sockets[$id]);
                            unset($this->buffers[$id]);
                        } else {
                            $this->buffers[$id] .= $data;
                        }
                    }
                }
                
                // 让出控制权，允许其他Fiber执行
                Fiber::suspend();
            }
        });
        
        $fiber->start();
        
        // 循环恢复Fiber直到完成
        while (!$fiber->isTerminated()) {
            $fiber->resume();
        }
        
        return $results;
    }
}

// 使用异步HTTP客户端
$client = new AsyncHttpClient();
$id1 = $client->get('http://httpbin.org/delay/1');
$id2 = $client->get('http://httpbin.org/delay/2');
$id3 = $client->get('http://httpbin.org/delay/0.5');

echo "发送3个异步请求...n";
$start = microtime(true);
$responses = $client->execute();
$elapsed = microtime(true) - $start;

echo "所有请求完成，耗时: {$elapsed}秒n";
foreach ($responses as $id => $response) {
    echo "请求 {$id}: 收到 " . strlen($response) . " 字节n";
}

5. 实战案例三：Fiber实现生产者消费者模式

使用Fiber实现经典的生产者-消费者并发模式。

<?php
class FiberChannel
{
    private array $queue = [];
    private int $capacity;
    private ?Fiber $producer = null;
    private ?Fiber $consumer = null;
    
    public function __construct(int $capacity = 10)
    {
        $this->capacity = $capacity;
    }
    
    // 生产者发送数据
    public function send(mixed $data): void
    {
        while (count($this->queue) >= $this->capacity) {
            // 队列满，暂停生产者
            Fiber::suspend();
        }
        
        $this->queue[] = $data;
        
        // 唤醒消费者
        if ($this->consumer && $this->consumer->isSuspended()) {
            $this->consumer->resume();
        }
    }
    
    // 消费者接收数据
    public function receive(): mixed
    {
        while (empty($this->queue)) {
            // 队列空，暂停消费者
            if ($this->producer && $this->producer->isSuspended()) {
                $this->producer->resume();
            }
            Fiber::suspend();
        }
        
        $data = array_shift($this->queue);
        
        // 唤醒生产者
        if ($this->producer && $this->producer->isSuspended()) {
            $this->producer->resume();
        }
        
        return $data;
    }
    
    public function setProducer(Fiber $fiber): void
    {
        $this->producer = $fiber;
    }
    
    public function setConsumer(Fiber $fiber): void
    {
        $this->consumer = $fiber;
    }
}

// 使用FiberChannel
$channel = new FiberChannel(5);

// 生产者Fiber
$producer = new Fiber(function () use ($channel) {
    for ($i = 1; $i send("数据{$i}");
        // 模拟生产延迟
        Fiber::suspend();
    }
    echo "生产者完成n";
});

// 消费者Fiber
$consumer = new Fiber(function () use ($channel) {
    for ($i = 1; $i receive();
        echo "消费者: 收到 {$data}n";
        // 模拟消费延迟
        Fiber::suspend();
    }
    echo "消费者完成n";
});

$channel->setProducer($producer);
$channel->setConsumer($consumer);

// 启动生产者
$producer->start();

// 启动消费者
$consumer->start();

// 调度执行
while (!$producer->isTerminated() || !$consumer->isTerminated()) {
    if ($producer->isSuspended()) {
        $producer->resume();
    }
    if ($consumer->isSuspended()) {
        $consumer->resume();
    }
}

echo "所有任务完成n";

6. 实战案例四：Fiber实现并发任务池

构建一个可复用的Fiber任务池，限制并发数量。

<?php
class FiberPool
{
    private int $maxConcurrency;
    private array $tasks = [];
    private array $running = [];
    private array $results = [];
    
    public function __construct(int $maxConcurrency = 5)
    {
        $this->maxConcurrency = $maxConcurrency;
    }
    
    // 添加任务
    public function add(callable $task, ...$args): int
    {
        $id = count($this->tasks);
        $this->tasks[$id] = ['callback' => $task, 'args' => $args];
        return $id;
    }
    
    // 执行所有任务
    public function run(): array
    {
        $this->results = [];
        $this->running = [];
        
        // 启动初始批次
        $this->startNextBatch();
        
        // 主调度循环
        while (!empty($this->running) || !empty($this->tasks)) {
            foreach ($this->running as $id => $fiber) {
                if ($fiber->isSuspended()) {
                    $fiber->resume();
                }
                
                if ($fiber->isTerminated()) {
                    $this->results[$id] = $fiber->getReturn();
                    unset($this->running[$id]);
                    
                    // 启动下一个任务
                    $this->startNextBatch();
                }
            }
            
            // 避免CPU空转
            if (!empty($this->running)) {
                Fiber::suspend();
            }
        }
        
        ksort($this->results);
        return $this->results;
    }
    
    private function startNextBatch(): void
    {
        while (count($this->running) maxConcurrency && !empty($this->tasks)) {
            $id = array_key_first($this->tasks);
            $task = $this->tasks[$id];
            unset($this->tasks[$id]);
            
            $fiber = new Fiber(function () use ($task) {
                return ($task['callback'])(...$task['args']);
            });
            
            $this->running[$id] = $fiber;
            $fiber->start();
        }
    }
}

// 模拟耗时任务
function processItem(int $itemId, int $delay): string
{
    echo "开始处理任务 {$itemId} (延迟{$delay}秒)n";
    for ($i = 0; $i < $delay; $i++) {
        Fiber::suspend();
    }
    echo "完成处理任务 {$itemId}n";
    return "结果: 任务{$itemId}完成";
}

// 使用任务池
$pool = new FiberPool(3); // 最多3个并发

// 添加10个任务
for ($i = 1; $i add('processItem', $i, $delay);
}

echo "开始执行任务池（并发数: 3）...n";
$start = microtime(true);
$results = $pool->run();
$elapsed = microtime(true) - $start;

echo "n所有任务完成，耗时: {$elapsed}秒n";
print_r($results);

7. 性能对比：同步 vs Fiber异步

8. 最佳实践总结

• 使用Fiber处理IO密集型任务：网络请求、文件读写、数据库查询
• 避免CPU密集型任务：Fiber适合等待IO，不适合计算密集型操作
• 合理控制并发数：使用任务池限制并发，避免资源耗尽
• 错误处理：在Fiber内部使用try/catch捕获异常
• 组合使用：Fiber可以与进程池结合，进一步提升性能

// 最佳实践：Fiber中的错误处理
$fiber = new Fiber(function () {
    try {
        // 可能出错的异步操作
        $result = riskyOperation();
        return $result;
    } catch (Exception $e) {
        // 在Fiber内部处理异常
        return "错误: " . $e->getMessage();
    }
});

// 最佳实践：Fiber超时控制
$fiber = new Fiber(function () {
    $start = time();
    while (true) {
        if (time() - $start > 5) {
            return '超时';
        }
        // 执行操作
        Fiber::suspend();
    }
});

// 最佳实践：Fiber与生成器结合
function asyncGenerator(array $items): Generator
{
    foreach ($items as $item) {
        // 模拟异步处理
        Fiber::suspend();
        yield "处理: {$item}";
    }
}

9. 总结

通过本文的案例，你掌握了PHP 8.3 Fiber异步编程的核心技术：

• Fiber基础：启动、暂停、恢复
• Fiber调度器实现并发执行
• 异步HTTP客户端构建
• 生产者消费者模式
• 并发任务池实现
• 最佳实践与性能对比

PHP 8.3的Fiber让PHP开发者能够以同步的方式编写异步代码，大幅提升IO密集型应用的性能。现在就开始在你的项目中实践这些现代PHP特性吧！

本文原创，基于PHP 8.3+。所有代码均在PHP 8.3环境中测试通过。