PHP 纤程于生产：四例实情，其胜 curl_multi 与队列

書：解耦PHP——清淨六角之構，適應框架終老之應
亦余著： 思乎Go (二書之系) — Go程式之全覽 + Go中六角之構
吾之項目： 赫爾墨斯IDE |GitHub(代码托管平台)— 一款供开发者使用，与Claude Code及其他AI编程工具配套之集成开发环境
吾： xgabriel.com|Git(Hub)

十一月，PHP 8.1 运输纤维。RFC(请求评论)其小也，API有四法，而 Laravel之代码库多未触及之。启之。vendor/ 新启 Laravel 之项目，则见 Guzzle、Symfony HttpClient 及 ReactPHP 皆暗用之。尔之自码乎？恐无之。

此于 CRUD 之应用，尚可。一旦尔之 HTTP 服务广布五重，流送多吉字节之文件，或织 webhook 之调度，则队列或阻塞之别，渐显其费。Fibers 治一特定之困，而队列过解之。curl_multi未解之解。兹列四例，以证其功。

何谓纤维（六十秒心智模型）

纤丝者，协程也。非线程，非进程。一栈执行，可自暂停，后时复续，尽在相同之PHP进程中。

通篇之API：

$fiber = new Fiber(function (): void {
    echo "step 1\n";
    $resumed = Fiber::suspend("paused");  // hands control back
    echo "step 2 with: $resumed\n";
});

$out = $fiber->start();           // "step 1", returns "paused"
$fiber->resume("hello");          // "step 2 with: hello"
$fiber->isTerminated();           // true

Fiber::suspend()者，乃玄妙之术也。惟可自运行之Fiber中唤之。若于主流中呼之，则得Fiber\FiberError: Cannot call Fiber::suspend() outside of a Fiber。呼start()或resume()者，得回所悬之物。

行此二者，方可行远：

Fibers乃协契之物。无物可夺其志。若Fiber不呼suspend()，行至终而阻主流。
其流程相同。无全局解释器锁，无内存复制，无进程间通信。全局变量共享。PDO连接亦然，此乃多数人初遇之陷阱（详下）。

例一：并行HTTP分叉

汝之结账接口，呼支付网关，诈欺之务，忠诚之API，运费计算器，及仓廪可用之服务。五呼，每约四百毫秒，次第而行。此乃乐途上之二秒也。

curl_multi_exec()解此困，然此API乃二十年之轮询循环也。do {} while且select风姿翩翩，凝神观之。其效甚佳。然览之，未得欣悦。

此乃纤丝之版本也。其法：每HTTP调用，皆居于其独纤丝中。一微调度器，泵之。curl_multi而复其纤缕，若柄之终也。

final class ParallelHttp
{
    /** @param array<string, string> $urls keyed by label */
    public static function get(array $urls): array
    {
        $mh = curl_multi_init();
        $handles = [];
        $fibers = [];

        foreach ($urls as $key => $url) {
            $fibers[$key] = new Fiber(function () use ($url, $mh, &$handles, $key) {
                $ch = curl_init($url);
                curl_setopt($ch, CURLOPT_RETURNTRANSFER, true);
                curl_setopt($ch, CURLOPT_TIMEOUT, 5);
                curl_multi_add_handle($mh, $ch);
                $handles[$key] = $ch;

                // hand control back to the scheduler until curl says done
                Fiber::suspend();

                $body = curl_multi_getcontent($ch);
                curl_multi_remove_handle($mh, $ch);
                curl_close($ch);
                return $body;
            });
            $fibers[$key]->start();
        }

        // pump until every handle is finished
        do {
            curl_multi_exec($mh, $active);
            curl_multi_select($mh, 0.05);
        } while ($active > 0);

        curl_multi_close($mh);

        $results = [];
        foreach ($fibers as $key => $fiber) {
            $results[$key] = $fiber->resume();
        }
        return $results;
    }
}

$out = ParallelHttp::get([
    'payment'  => 'https://api.example.com/payment/check',
    'fraud'    => 'https://api.example.com/fraud/score',
    'loyalty'  => 'https://api.example.com/loyalty/balance',
    'shipping' => 'https://api.example.com/shipping/quote',
    'stock'    => 'https://api.example.com/warehouse/stock',
]);

