Paralelní kroky

Někdy mohou kroky běžet současně. Metoda ->parallel() spustí skupinu obslužných funkcí souběžně. Workflow pokračuje až po dokončení všech paralelních obslužných funkcí.

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

V tomto příkladu:

1. Nejprve se spustí FetchUserHandler (sekvenčně)
2. Tři obslužné funkce běží souběžně (paralelně)
3. Po dokončení všech tří se spustí MergeAndSaveHandler (sekvenčně)

Jak to funguje

Když workflow dosáhne paralelního kroku, Queuety zařadí do fronty jednu úlohu pro každou obslužnou funkci ve skupině. Každá úloha dostane aktuální stav workflow a běží nezávisle. Workflow sleduje počet nedokončených paralelních úloh a pokračuje k dalšímu kroku až po jejich dokončení.

Návratová hodnota každé paralelní obslužné funkce se sloučí se stavem workflow. Pokud více obslužných funkcí vrátí stejný klíč, platí hodnota té, která skončí jako poslední.

Sloučení stavu

Každá paralelní obslužná funkce vrací data, která se sloučí do sdíleného stavu workflow:

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

Po dokončení paralelního kroku stav obsahuje hodnoty social_profile i payment_history a následující sekvenční krok k nim má přístup.

Kombinování sekvenčních a paralelních kroků

Volání ->then() a ->parallel() můžete libovolně kombinovat:

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

Zpracování chyb

Pokud selže některá obslužná funkce v paralelní skupině, workflow použije běžnou logiku opakování pro danou funkci. Ostatní paralelní obslužné funkce, které už skončily, se znovu nespouštějí. Jakmile neúspěšná funkce uspěje, workflow pokračuje. Pokud vyčerpá všechny opakované pokusy a je odložena, workflow se označí jako neúspěšné a zbývající úlohy větví se odloží. Pomocí retry_workflow() můžete neúspěšný krok spustit znovu.