Pași paraleli

Uneori, mai mulți pași pot rula simultan. Metoda ->parallel() expediază un grup de gestionare pentru executare simultană. Fluxul de lucru avansează numai după ce s-au încheiat toate gestionarele paralele.

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

În acest exemplu:

1. FetchUserHandler rulează prima (secvențial)
2. Trei gestionare rulează simultan (în paralel)
3. MergeAndSaveHandler rulează după încheierea tuturor celor trei (secvențial)

Cum funcționează

Când se ajunge la un pas paralel, Queuety expediază câte o sarcină pentru fiecare gestionar din grup. Fiecare sarcină primește starea curentă a fluxului de lucru și rulează independent. Fluxul urmărește numărul de sarcini paralele încă active și avansează la pasul următor numai după încheierea tuturor.

Valoarea returnată de fiecare gestionar paralel este îmbinată în starea fluxului de lucru. Dacă mai multe gestionare returnează aceeași cheie, câștigă cel care se încheie ultimul.

Îmbinarea stării

Fiecare gestionar paralel returnează date care sunt îmbinate în starea comună a fluxului de lucru:

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

După finalizarea pasului paralel, starea conține atât social_profile, cât și payment_history, iar pasul secvențial următor le poate accesa pe toate.

Combinarea pașilor secvențiali și paraleli

Puteți combina liber apelurile ->then() și ->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 );

Tratarea erorilor

Dacă eșuează oricare handler dintr-un grup paralel, fluxul de lucru urmează logica normală de reîncercare pentru acel handler. Ceilalți handleri paraleli care s-au încheiat deja nu sunt rulați din nou. După ce handlerul eșuat reușește, fluxul de lucru continuă. Dacă își epuizează reîncercările și este pus în carantină, fluxul de lucru este marcat ca eșuat, iar joburile rămase din ramură sunt puse în carantină; folosiți retry_workflow() pentru a rula din nou pasul eșuat.