一合成基准测试，对五端点施加人工400毫秒服务器端延迟，得序贯cURL耗时约2050毫秒；纤网+curl_multi版本毕，约四百四十毫秒。此乃所期之算。实际用时，不过最慢调用加之少许调度之耗。Guzzle异步文档报相似形状及ReactPHP之HTTP 客户端基准测试同列而坐。

纤维版非胜于原始curl_multi其下本同此网络之素。所异者，每光纤得常体之功能耳。无回调，无then()无承诺链。此乃成败处理非简时之要。试/捕之理，恰如所期。

第二案：反向压力之流处理

览十吉字节之CSV而fopen加fgetcsv工巧至矣，然批写于迟滞之库，则谬矣。朴拙之脚本，速若PHP之析，而下游之消者，或耗尽其列，或弃之如敝屣。

纤维使生产者可暂停，待消费者落后。此乃有界缓冲之模式：

final class BoundedChannel
{
    /** @var list<array<string, string>> */
    private array $buffer = [];
    private ?Fiber $waitingProducer = null;
    private ?Fiber $waitingConsumer = null;
    private bool $closed = false;

    public function __construct(private readonly int $capacity = 100) {}

    public function send(array $row): void
    {
        while (count($this->buffer) >= $this->capacity) {
            $this->waitingProducer = Fiber::getCurrent();
            Fiber::suspend(); // wait for the consumer to drain
        }
        $this->buffer[] = $row;
        if ($this->waitingConsumer !== null) {
            $c = $this->waitingConsumer;
            $this->waitingConsumer = null;
            $c->resume();
        }
    }

    public function receive(): ?array
    {
        while ($this->buffer === [] && !$this->closed) {
            $this->waitingConsumer = Fiber::getCurrent();
            Fiber::suspend();
        }
        if ($this->buffer === []) return null;
        $row = array_shift($this->buffer);
        if ($this->waitingProducer !== null) {
            $p = $this->waitingProducer;
            $this->waitingProducer = null;
            $p->resume();
        }
        return $row;
    }

    public function close(): void { $this->closed = true; }
}

一真实之吞咽管道，以通道相接，织生产之纤与消费之纤：

$channel = new BoundedChannel(capacity: 500);

$producer = new Fiber(function () use ($channel): void {
    $fp = fopen('orders-2026.csv', 'r');
    $header = fgetcsv($fp);
    while (($row = fgetcsv($fp)) !== false) {
        $channel->send(array_combine($header, $row));
    }
    fclose($fp);
    $channel->close();
});

$consumer = new Fiber(function () use ($channel): void {
    $batch = [];
    while (($row = $channel->receive()) !== null) {
        $batch[] = $row;
        if (count($batch) === 50) {
            OrderImporter::insertBatch($batch); // blocking DB write
            $batch = [];
        }
    }
    if ($batch !== []) OrderImporter::insertBatch($batch);
});

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

缓冲区满五百行时，生产者自停。消费者清空，生产者复行。内存恒定。此模式之九吉字节CSV，运行间驻留内存恒在百兆以下，盖因缓冲量限五百行。有趣者非速也（数据库为瓶颈），乃脚本无论何速，终不致耗尽内存。fgetcsv阅之。

一物之生，使人懈怠，然物之生，不能自内助之，则遍染其链。纤维则不然，可自内助之，此实为便捷之胜。

案三：易生成器协程

纤维未兴，世人于PHP为异步事，常作此：

function fetchUser(int $id): Generator {
    $raw = yield $http->get("/users/$id");        // a Promise
    $orders = yield $http->get("/orders?u=$id");
    return ['user' => $raw, 'orders' => $orders];
}

汝需一"协程运行者"，循生成器而行，待每所予之承，复推其果回。此法可行（Recoil、Amp v2重用之， Laravel Octane亦有所用），然其式: Generator恒误。此函数返array类型系统，未尝合乎实情。

纤维解此。同此理，真返类型：

public function fetchUser(int $id): UserBundle
{
    $user = $this->client->get("/users/$id");      // suspends internally
    $orders = $this->client->get("/orders?u=$id");
    return new UserBundle($user, $orders);
}

客待网隙，暂止纤线。此法若直书同步之码。PHPStan、Psalm、IDE补全：皆得正型。Amp团队于v3中，周纤线而重撰诸事（发布记），正为此故。ReactPHP已发。react/async 4.x 亦然：await() 乃基于纤维之原语，使应许返回之库若作同步观。

