Paralleelsammud

Mõnikord saab mitu sammu käitada korraga. Meetod ->parallel() käivitab käitlejate rühma samaaegselt. Töövoog liigub edasi alles siis, kui kõik paralleelsed käitlejad on lõpetanud.

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

Selles näites:

1. FetchUserHandler käivitub esimesena (järjestikku)
2. Kolm käitlejat käivituvad samaaegselt (paralleelselt)
3. MergeAndSaveHandler käivitub pärast kõigi kolme lõpetamist (järjestikku)

Kuidas see töötab

Paralleelse sammuni jõudes käivitab Queuety rühma iga käitleja kohta ühe töö. Iga töö saab praeguse töövoo oleku ja töötab iseseisvalt. Töövoog jälgib, mitu paralleelset tööd on pooleli, ning jätkab järgmise sammuga alles siis, kui kõik on lõpetanud.

Iga paralleelse käitleja tagastusväärtus liidetakse töövoo olekusse. Kui mitu käitlejat tagastavad sama võtme, jääb kehtima viimati lõpetanud käitleja väärtus.

Olekute liitmine

Iga paralleelne käitleja tagastab andmed, mis liidetakse töövoo ühisesse olekusse:

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

Pärast paralleelse sammu lõppemist sisaldab olek nii social_profile kui ka payment_history väärtust ning järgmine järjestikune samm pääseb kõigele ligi.

Järjestikuste ja paralleelsete sammude kombineerimine

Võite vabalt kombineerida väljakutseid ->then() ja ->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 );

Vigade käsitlemine

Kui mõni paralleelrühma käsitleja ebaõnnestub, järgib töövoog selle käsitleja puhul tavapärast korduskatsete loogikat. Teisi juba lõpetanud paralleelseid käsitlejaid uuesti ei käivitata. Kui ebaõnnestunud käsitleja lõpuks õnnestub, jätkub töövoog. Kui korduskatsed saavad otsa ja töö mattakse, märgitakse töövoog ebaõnnestunuks ning ülejäänud haru tööd maetakse; kasutage retry_workflow() ebaõnnestunud sammu uuesti käivitamiseks.