PHP8.2新特性实战：纤程(Fibers)实现高性能并发编程

一、纤程核心原理

轻量级用户态线程的PHP实现方案：

// 基本纤程使用示例
$fiber = new Fiber(function() {
    echo "纤程开始执行n";
    Fiber::suspend();
    echo "纤程恢复执行n";
    return "结果数据";
});

echo "启动纤程n";
$fiber->start();

echo "主线程继续执行n";
sleep(1);

echo "恢复纤程n";
$value = $fiber->resume();
echo "纤程返回值: ".$value."n";

// 输出顺序：
// 启动纤程
// 纤程开始执行
// 主线程继续执行
// 恢复纤程
// 纤程恢复执行
// 纤程返回值: 结果数据

核心优势：轻量级、无堆栈切换开销、同步编码风格、高并发支持

二、高级应用模式

1. HTTP并发请求

// 并发请求处理器
class HttpFiberManager {
    private array $fibers = [];
    
    public function addRequest(string $url): void {
        $this->fibers[] = new Fiber(function() use ($url) {
            $ch = curl_init($url);
            curl_setopt($ch, CURLOPT_RETURNTRANSFER, true);
            
            Fiber::suspend(); // 暂停等待调度
            
            $response = curl_exec($ch);
            curl_close($ch);
            
            return $response;
        });
    }
    
    public function run(): array {
        // 启动所有纤程
        foreach ($this->fibers as $fiber) {
            $fiber->start();
        }
        
        // 执行所有请求
        $running = true;
        while ($running) {
            $running = false;
            
            foreach ($this->fibers as $fiber) {
                if ($fiber->isTerminated()) continue;
                
                $fiber->resume();
                $running = $running || !$fiber->isTerminated();
            }
            
            usleep(1000); // 避免CPU空转
        }
        
        // 收集结果
        return array_map(fn($f) => $f->getReturn(), $this->fibers);
    }
}

// 使用示例
$manager = new HttpFiberManager();
$manager->addRequest('https://api.example.com/data1');
$manager->addRequest('https://api.example.com/data2');
$results = $manager->run();

2. 数据库连接池

// 纤程安全连接池
class FiberConnectionPool {
    private SplQueue $pool;
    private int $maxSize;
    
    public function __construct(
        private Closure $connector,
        int $maxSize = 10
    ) {
        $this->maxSize = $maxSize;
        $this->pool = new SplQueue();
    }
    
    public function getConnection(): mixed {
        if (!$this->pool->isEmpty()) {
            return $this->pool->dequeue();
        }
        
        // 新建连接
        return ($this->connector)();
    }
    
    public function releaseConnection($conn): void {
        if ($this->pool->count() maxSize) {
            $this->pool->enqueue($conn);
        } else {
            // 关闭超出的连接
            if (method_exists($conn, 'close')) {
                $conn->close();
            }
        }
    }
}

// 使用示例
$pool = new FiberConnectionPool(fn() => new PDO(...));
Fiber::suspend();

$db = $pool->getConnection();
try {
    $stmt = $db->query('SELECT * FROM users');
    $results = $stmt->fetchAll();
} finally {
    $pool->releaseConnection($db);
}

三、Web应用实战案例

1. 高并发API网关

// API网关处理器
class ApiGateway {
    public function handle(array $endpoints): array {
        $fiber = new Fiber(function() use ($endpoints) {
            $results = [];
            $fibers = [];
            
            // 创建子纤程
            foreach ($endpoints as $key => $url) {
                $fibers[$key] = new Fiber(function() use ($url) {
                    return $this->fetchApi($url);
                });
                $fibers[$key]->start();
            }
            
            // 调度子纤程
            do {
                $allDone = true;
                
                foreach ($fibers as $key => $fiber) {
                    if ($fiber->isTerminated()) continue;
                    
                    $fiber->resume();
                    $allDone = $allDone && $fiber->isTerminated();
                    
                    if ($fiber->isTerminated()) {
                        $results[$key] = $fiber->getReturn();
                    }
                }
                
                Fiber::suspend();
            } while (!$allDone);
            
            return $results;
        });
        
        $fiber->start();
        
        while (!$fiber->isTerminated()) {
            $fiber->resume();
        }
        
        return $fiber->getReturn();
    }
    
    private function fetchApi(string $url): array {
        $ch = curl_init($url);
        // ... curl配置
        Fiber::suspend();
        return json_decode(curl_exec($ch), true);
    }
}

// 使用示例
$gateway = new ApiGateway();
$responses = $gateway->handle([
    'user' => 'https://api.example.com/user/123',
    'orders' => 'https://api.example.com/orders?user=123'
]);