Parallelle stappen

Soms kunnen meerdere stappen tegelijk worden uitgevoerd. De methode ->parallel() start een groep handlers gelijktijdig. De workflow gaat pas verder wanneer alle parallelle handlers zijn voltooid.

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

In dit voorbeeld:

1. wordt FetchUserHandler eerst uitgevoerd (sequentieel)
2. worden drie handlers gelijktijdig uitgevoerd (parallel)
3. wordt MergeAndSaveHandler uitgevoerd nadat alle drie zijn voltooid (sequentieel)

Werking

Wanneer een parallelle stap wordt bereikt, start Queuety één taak per handler in de groep. Elke taak ontvangt de huidige workflowstatus en wordt onafhankelijk uitgevoerd. De workflow houdt bij hoeveel parallelle taken nog lopen en gaat pas verder met de volgende stap wanneer ze allemaal zijn voltooid.

De retourwaarde van elke parallelle handler wordt samengevoegd met de workflowstatus. Als meerdere handlers dezelfde sleutel retourneren, wint de handler die als laatste wordt voltooid.

Status samenvoegen

Elke parallelle handler retourneert gegevens die worden samengevoegd met de gedeelde workflowstatus:

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

Nadat de parallelle stap is voltooid, bevat de status zowel social_profile als payment_history, en heeft de volgende sequentiële stap toegang tot alle gegevens.

Sequentiële en parallelle stappen combineren

U kunt aanroepen van ->then() en ->parallel() naar wens combineren:

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

Foutafhandeling

Als een handler in een parallelle groep mislukt, volgt de workflow de normale logica voor nieuwe pogingen voor die handler. Andere parallelle handlers die al zijn voltooid, worden niet opnieuw uitgevoerd. Zodra de mislukte handler slaagt, gaat de workflow verder. Als alle nieuwe pogingen mislukken en de taak wordt begraven, wordt de workflow als mislukt gemarkeerd en worden de resterende vertakkingstaken begraven; gebruik retry_workflow() om de mislukte stap opnieuw uit te voeren.