Kroki równoległe

Czasami wiele kroków może działać jednocześnie. Metoda ->parallel() przekazuje do wykonania grupę procedur obsługi równolegle. Przepływ pracy przechodzi dalej dopiero po zakończeniu wszystkich procedur.

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

W tym przykładzie:

1. Najpierw uruchamiana jest FetchUserHandler (sekwencyjnie)
2. Trzy procedury obsługi działają równolegle
3. Po ich zakończeniu uruchamiana jest MergeAndSaveHandler (sekwencyjnie)

Jak to działa

Po osiągnięciu kroku równoległego Queuety uruchamia jedno zadanie dla każdej procedury obsługi w grupie. Każde zadanie otrzymuje bieżący stan przepływu pracy i działa niezależnie. Przepływ śledzi liczbę nierozliczonych zadań równoległych i przechodzi do następnego kroku dopiero po ich ukończeniu.

Wartość zwracana przez każdą równoległą procedurę obsługi jest scalana ze stanem przepływu pracy. Jeśli kilka procedur zwróci ten sam klucz, wygrywa ta, która zakończy się jako ostatnia.

Scalanie stanu

Każda równoległa procedura obsługi zwraca dane, które są scalane ze wspólnym stanem przepływu pracy:

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

Po zakończeniu kroku równoległego stan zawiera zarówno social_profile, jak i payment_history, a następny krok sekwencyjny może korzystać z całości.

Łączenie kroków sekwencyjnych i równoległych

Możesz swobodnie łączyć wywołania ->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 );

Obsługa błędów

Jeśli dowolny handler w grupie równoległej zakończy się niepowodzeniem, workflow stosuje standardową logikę ponawiania dla tego handlera. Pozostałe handlery równoległe, które już się zakończyły, nie są uruchamiane ponownie. Gdy nieudany handler zakończy się powodzeniem, workflow jest kontynuowany. Jeśli wyczerpie limit ponowień i zostanie przeniesiony do kwarantanny, workflow zostaje oznaczony jako nieudany, a pozostałe zadania w gałęzi trafiają do kwarantanny; użyj retry_workflow(), aby ponownie uruchomić nieudany krok.