Parallelle trin

Nogle gange kan flere trin køre samtidig. Metoden ->parallel() sender en gruppe behandlere afsted samtidigt. Workflowet går først videre, når alle parallelle behandlere er fuldført.

Metoden ->parallel()

use Queuety\Queuety;

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

I dette eksempel:

1. FetchUserHandler køres først (sekventielt)
2. Tre behandlere køres samtidig (parallelt)
3. MergeAndSaveHandler køres, når alle tre er fuldført (sekventielt)

Sådan fungerer det

Når et parallelt trin nås, sender Queuety ét job afsted pr. behandler i gruppen. Hvert job modtager den aktuelle workflowtilstand og kører uafhængigt. Workflowet holder styr på antallet af udestående parallelle jobs og går først videre til næste trin, når alle er fuldført.

Hver parallel behandlers returværdi flettes ind i workflowtilstanden. Hvis flere behandlere returnerer samme nøgle, vinder den, der fuldføres sidst.

Sammenfletning af tilstand

Hver parallel behandler returnerer data, der flettes ind i den fælles workflowtilstand:

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 [];
    }
}

Når det parallelle trin er fuldført, indeholder tilstanden både social_profile og payment_history, og det næste sekventielle trin kan tilgå det hele.

Kombination af sekventiel og parallel kørsel

Du kan frit kombinere kald til ->then() og ->parallel():

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

Fejlhåndtering

Hvis en behandler i en parallel gruppe fejler, følger workflowet den normale logik for nye forsøg for den behandler. Andre parallelle behandlere, der allerede er fuldført, køres ikke igen. Når den fejlede behandler lykkes, fortsætter workflowet. Hvis alle forsøg opbruges, og jobbet begraves, markeres workflowet som fejlet, og de resterende forgreningsjobs begraves. Brug retry_workflow() til at køre det fejlede trin igen.