Паралелни стъпки

Понякога няколко стъпки могат да се изпълняват едновременно. Методът ->parallel() изпраща група обработчици за едновременно изпълнение. Работният поток продължава едва след завършването на всички паралелни обработчици.

Методът ->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 ] );

В този пример:

1. Първо се изпълнява FetchUserHandler (последователно)
2. Три обработчика се изпълняват едновременно (паралелно)
3. MergeAndSaveHandler се изпълнява след завършването и на трите (последователно)

Как работи

Когато се достигне паралелна стъпка, Queuety изпраща по една задача за всеки обработчик в групата. Всяка задача получава текущото състояние на работния поток и се изпълнява независимо. Работният поток следи броя на оставащите паралелни задачи и продължава към следващата стъпка едва след приключването на всички.

Върнатата стойност от всеки паралелен обработчик се обединява със състоянието на работния поток. Ако няколко обработчика върнат един и същ ключ, остава стойността на този, който завърши последен.

Обединяване на състоянието

Всеки паралелен обработчик връща данни, които се обединяват със споделеното състояние на работния поток:

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

След приключването на паралелната стъпка състоянието съдържа и social_profile, и payment_history, а следващата последователна стъпка има достъп до всички данни.

Комбиниране на последователно и паралелно изпълнение

Можете свободно да комбинирате извикванията на ->then() и ->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 );

Обработка на грешки

Ако някой обработчик в паралелна група се провали, работният поток следва обичайната логика за повторен опит за този обработчик. Другите паралелни обработчици, които вече са завършили, не се изпълняват отново. След като неуспешният обработчик се изпълни успешно, работният поток продължава. Ако той изчерпи повторните си опити и бъде скрит, работният поток се маркира като неуспешен, а оставащите задачи по разклоненията се скриват; използвайте retry_workflow(), за да изпълните неуспешната стъпка отново.