Параллельные шаги

Иногда несколько шагов могут выполняться одновременно. Метод ->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(), чтобы снова выполнить шаг с ошибкой.