コンテンツへスキップ

並列ステップ

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 ] );

この例では、次の順に実行されます。

  1. 最初にFetchUserHandlerが実行される（逐次）
  2. 3つのハンドラーが並行して実行される
  3. 3つすべてが完了した後にMergeAndSaveHandlerが実行される（逐次）

仕組み

並列ステップに到達すると、Queuetyはグループ内の各ハンドラーに1つずつジョブをディスパッチします。各ジョブは現在のワークフロー状態を受け取り、独立して実行されます。ワークフローは未完了の並列ジョブ数を追跡し、すべて完了してから次のステップに進みます。

各並列ハンドラーの戻り値はワークフロー状態にマージされます。複数のハンドラーが同じキーを返す場合、最後に完了したハンドラーの値が使われます。

状態のマージ

各並列ハンドラーは、共有ワークフロー状態にマージされるデータを返します。

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()を使います。