Vzporedni koraki

Včasih lahko več korakov poteka hkrati. Metoda ->parallel() odpremi skupino obdelovalnikov za sočasno izvajanje. Delovni tok napreduje šele, ko se dokončajo vsi vzporedni obdelovalniki.

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 tem primeru:

1. Najprej se izvede FetchUserHandler (zaporedno)
2. Trije obdelovalniki se izvedejo sočasno (vzporedno)
3. MergeAndSaveHandler se izvede, ko se vsi trije dokončajo (zaporedno)

Kako deluje

Ko je na vrsti vzporedni korak, Queuety odpremi po eno opravilo za vsak obdelovalnik v skupini. Vsako opravilo prejme trenutno stanje delovnega toka in se izvede neodvisno. Delovni tok spremlja število čakajočih vzporednih opravil in nadaljuje z naslednjim korakom šele, ko se vsa dokončajo.

Vrnjena vrednost vsakega vzporednega obdelovalnika se združi s stanjem delovnega toka. Če več obdelovalnikov vrne isti ključ, obvelja tisti, ki se dokonča zadnji.

Združevanje stanja

Vsak vzporedni obdelovalnik vrne podatke, ki se združijo v skupno stanje delovnega toka:

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čanju vzporednega koraka stanje vsebuje social_profile in payment_history, naslednji zaporedni korak pa lahko dostopa do vsega.

Kombiniranje zaporednih in vzporednih korakov

Klice ->then() in ->parallel() lahko poljubno kombinirate:

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

Obravnava napak

Če kateri koli obravnavalnik v vzporedni skupini odpove, potek dela za ta obravnavalnik uporabi običajno logiko ponovnih poskusov. Drugi vzporedni obravnavalniki, ki so že zaključeni, se ne zaženejo znova. Ko neuspešni obravnavalnik uspe, se potek dela nadaljuje. Če porabi vse ponovne poskuse in je prestavljen v čakalno vrsto neuspelih opravil, je potek dela označen kot neuspešen, preostala opravila vej pa so prestavljena v čakalno vrsto neuspelih opravil; uporabite retry_workflow(), da znova zaženete neuspešni korak.