并行步骤

以 Markdown 格式查看在 GitHub 上编辑文档位置 2cc7b2b

有时多个步骤可以同时运行。->parallel() 方法会并发调度一组处理器。只有所有并行处理器都完成后，工作流才会继续推进。

`->parallel()` 方法

PHP

use Queuety\Queuety;

Queuety::workflow( 'enrich_profile' )
    ->then( FetchUserHandler::class )
    ->parallel( [
        FetchSocialProfileHandler::class,
        FetchPaymentHistoryHandler::class,
        FetchActivityLogHandler::class,
    ] )
    ->then( MergeAndSaveHandler::class )
    ->dispatch( [ 'user_id' => 42 ] );

此示例中：

FetchUserHandler 首先运行（顺序执行）
三个处理器并发运行（并行执行）
MergeAndSaveHandler 在三者全部完成后运行（顺序执行）

工作原理

到达并行步骤时，Queuety 会为组中的每个处理器调度一个作业。每个作业都会收到当前工作流状态并独立运行。工作流会跟踪尚未完成的并行作业数，只有全部完成后才会推进到下一步骤。

每个并行处理器的返回值都会合并到工作流状态中。如果多个处理器返回相同的键，最后完成的处理器会覆盖该键。

状态合并

每个并行处理器都会返回数据，并将其合并到共享工作流状态中：

PHP

class FetchSocialProfileHandler implements Step {
    public function handle( array $state ): array {
        $profile = fetch_social_api( $state['user_id'] );
        return [ 'social_profile' => $profile ];
    }

    public function config(): array {
        return [];
    }
}

class FetchPaymentHistoryHandler implements Step {
    public function handle( array $state ): array {
        $payments = fetch_payment_api( $state['user_id'] );
        return [ 'payment_history' => $payments ];
    }

    public function config(): array {
        return [];
    }
}

并行步骤完成后，状态中会同时包含 social_profile 和 payment_history，后续顺序步骤可以访问所有这些数据。

混合使用顺序和并行步骤

你可以自由混合调用 ->then() 和 ->parallel()：

PHP

Queuety::workflow( 'complex_pipeline' )
    ->then( SetupHandler::class )               // sequential
    ->parallel( [ TaskA::class, TaskB::class ] ) // parallel
    ->then( IntermediateHandler::class )         // sequential
    ->parallel( [ TaskC::class, TaskD::class ] ) // parallel
    ->then( FinalizeHandler::class )             // sequential
    ->dispatch( $payload );

错误处理

如果并行组中的任何处理器失败，工作流会按照该处理器的常规重试逻辑执行。其他已完成的并行处理器不会重新运行。失败的处理器成功后，工作流会继续执行。如果重试次数耗尽后任务被搁置，工作流将标记为失败，其余分支任务也会被搁置；使用 retry_workflow() 再次运行失败的步骤。