异步 Promise

OxPHP 在后台线程中运行 PHP 闭包，使用一个独立于 HTTP 工作进程池的专用线程池。耗时的工作会被放到那里执行，而不是阻塞请求处理。

工作原理

派发（Dispatch） — 调用 oxphp_async()，传入一个闭包以及可选的参数。OxPHP 会序列化闭包的 use 变量和参数，将它们发送到异步池，并立即返回一个 promise ID
执行（Execute） — 一个专用的异步工作线程反序列化这些数据，运行闭包，并序列化结果
等待（Await） — 调用 oxphp_async_await()，传入 promise ID。在启用纤程的工作进程模式下，当前纤程会挂起，同一线程上的其他请求可以继续处理。在传统模式下，工作线程会阻塞，直到结果就绪
清理（Cleanup） — 任何未被显式等待的 promise 都会在请求结束时被自动取消并清理

配置

Variable	Default	Description
`ASYNC_WORKERS`	`0` (disabled)	专用异步工作线程的数量。设为 `0` 可完全禁用异步池
`ASYNC_QUEUE_CAPACITY`	`0` (auto)	待处理异步任务的最大数量。为 `0` 时，默认取 `ASYNC_WORKERS × 64`
`ASYNC_MAX_FIBERS`	`256`	每个工作进程上并发异步任务纤程的上限。进程全局的在途上限（排队 + 运行中的任务）为 `ASYNC_MAX_FIBERS × ASYNC_WORKERS`；超过该上限的派发会立即被拒绝（非阻塞），抛出 `OxPHP\Async\AsyncException`，因此扇出组合不会因等待其自身占用的容量而死锁

Note

异步池默认是禁用的（ASYNC_WORKERS=0）。在池被禁用时，四个异步函数依然存在，但被调用时会抛出 OxPHP\Async\AsyncException。将 ASYNC_WORKERS 设为大于 0 的值即可启用后台执行。

派发任务

向 oxphp_async() 传入一个闭包以及可选的参数。它会立即返回一个 promise ID（整数）：

php

<?php
$promise = oxphp_async(function (string $url) {
    return file_get_contents($url);
}, 'https://api.example.com/data');

// The closure is running in the background.
// Do other work here...

$result = oxphp_async_await($promise);
echo $result;

向闭包传递数据

使用 use 变量或函数参数来传递数据。仅支持标量类型和数组：

php

<?php
$apiKey = 'sk-abc123';
$ids = [1, 2, 3];

$promise = oxphp_async(function () use ($apiKey, $ids) {
    // $apiKey and $ids are available here
    return count($ids);
});

等待结果

单个 promise

php

<?php
$result = oxphp_async_await($promise);           // Wait indefinitely
$result = oxphp_async_await($promise, 5.0);      // Wait up to 5 seconds

超时值为 0.0（默认值）时会无限期等待。超时时会抛出 OxPHP\Async\TimeoutException。

所有 promise

oxphp_async_await_all() 会等待每一个 promise，并返回一个以 promise ID 为键的关联数组：

php

<?php
$p1 = oxphp_async(fn() => file_get_contents('https://api.example.com/users'));
$p2 = oxphp_async(fn() => file_get_contents('https://api.example.com/orders'));

$results = oxphp_async_await_all([$p1, $p2], 10.0);

$users  = $results[$p1];
$orders = $results[$p2];

Note

oxphp_async_await_all() 按数组顺序依次等待各个 promise。所有闭包在异步池上并发运行，但调用线程一次收集一个结果。

第一个落定的 promise（race）

oxphp_async_await_race() 会在有一个 promise 落定时立即返回，无论它是被兑现还是被拒绝：

php

<?php
$p1 = oxphp_async(fn() => fetch_from_primary_db());
$p2 = oxphp_async(fn() => fetch_from_replica_db());

$winner = oxphp_async_await_race([$p1, $p2], 5.0);
// $winner = ['id' => int, 'value' => mixed]

echo "Promise {$winner['id']} won: {$winner['value']}";

在 oxphp_async_await_race() 返回之后，未胜出的 promise 仍可被单独等待。如果胜出的 promise 被拒绝，则会抛出 OxPHP\Async\AsyncException —— 落败者仍可被等待。这相当于 JavaScript 中的 Promise.race。

第一个被兑现的 promise

oxphp_async_await_any() 会在有一个 promise 被兑现时立即返回。拒绝会被累积起来，只有当每个 promise 都被拒绝时才会变为可观察。这相当于 JavaScript 中的 Promise.any —— 适用于回退 / 冗余模式（“任意一个能响应的镜像”）。

php

<?php
$mirror_a = oxphp_async(fn() => fetch('https://mirror-a.example.com/data'));
$mirror_b = oxphp_async(fn() => fetch('https://mirror-b.example.com/data'));
$mirror_c = oxphp_async(fn() => fetch('https://mirror-c.example.com/data'));

