PHP小码农

Appearance

Fiber 创建与启动 ​

概述 ​

创建和启动 Fiber 是使用纤程的第一步。本章将详细讲解如何通过 new Fiber() 创建纤程实例、使用 Fiber::start() 启动执行、通过 Fiber::suspend() 暂停纤程、以及通过 Fiber::resume() 恢复纤程的执行。掌握这些核心操作是使用 Fiber 进行异步编程的基础。

核心方法

Fiber::start() → Fiber::suspend() → Fiber::resume() 是 Fiber 最基本的使用模式。

基础概念 ​

new Fiber($callback) ​

创建一个新的 Fiber 实例。回调函数是纤程的入口点，当纤程启动时会执行这个回调。

Fiber::start() ​

启动纤程的执行。可以向回调函数传递初始参数。纤程开始执行直到遇到 Fiber::suspend() 或回调结束。

Fiber::suspend() ​

在纤程内部调用，暂停纤程的执行。可以传递一个值给调用者（resume 的返回值）。

Fiber::resume() ​

从主线程恢复暂停的纤程。可以传递一个值给纤程（suspend 的返回值）。

语法与代码 ​

创建并启动 Fiber ​

php
<?php
declare(strict_types=1);

// 创建 Fiber
$fiber = new Fiber(function (): void {
    echo "Fiber 开始执行\n";
    Fiber::suspend();
    echo "Fiber 继续执行\n";
});

echo "1. Fiber 状态: " . $fiber->status() . "\n"; // INIT (1)

// 启动 Fiber
$fiber->start();
echo "2. Fiber 状态: " . $fiber->status() . "\n"; // SUSPENDED (3)

// 恢复 Fiber
$fiber->resume();
echo "3. Fiber 状态: " . $fiber->status() . "\n"; // FINISHED (4)

Fiber::start() 传参 ​

php
<?php
declare(strict_types=1);

$fiber = new Fiber(function (string $name, int $age): string {
    $message = "姓名: {$name}, 年龄: {$age}";
    Fiber::suspend($message);
    return "处理完成";
});

$result = $fiber->start('Alice', 30);
echo $result . "\n"; // 姓名: Alice, 年龄: 30

$finalResult = $fiber->resume();
echo $finalResult . "\n"; // 处理完成

Fiber::suspend() 传值 ​

php
<?php
declare(strict_types=1);

$fiber = new Fiber(function (): void {
    for ($i = 1; $i <= 3; $i++) {
        $received = Fiber::suspend("第{$i}次暂停");
        echo "收到: {$received}\n";
    }
});

// 首次 start
$result = $fiber->start();
echo "暂停返回: {$result}\n"; // 第1次暂停

// 后续 resume
$result = $fiber->resume('数据A');
echo "暂停返回: {$result}\n"; // 第2次暂停
// 收到: 数据A

$result = $fiber->resume('数据B');
echo "暂停返回: {$result}\n"; // 第3次暂停
// 收到: 数据B

$fiber->resume('数据C');
// 收到: 数据C
// Fiber 自然结束

Fiber::resume() 传值 ​

php
<?php
declare(strict_types=1);

$fiber = new Fiber(function (): int {
    echo "Fiber 等待输入...\n";
    $input = Fiber::suspend();
    $result = (int) $input * 2;
    Fiber::suspend($result);
    return $result;
});

$fiber->start();
// Fiber 等待输入...

$doubled = $fiber->resume(21);
echo "计算结果: {$doubled}\n"; // 计算结果: 42

$fiber->resume();

Fiber 返回值 ​

php
<?php
declare(strict_types=1);

$fiber = new Fiber(function (): array {
    $items = [];
    while (count($items) < 3) {
        $items[] = Fiber::suspend(count($items) . ' items');
    }
    return $items;
});

$result = $fiber->start();  // 0 items
$result = $fiber->resume('a'); // 1 items
$result = $fiber->resume('b'); // 2 items
$final = $fiber->resume('c'); // Fiber 结束

print_r($final);
// Array ( [0] => a [1] => b [2] => c )

详细说明 ​

启动流程详解 ​

1. $fiber = new Fiber($callback)
   → 状态: INIT
   → 回调尚未执行

2. $fiber->start()
   → 状态: RUNNING
   → 回调开始执行
   → 遇到 Fiber::suspend($value)
   → 状态: SUSPENDED
   → start() 返回 $value

3. $fiber->resume($sendValue)
   → 状态: RUNNING
   → Fiber::suspend() 的返回值为 $sendValue
   → 回调继续执行
   → 遇到 Fiber::suspend($value2)
   → 状态: SUSPENDED
   → resume() 返回 $value2

4. 回调执行完毕（return 或结束）
   → 状态: FINISHED
   → resume() 返回回调的 return 值

错误状态下的方法调用 ​

php
<?php
declare(strict_types=1);

$fiber = new Fiber(function (): void {
    Fiber::suspend();
});

// 在 INIT 状态下调用 resume → FiberError
try {
    $fiber->resume();
} catch (\FiberError $e) {
    echo "错误: " . $e->getMessage() . "\n";
    // Cannot resume a fiber that has not been started
}

// 在 INIT 状态下调用 throw → FiberError
try {
    $fiber->throw(new \RuntimeException('test'));
} catch (\FiberError $e) {
    echo "错误: " . $e->getMessage() . "\n";
    // Cannot throw into a fiber that has not been started
}

