PHPAPI任务编排怎么实现

本文目录导读：

1. 简单串行/并行编排（纯 PHP + curl/Guzzle）
2. 有状态任务编排（State Machine / Workflow Engine）
3. 声明式编排（Pipeline / Middleware Pattern）
4. 事件驱动编排（Message Queue + Workers）
5. 重型框架：symfony/workflow + messenger
6. 总结：如何选择？

PHP 中的 API 任务编排（Task Orchestration）通常指将多个异步或同步的 API 调用按照业务逻辑组合、调度、并发执行,最终聚合结果的过程。

这通常用在复杂业务聚合（如 BFF 层）、工作流引擎（Workflow）或数据流水线（Pipeline）中。

下面我为你梳理几种主流的实现方式,从简单到复杂。

---

简单串行/并行编排（纯 PHP + curl/Guzzle）

这是最基础的方式，适合 API 数量不多、逻辑固定的场景。

串行执行（Dependency Chain）

// 步骤1: 获取用户
$user = apiCall('/user/info', ['user_id' => 123]);
// 步骤2: 依赖步骤1的结果获取订单
$orders = apiCall('/user/orders', ['user_id' => $user['id']]);
// 步骤3: 聚合结果
$result = [
    'user'   => $user,
    'orders' => $orders,
];

并行执行（Fan-out/Fan-in）

需要小心并发和超时控制，PHP 缺少原生的协程，但可以利用 curl_multi_* 或 Guzzle 的 Pool。

Guzzle 6/7 + Pool 示例：

use GuzzleHttp\Client;
use GuzzleHttp\Promise;
use GuzzleHttp\Pool;
$client = new Client(['base_uri' => 'https://api.example.com/']);
$requests = [
    'user'   => $client->getAsync('/user/info'),
    'orders' => $client->getAsync('/user/orders'),
    'meta'   => $client->getAsync('/meta/config'),
];
// 等待所有请求完成
$results = Promise\Utils::settle($requests)->wait();
// 处理聚合结果
foreach ($results as $key => $result) {
    if ($result['state'] === 'fulfilled') {
        $data[$key] = json_decode($result['value']->getBody(), true);
    } else {
        // 处理错误
        $data[$key] = ['error' => $reason];
    }
}

---

有状态任务编排（State Machine / Workflow Engine）

当编排逻辑复杂（有分支、重试、补偿、人工审核）时，需要有状态的支持，PHP 中可以使用以下几种思路。

1 轻量级：spatie/async + 自定义状态表

spatie/async 提供了 Promise 风格的并发支持,适合中等复杂度的流水线。

use Spatie\Async\Pool;
$pool = Pool::create();
$pool->add(function () {
    // API 调用 1
    return apiCall('/step1');
})->then(function ($output) {
    // 根据 step1 的结果决定是否调用 step2
    if ($output['status'] === 'success') {
        return apiCall('/step2', ['data' => $output]);
    }
    return null;
})->catch(function ($e) {
    // 错误处理 / 补偿逻辑
    logError($e);
    return ['error' => $e->getMessage()];
});
$pool->wait();

2 企业级：temporalio/sdk-php

Temporal 是目前最强大的开源分布式工作流引擎，它有 PHP SDK，非常适合长运行、高可靠性的 API 编排。

核心概念：

• Workflow：整个编排逻辑（一个 PHP 类，用 #[WorkflowInterface] 标记）。
• Activity：单个 API 调用（用 #[ActivityInterface] 标记）。
• 确定性：Workflow 代码必须是确定性的（不能有随机数、外部 IO 等）。

示例：