try {
    $winner = oxphp_async_await_any([$mirror_a, $mirror_b, $mirror_c], 5.0);
    // ['id' => one of the input ids, 'value' => its result]
} catch (\OxPHP\Async\AggregateAsyncException $e) {
    foreach ($e->getErrors() as $i => $err) {
        // $err keyed by input position 0..N-1
    }
    foreach ($e->getErrorMap() as $promise_id => $err) {
        // $err keyed by promise id
    }
} catch (\OxPHP\Async\TimeoutException $e) {
    foreach ($e->getPartialErrors() as $promise_id => $err) {
        // promises that already rejected before the deadline
    }
    $cancelled = $e->getCancelledPromiseIds();
    // Promise ids that had not settled at the deadline. The cancel flag
    // is set on each AND their receivers were dropped — passing any of
    // these ids to oxphp_async_await*() afterwards throws "unknown or
    // already-awaited promise id". Treat the list as an audit trail, not
    // a resumable queue.
}

在胜出那一刻仍处于等待状态的未胜出 promise 仍可被单独等待。而在胜出者之前就已被拒绝的 promise 则不能 —— 它们的结果在被累积为候选错误时就已被消费。

异常类型

Class	Thrown by	Notes
`OxPHP\Async\AsyncException`	`oxphp_async_await()`, `oxphp_async_await_all()`, `oxphp_async_await_race()`	单个错误，包含消息以及可选的原始异常详情。
`OxPHP\Async\TimeoutException`	四个 await-* 在到达超时时限时	继承自 `AsyncException`。对于 `oxphp_async_await_any()` 的超时，会填充 `getPartialErrors()` 和 `getCancelledPromiseIds()`；对于其他调用点，两者均返回 `[]`。
`OxPHP\Async\AggregateAsyncException`	`oxphp_async_await_any()` 在每个 promise 都被拒绝时	继承自 `AsyncException`。提供 `getErrors()`（按位置排列，键为 0..N-1）、`getErrorMap()`（以 id 为键）、`getPromiseIds()`。

错误处理

在异步闭包内部抛出的异常会被捕获，并在等待时以 OxPHP\Async\AsyncException 的形式重新抛出：

php

<?php
$promise = oxphp_async(function () {
    throw new \RuntimeException('Something failed');
});

try {
    $result = oxphp_async_await($promise);
} catch (\OxPHP\Async\AsyncException $e) {
    // "Async task failed: [RuntimeException] Something failed"
    echo $e->getMessage();
}

异步闭包内部的 exit() 和 die() 也会被捕获并转换为 OxPHP\Async\AsyncException。异步工作进程会存活下来并继续处理新的任务。

异常层级

text

\Exception
  └── OxPHP\Async\AsyncException                   # All async errors
        ├── OxPHP\Async\TimeoutException           # Timeout-specific
        └── OxPHP\Async\AggregateAsyncException    # Multiple failures (await_all / await_any)

纤程集成

在工作进程模式下，oxphp_async_await() 会与 OxPHP 的纤程调度器协作。它不会阻塞工作线程，而是让当前纤程在等待结果期间挂起。一旦结果就绪，调度器会将其恢复，因此其他请求可以在同一线程上继续推进。

在传统模式下（没有 worker 文件），oxphp_async_await() 会同步阻塞工作线程。工作进程在等待期间无法处理其他请求。

为获得最佳性能，请将异步 promise 与工作进程模式结合使用：

worker.php

<?php
// worker.php
require __DIR__ . '/../vendor/autoload.php';

oxphp_worker(function () {
    // These two API calls run concurrently on the async pool
    // while the fiber suspends — the worker thread is free for other requests
    $p1 = oxphp_async(fn() => file_get_contents('https://api.example.com/users'));
    $p2 = oxphp_async(fn() => file_get_contents('https://api.example.com/orders'));

    $results = oxphp_async_await_all([$p1, $p2]);
    echo json_encode($results);
});

组合（嵌套异步）

一个异步任务自身也可以调用 oxphp_async() 并等待其结果。由于每个任务都在一个调度器纤程内运行，等待一个嵌套的 promise 会挂起该任务的纤程并释放其工作进程去运行嵌套任务 —— 因此一个任务可以扇出到子任务，而不会在等待期间一直占用某个工作进程。

php

<?php
$p = oxphp_async(function (): int {
    // Dispatched and awaited from inside an async task
    $inner = oxphp_async(fn () => 21);
    return oxphp_async_await($inner) * 2;
});

$result = oxphp_async_await($p); // 42

oxphp_async_await_all()、oxphp_async_await_race() 和 oxphp_async_await_any() 同样可以在任务纤程内部调用。并发任务纤程的数量（排队 + 运行中）受 ASYNC_MAX_FIBERS × ASYNC_WORKERS 限制；一个会超出上限的派发会立即被拒绝并抛出 OxPHP\Async\AsyncException，而不是阻塞，因此扇出不会因等待其自身占用的容量而死锁。

限制

异步闭包在独立的线程上运行。这对哪些数据可以跨越线程边界施加了限制：

