Paralēlās darbības

Dažkārt vairākas darbības var izpildīt vienlaikus. Metode ->parallel() nosūta apstrādātāju grupu vienlaicīgai izpildei. Darbplūsma turpinās tikai tad, kad visi paralēlie apstrādātāji ir pabeiguši darbu.

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

Šajā piemērā:

1. Vispirms tiek izpildīta FetchUserHandler (secīgi)
2. Trīs apstrādātāji darbojas vienlaikus (paralēli)
3. MergeAndSaveHandler tiek izpildīta pēc visu trīs pabeigšanas (secīgi)

Kā tas darbojas

Kad tiek sasniegta paralēlā darbība, Queuety katram grupas apstrādātājam nosūta atsevišķu darbu. Katrs darbs saņem pašreizējo darbplūsmas stāvokli un tiek izpildīts neatkarīgi. Darbplūsma uzskaita, cik paralēlo darbu vēl ir izpildē, un pāriet pie nākamās darbības tikai tad, kad visi ir pabeigti.

Katra paralēlā apstrādātāja atgrieztā vērtība tiek sapludināta ar darbplūsmas stāvokli. Ja vairāki apstrādātāji atgriež vienu un to pašu atslēgu, uzvar pēdējais, kas pabeidz darbu.

Stāvokļu sapludināšana

Katrs paralēlais apstrādātājs atgriež datus, kas tiek sapludināti ar kopīgo darbplūsmas stāvokli:

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ēc paralēlās darbības pabeigšanas stāvoklī ir gan social_profile, gan payment_history, un nākamā secīgā darbība var piekļūt visam stāvoklim.

Secīgu un paralēlu darbību apvienošana

Varat brīvi kombinēt izsaukumus ->then() un ->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 );

Kļūdu apstrāde

Ja kāds paralēlās grupas apstrādātājs neizdodas, darbplūsma piemēro šim apstrādātājam parasto atkārtotas mēģināšanas loģiku. Citi paralēlie apstrādātāji, kuru darbs jau ir pabeigts, netiek palaisti atkārtoti. Kad neizdevušais apstrādātājs nostrādā veiksmīgi, darbplūsma turpinās. Ja tas izsmēlis atkārtotos mēģinājumus un ir paslēpts, darbplūsma tiek atzīmēta kā neizdevusies, bet atlikušie atzaru darbi tiek paslēpti; izmantojiet retry_workflow(), lai vēlreiz izpildītu neizdevušos darbību.