Paralelné kroky

Niekedy možno spustiť viacero krokov súčasne. Metóda ->parallel() odosiela skupinu obslužných funkcií na súbežné vykonanie. Pracovný postup pokračuje až po dokončení všetkých paralelných obslužných funkcií.

Metóda ->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 príklade:

1. Najprv sa vykoná FetchUserHandler (postupne)
2. Tri obslužné funkcie sa vykonávajú súbežne (paralelne)
3. Po dokončení všetkých troch sa vykoná MergeAndSaveHandler (postupne)

Ako to funguje

Keď sa dosiahne paralelný krok, Queuety odošle jednu úlohu pre každú obslužnú funkciu v skupine. Každá úloha dostane aktuálny stav pracovného postupu a vykonáva sa nezávisle. Pracovný postup sleduje počet nedokončených paralelných úloh a pokračuje k ďalšiemu kroku až po dokončení všetkých.

Návratová hodnota každej paralelnej obslužnej funkcie sa zlúči so stavom pracovného postupu. Ak viacero obslužných funkcií vráti rovnaký kľúč, vyhrá tá, ktorá sa dokončí ako posledná.

Zlučovanie stavu

Každá paralelná obslužná funkcia vráti údaje, ktoré sa zlúčia so zdieľaným stavom pracovného postupu:

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 social_profile aj payment_history a nasledujúci postupný krok má k dispozícii všetky tieto údaje.

Kombinovanie postupných a paralelných krokov

Môžete ľubovoľne kombinovať volania ->then() a ->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 );

Spracovanie chýb

Ak zlyhá ktorákoľvek obslužná funkcia v paralelnej skupine, pracovný postup použije bežnú logiku opakovania pre danú funkciu. Ostatné paralelné obslužné funkcie, ktoré už skončili, sa nespúšťajú znova. Po úspešnom dokončení zlyhanej funkcie pracovný postup pokračuje. Ak sa vyčerpajú pokusy o opakovanie a úloha sa odloží, pracovný postup sa označí ako neúspešný a zostávajúce vetvové úlohy sa odložia; pomocou retry_workflow() spustite zlyhaný krok znova.