Rinnakkaiset vaiheet

Joskus useita vaiheita voidaan suorittaa samanaikaisesti. ->parallel()-menetelmä käynnistää käsittelijäryhmän rinnakkain. Työnkulku etenee vasta, kun kaikki rinnakkaiset käsittelijät ovat valmistuneet.

->parallel()-menetelmä

use Queuety\Queuety;

Queuety::workflow( 'enrich_profile' )
    ->then( FetchUserHandler::class )
    ->parallel( [
        FetchSocialProfileHandler::class,
        FetchPaymentHistoryHandler::class,
        FetchActivityLogHandler::class,
    ] )
    ->then( MergeAndSaveHandler::class )
    ->dispatch( [ 'user_id' => 42 ] );

Tässä esimerkissä:

1. FetchUserHandler suoritetaan ensin (peräkkäin)
2. Kolme käsittelijää suoritetaan samanaikaisesti (rinnakkain)
3. MergeAndSaveHandler suoritetaan, kun kaikki kolme ovat valmistuneet (peräkkäin)

Näin se toimii

Kun rinnakkainen vaihe saavutetaan, Queuety lähettää yhden työn kutakin ryhmän käsittelijää kohti. Kukin työ saa nykyisen työnkulun tilan ja suoritetaan itsenäisesti. Työnkulku seuraa, kuinka monta rinnakkaista työtä on kesken, ja etenee seuraavaan vaiheeseen vasta, kun kaikki ovat valmistuneet.

Kunkin rinnakkaisen käsittelijän palautusarvo yhdistetään työnkulun tilaan. Jos useampi käsittelijä palauttaa saman avaimen, viimeisenä valmistuneen arvo jää voimaan.

Tilan yhdistäminen

Kukin rinnakkainen käsittelijä palauttaa tietoja, jotka yhdistetään työnkulun yhteiseen tilaan:

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

Rinnakkaisen vaiheen valmistuttua tila sisältää sekä social_profile että payment_history, ja seuraava peräkkäinen vaihe voi käyttää niitä kaikkia.

Peräkkäisten ja rinnakkaisten vaiheiden yhdistäminen

Voit yhdistellä vapaasti toimintojen ->then() ja ->parallel() kutsuja:

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

Virheenkäsittely

Jos jokin rinnakkaisen ryhmän käsittelijä epäonnistuu, työnkulku noudattaa kyseisen käsittelijän normaalia uudelleenyrityslogiikkaa. Muita jo valmistuneita rinnakkaisia käsittelijöitä ei suoriteta uudelleen. Kun epäonnistunut käsittelijä onnistuu, työnkulku jatkuu. Jos sen uudelleenyritykset loppuvat ja työ siirretään sivuun, työnkulku merkitään epäonnistuneeksi ja jäljellä olevat haaratyöt siirretään sivuun; suorita epäonnistunut vaihe uudelleen toiminnolla retry_workflow().