若汝有2019年基于Generator之"异步"助手，拔除之换纤维，常削文件三成至四成，纯为虚仪。行无变改。

案四：于ReactPHP / AMPHP之中，不更全代码基

不须尽力于事件循环，亦可使用Fibers。此乃多数"是否应使用Amp"之问所忽略之点也。

一Symfony Messenger处理器，需分派至三处Webhooks，可于本地引入一感知Fiber之HTTP客户端，以行微细之事await于处理者之中，否则保持原味。

use Amp\Http\Client\HttpClientBuilder;
use Amp\Http\Client\Request;
use function Amp\async;
use function Amp\Future\awaitAll;

final class DispatchWebhooksHandler
{
    public function __invoke(OrderPlaced $event): void
    {
        $client = HttpClientBuilder::buildDefault();
        $urls = WebhookSubscriptions::for($event->orderId);

        $futures = array_map(
            fn(string $url) => async(fn() =>
                $client->request(new Request($url, 'POST', json_encode($event)))
            ),
            $urls,
        );

        [$errors, $responses] = awaitAll($futures);
        foreach ($errors as $url => $err) {
            FailedWebhook::record($url, $err->getMessage());
        }
    }
}

async()生一纟纟丝。awaitAll()阻其处理器，俟诸纤丝毕功，然唯此处理器而已。其余 Symfony 之务，照常进行。无全局事件循环。勿言“将整框架转为异步。”

其要在于 Amp'sEventLoop初用而懒初始化，无待而净止。同此。react/async尔得协程工整之法，而未染余码之弊。

纤维无益处处

PDO阻塞。插入PDO::query()纤维内非为异步。纤维阻尽进程，待 libpq 返归。同此。mysqli以默认模式。同此。file_get_contents()于本地文件。同此。sleep().

之工作，如图像缩放、JSON编码二百兆之结构、bcrypt，于纤线无所益。纤线者，为I/O之待也。若瓶颈乃紧for之循环，以事 hashing，则需 pthreads、并行，或pcntl_fork，非 coroutines也。

心验之：此操作于Node.js，当自Promise而返乎？libuv之助而呼耶？若然，则基于纤维之裹缚可助。否（汝之窒碍在PHP本身），纤维则无用矣。

其大注曰：有异步之PDO替者（amphp/postgres，amphp/mysql）。者，与纤线协作者也。其存在，当知之。其互换，鲜有免费者。雅言与教义，皆假阻塞驱动为前提。

错误处理之妙谛

纤线内之例外，几若所期。几若。

$fiber = new Fiber(function (): void {
    Fiber::suspend();
    throw new RuntimeException("boom");
});

$fiber->start();
try {
    $fiber->resume(); // exception propagates HERE, not where it was thrown
} catch (RuntimeException $e) {
    // caught
}

例外行至所召之者resume()固常无碍。所陷者，Fiber::throw()：

$fiber = new Fiber(function (): void {
    try {
        Fiber::suspend();
    } catch (DomainException $e) {
        // the throw() call lands inside the suspend
    }
});
$fiber->start();
$fiber->throw(new DomainException("cancelled"));

throw()于suspend()处注入异常。若Fiber不摄之，则异常上涌至调用者。throw()，世人用以取消，然设一例外，越三纤而失其踪，莫知所归。

要诀：纤体宜短。若纤之入口需更精微之错处处理，非止于试/捕，则当返一Result式之象，而非抛之。suspend()

此意于汝代码之基

纤维乃本源，非可外铄之器。当求之境：并行I/O之散，知压之流，易生成器异步之助，及嵌于阻隔之框架内。不当求之境：CPU之劳，PDO，无明悬停之故处。

尔若已运 Laravel 或 Symfony 之应用，其最低风险之始，乃第四案：一独应之器，借 Amp 或 react/async，生数纤维于本处，即出。无全局事件之环。无基建之变。度其胜，乃扩之。

尔今应用，其最缓之散布何在？第四案，无改框架，可应之乎？

若此有益

纤维者，运行时之素也。所重者，架构之问耳：此散逸之势，当存于域层耶，抑或存于适配器耶？此乃众代码之纠葛所由生也。PHP之解耦者，框架之常不足，代码基所求之架构层也；其述如何使异步I/O于边缘，而域码安然同步.

有Kindle、Paperback、Hardcover版。英文、德文、日文版今已出，葡萄牙文及西班牙文版将即至。

推薦訂閱源

DEV Community