// 正确流程
$fiber->start();  // 正常启动
$fiber->resume();  // 正常恢复（Fiber 结束）
// $fiber->resume();  // FiberError: Cannot resume a fiber that has finished

多个 Fiber 的交互 ​

php
<?php
declare(strict_types=1);

$fiberA = new Fiber(function (): void {
    echo "Fiber A: 步骤 1\n";
    Fiber::suspend();
    echo "Fiber A: 步骤 2\n";
});

$fiberB = new Fiber(function (): void {
    echo "Fiber B: 步骤 1\n";
    Fiber::suspend();
    echo "Fiber B: 步骤 2\n";
});

// 交叉执行
$fiberA->start();   // Fiber A: 步骤 1
$fiberB->start();   // Fiber B: 步骤 1
$fiberA->resume();  // Fiber A: 步骤 2
$fiberB->resume();  // Fiber B: 步骤 2

实战示例 ​

实战：生产者-消费者模式 ​

php
<?php
declare(strict_types=1);

class Channel
{
    /** @var \SplQueue<mixed> */
    private \SplQueue $queue;
    private ?\Fiber $waitingConsumer = null;
    private ?\Fiber $waitingProducer = null;

    public function __construct()
    {
        $this->queue = new \SplQueue();
    }

    public function send(mixed $value): void
    {
        if ($this->waitingConsumer !== null) {
            $consumer = $this->waitingConsumer;
            $this->waitingConsumer = null;
            $consumer->resume($value);
        } else {
            $this->queue->enqueue($value);
            Fiber::suspend();
        }
    }

    public function receive(): mixed
    {
        if (!$this->queue->isEmpty()) {
            return $this->queue->dequeue();
        }

        $this->waitingConsumer = \Fiber::getCurrent();
        return Fiber::suspend();
    }
}

$channel = new Channel();

// 消费者
$consumer = new Fiber(function () use ($channel): void {
    for ($i = 0; $i < 3; $i++) {
        $value = $channel->receive();
        echo "消费: {$value}\n";
    }
});

// 生产者
$producer = new Fiber(function () use ($channel): void {
    $items = ['apple', 'banana', 'cherry'];
    foreach ($items as $item) {
        echo "生产: {$item}\n";
        $channel->send($item);
    }
});

$consumer->start();
$producer->start();
$consumer->resume();
$producer->resume();
$consumer->resume();
$producer->resume();

实战：任务调度器 ​

php
<?php
declare(strict_types=1);

class SimpleScheduler
{
    /** @var \SplQueue<\Fiber> */
    private \SplQueue $fibers;

    public function __construct()
    {
        $this->fibers = new \SplQueue();
    }

    public function spawn(callable $callback): void
    {
        $this->fibers->enqueue(new \Fiber($callback));
    }

    public function run(): void
    {
        while (!$this->fibers->isEmpty()) {
            $fiber = $this->fibers->dequeue();

            if ($fiber->status() === \Fiber::STATUS_INIT) {
                $fiber->start();
            } elseif ($fiber->status() === \Fiber::STATUS_SUSPENDED) {
                $fiber->resume();
            }

            if ($fiber->status() === \Fiber::STATUS_SUSPENDED) {
                $this->fibers->enqueue($fiber);
            }
        }
    }
}

$scheduler = new SimpleScheduler();

$scheduler->spawn(function (): void {
    for ($i = 0; $i < 3; $i++) {
        echo "任务 A-{$i}\n";
        Fiber::suspend();
    }
});

$scheduler->spawn(function (): void {
    for ($i = 0; $i < 3; $i++) {
        echo "任务 B-{$i}\n";
        Fiber::suspend();
    }
});

$scheduler->run();

注意事项 ​

不能在非 Fiber 上下文中调用 suspend ​

php
<?php
declare(strict_types=1);

// Fiber::suspend() 只能在 Fiber 内部调用
// Fiber::suspend(); // FiberError: Cannot suspend outside of a fiber

Fiber 的异常传播 ​

如果在 Fiber 中抛出未捕获的异常，该异常会传播到调用 start()/resume() 的上下文中。

php
<?php
declare(strict_types=1);

$fiber = new Fiber(function (): void {
    Fiber::suspend();
    throw new \RuntimeException('Fiber 内部异常');
});

$fiber->start();

try {
    $fiber->resume(); // 异常传播到此处
} catch (\RuntimeException $e) {
    echo "捕获异常: {$e->getMessage()}\n";
}

最佳实践 ​

  1. 封装 Fiber 创建：将 Fiber 创建和启动封装为更高层的抽象，降低使用复杂度。
  2. 使用 Fiber::getCurrent()：在 Fiber 内部通过 Fiber::getCurrent() 获取当前 Fiber 实例。
  3. 异常处理：确保在 Fiber 内部或外部正确处理异常。
  4. 使用 try/finally：在 Fiber 回调中使用 try/finally 确保资源释放。
php
<?php
declare(strict_types=1);

function spawnFiber(callable $callback): \Fiber
{
    return new \Fiber(function () use ($callback): void {
        try {
            $callback();
        } catch (\Throwable $e) {
            echo "Fiber 异常: {$e->getMessage()}\n";
        }
    });
}

参考链接 ​