Allowed	Not allowed
`null`、`bool`、`int`、`float`、`string`	普通对象（任何未实现 `OxPHP\Shared\Shareable` 的类）
标量类型的数组	资源（文件句柄、数据库连接、流）
嵌套的标量数组	`use` 捕获了非 Shareable 对象的闭包
`Shared\*` 实例（`Counter`、`Map`、`Channel`、`Atomic`、`Flag`、`Mutex`、`Once`、`Pool`、`Registry`）以及其他实现了 `OxPHP\Shared\Shareable` 的类

其他约束：

仅限用户函数 —— 闭包必须是用户自定义的，而不能是对内置函数的封装
序列化开销 —— 参数和返回值在跨越线程边界时会被序列化。较大的数组或字符串会增加延迟
普通 PHP 值没有共享状态 —— 每个异步工作进程都有自己的 PHP 环境。普通变量、数组和类实例在跨越边界时会被复制（或被拒绝）。请使用共享状态原语（Shared\Counter、Shared\Map、Shared\Channel……）来传递对两个线程都可见的引用

Docker 示例

compose.yaml

services:
  app:
    image: ghcr.io/oxphp/oxphp:0.10.0
    ports:
      - "80:80"
    environment:
      - DOCUMENT_ROOT=/var/www/html/public
      - WORKER_MODE_ENABLED=true
      - ENTRY_FILE=worker.php
      - ASYNC_WORKERS=4
      - ASYNC_QUEUE_CAPACITY=256

故障排查

"Async pool is disabled. Set ASYNC_WORKERS > 0 to enable."

异步池尚未配置。当 ASYNC_WORKERS=0（默认值）时，异步函数虽已注册，但每次调用都会抛出 OxPHP\Async\AsyncException。

修复： 将 ASYNC_WORKERS 设为一个正值：

bash

ASYNC_WORKERS=4

"Failed to dispatch async task (pool full)"

异步池正在运行，但所有队列槽位都已被占满，或者进程全局的在途上限（ASYNC_MAX_FIBERS × ASYNC_WORKERS，涵盖排队 + 运行中的任务）已达到。两者都会在派发时抛出 OxPHP\Async\AsyncException 并使 oxphp_async_tasks_rejected_total 递增。

检查： 确认池正在接受任务：

bash

curl -s http://localhost:9090/config | jq '.async_workers'

修复： 提高 ASYNC_WORKERS 或 ASYNC_QUEUE_CAPACITY。

"Cannot pass object values in use-vars to async closure"

对象无法跨越线程边界进行序列化。

修复： 在派发之前先提取你需要的标量数据：

php

<?php
// Wrong: passing an object
$promise = oxphp_async(function () use ($user) { ... });

// Correct: passing scalar data extracted from the object
$userId = $user->getId();
$userName = $user->getName();
$promise = oxphp_async(function () use ($userId, $userName) { ... });

等待在传统模式下挂起

在传统模式下，oxphp_async_await() 会阻塞工作线程。如果所有 PHP 工作进程都被阻塞在等待异步结果上，服务器就会停止处理请求。

修复： 启用工作进程模式（WORKER_MODE_ENABLED=true），使 oxphp_async_await() 挂起纤程而不是阻塞线程。

超时会取消被放弃的任务

OxPHP\Async\TimeoutException 会在到达超时时限的那一刻在等待端被抛出。后台任务不再被放任于无人观察的运行状态：一个停在 oxphp_sleep() 中或挂起等待子 promise 的任务会被恢复并展开（其 finally 块会运行），而一个从不让出的 CPU 密集型任务会在一个操作码边界处被中断 —— 因此取消是尽力而为的，具有一个较短的延迟上界，而非即时。同样的取消也适用于 oxphp_async_await_all() 放弃的那些 promise，以及 oxphp_async_await_race() / oxphp_async_await_any() 的落败者。一个仍无法及时被中断的任务会被搁浅（stranded） —— 被取消但在请求结束时排空，这可能会使 RSHUTDOWN 延长最多几秒；请关注 oxphp_async_tasks_stranded_total。

最佳实践

务必设置超时 —— 在生产环境中为 oxphp_async_await() 调用设置超时，以防止无限期等待
使用工作进程模式 —— 以获得基于纤程的非阻塞等待，而不是阻塞工作线程
保持闭包精简 —— 派发聚焦的工作单元，而不是整个请求处理器
在派发前提取标量 —— 在传给闭包之前，先从对象中取出 ID、字符串和配置值
监控异步池 —— 在 Prometheus 指标中检查 oxphp_async_tasks_rejected_total。如果拒绝数在上升，就提高 ASYNC_WORKERS 或 ASYNC_QUEUE_CAPACITY

参见

工作进程模式 —— 具有基于纤程并发能力的持久 PHP 进程
PHP 函数 —— oxphp_async()、oxphp_async_await() 及相关函数参考
指标 —— 异步池 Prometheus 指标
配置参考 —— ASYNC_WORKERS 和 ASYNC_QUEUE_CAPACITY