Parallele Schritte

Manchmal können mehrere Schritte gleichzeitig ausgeführt werden. Die Methode ->parallel() startet eine Gruppe von Handlern parallel. Der Workflow wird erst fortgesetzt, wenn alle parallelen Handler abgeschlossen sind.

Die 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 diesem Beispiel:

1. FetchUserHandler wird zuerst ausgeführt (sequenziell)
2. Drei Handler werden gleichzeitig ausgeführt (parallel)
3. MergeAndSaveHandler wird ausgeführt, nachdem alle drei abgeschlossen sind (sequenziell)

So funktioniert es

Sobald ein paralleler Schritt erreicht ist, startet Queuety einen Auftrag pro Handler in der Gruppe. Jeder Auftrag erhält den aktuellen Workflow-Status und wird unabhängig ausgeführt. Der Workflow erfasst die Anzahl der noch ausstehenden parallelen Aufträge und wird erst fortgesetzt, wenn alle abgeschlossen sind.

Der Rückgabewert jedes parallelen Handlers wird mit dem Workflow-Status zusammengeführt. Geben mehrere Handler denselben Schlüssel zurück, gilt der zuletzt abgeschlossene Handler.

Status zusammenführen

Jeder parallele Handler gibt Daten zurück, die mit dem gemeinsamen Workflow-Status zusammengeführt werden:

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

Nach Abschluss des parallelen Schritts enthält der Status sowohl social_profile als auch payment_history, und der nächste sequenzielle Schritt kann auf alles zugreifen.

Sequenzielle und parallele Schritte kombinieren

Du kannst Aufrufe von ->then() und ->parallel() beliebig kombinieren:

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

Fehlerbehandlung

Schlägt ein Handler in einer parallelen Gruppe fehl, folgt der Workflow der üblichen Wiederholungslogik für diesen Handler. Andere parallele Handler, die bereits abgeschlossen sind, werden nicht erneut ausgeführt. Sobald der fehlgeschlagene Handler erfolgreich ist, wird der Workflow fortgesetzt. Sind alle Wiederholungen ausgeschöpft und der Auftrag zurückgestellt, wird der Workflow als fehlgeschlagen markiert und die übrigen Zweigaufträge werden zurückgestellt. Verwende retry_workflow(), um den fehlgeschlagenen Schritt erneut auszuführen.