// 定义 Activity
#[ActivityInterface('payment')]
class PaymentActivity
{
    public function charge(string $orderId, float $amount): string
    {
        // 调用支付宝/微信支付 API
        return $response['transactionId'];
    }
    public function refund(string $transactionId): bool
    {
        // 调用退款 API
        return true;
    }
}
// 定义 Workflow
#[WorkflowInterface('order_fulfillment')]
class OrderFulfillmentWorkflow
{
    #[WorkflowMethod]
    public function run(string $orderId)
    {
        // Step 1: 调用支付 Activity
        $txId = yield Workflow::executeActivity('payment.charge', $orderId);
        // Step 2: 异步通知仓库（并行）
        $promise = Workflow::executeActivity('warehouse.prepare', $orderId);
        // Step 3: 等待 1 分钟后再检查状态（内置重试）
        yield Workflow::timer(60);
        $status = yield $promise;
        if ($status === 'failed') {
            // 补偿：退款
            yield Workflow::executeActivity('payment.refund', $txId);
            throw new \Exception('Fulfillment failed');
        }
        return ['txId' => $txId, 'shipping' => $status];
    }
}

优点： 内置重试、超时、补偿（Saga）、定时器、可观测性。 缺点： 部署运维复杂（需要运行 Temporal Server）。

---

声明式编排（Pipeline / Middleware Pattern）

适合固定步骤、顺序执行的 API 流水线,常用中间件模式实现。

PHP League Pipeline 示例：

use League\Pipeline\PipelineBuilder;
use League\Pipeline\StageInterface;
// Stage 1: 调用验证 API
class ValidateStage implements StageInterface
{
    public function __invoke($payload)
    {
        $payload['validated'] = apiCall('/validate', $payload);
        return $payload;
    }
}
// Stage 2: 调用核心处理 API
class ProcessStage implements StageInterface
{
    public function __invoke($payload)
    {
        if (!$payload['validated']['ok']) {
            throw new \Exception('Validation failed');
        }
        $payload['result'] = apiCall('/process', $payload);
        return $payload;
    }
}
// 构建并执行
$pipeline = (new PipelineBuilder())
            ->add(new ValidateStage())
            ->add(new ProcessStage())
            ->build();
$output = $pipeline($input);

---

事件驱动编排（Message Queue + Workers）

适合异步、松耦合、高吞吐的场景,编排逻辑分散在多个消费者中。

架构：

1. API 入口：接收请求，发送到 RabbitMQ/Kafka。
2. Worker 1：消费订单创建消息，调用支付 API，成功后发送 payment_success 事件。
3. Worker 2：消费 payment_success 事件，调用库存 API，成功后发送 inventory_allocated 事件。
4. Worker 3：消费 inventory_allocated,聚合结果并返回。

PHP 实现：

• 工具： php-amqplib、laminas-pubsub、roadrunner。
• 优势： 天然异步、可扩展、容错。
• 劣势： 状态跟踪困难，需要额外的事件溯源（Event Sourcing）来保障一致性。

---

重型框架：symfony/workflow + messenger

Symfony 的 Workflow 组件可以定义有限状态机，配合 Messenger 组件实现异步中间件。

核心流程：

1. 定义状态机：pending -> payment -> shipping -> done。
2. 使用中间件：每个状态转换时，调用一个 Message 处理 API 调用。
3. 使用 Saga：如果某个状态失败,发送补偿消息。

适用场景： 已有 Symfony 技术栈,希望减少外部依赖。

---

如何选择？

场景	推荐方案	原因
少量 API，串行/并行	Guzzle Pool + Promise	最简单，零依赖。
中等复杂度，有重试	spatie/async + 自定义状态表	轻量，可处理分支。
长运行、分布式、补偿	Temporal PHP SDK	专业、可靠。
严格流水线，固定顺序	League Pipeline	逻辑清晰，测试友好。
高吞吐、事件驱动	Message Queue + Workers	松耦合，可扩展。
已有 Symfony 项目	symfony/workflow + messenger	集成度高。

新手建议路线： 从 Guzzle 的并发池开始 -> 引入 spatie/async 处理分支 -> 如果业务复杂到需要状态持久化，再考虑 Temporal 或商业编排引擎（如 Camunda）。