Lygiagretūs veiksmai

Kartais keli veiksmai gali būti vykdomi vienu metu. Metodas ->parallel() vienu metu paleidžia veiksmų apdorojimo funkcijų grupę. Darbo seka tęsiama tik tada, kai baigiamos visos lygiagrečios apdorojimo funkcijos.

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

Šiame pavyzdyje:

1. Pirmiausia vykdomas FetchUserHandler (nuosekliai)
2. Trys apdorojimo funkcijos vykdomos vienu metu (lygiagrečiai)
3. MergeAndSaveHandler vykdomas baigus visas tris (nuosekliai)

Kaip tai veikia

Pasiekus lygiagretų veiksmą, Queuety paleidžia po vieną užduotį kiekvienai grupės apdorojimo funkcijai. Kiekviena užduotis gauna dabartinę darbo sekos būseną ir vykdoma nepriklausomai. Darbo seka seka, kiek lygiagrečių užduočių dar liko, ir pereina prie kito veiksmo tik joms visoms pasibaigus.

Kiekvienos lygiagrečios apdorojimo funkcijos grąžinama reikšmė sujungiama su darbo sekos būsena. Jei kelios apdorojimo funkcijos grąžina tą patį raktą, laimi paskutinė užbaigta funkcija.

Būsenų sujungimas

Kiekviena lygiagreti apdorojimo funkcija grąžina duomenis, kurie sujungiami su bendra darbo sekos būsena:

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

Pasibaigus lygiagrečiam veiksmui, būsenoje yra ir social_profile, ir payment_history, o kitas nuoseklus veiksmas gali pasiekti visus šiuos duomenis.

Nuoseklių ir lygiagrečių veiksmų derinimas

Galite laisvai derinti ->then() ir ->parallel() iškvietimus:

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

Klaidų tvarkymas

Jei kuri nors lygiagrečios grupės apdorojimo funkcija nepavyksta, darbo seka taiko jai įprastą pakartojimo logiką. Kitos jau baigusios lygiagrečios apdorojimo funkcijos nevykdomos iš naujo. Sėkmingai įvykdžius nepavykusią funkciją, darbo seka tęsiama. Jei ji išnaudoja visus pakartojimus ir perkeliama į nepavykusių užduočių saugyklą, darbo seka pažymima kaip nepavykusi, o likusios šakų užduotys taip pat perkeliamos į nepavykusių užduočių saugyklą; naudokite retry_workflow(), kad vėl paleistumėte nepavykusį veiksmą.