Паралельні кроки

Іноді кілька кроків можна виконувати одночасно. Метод ->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(), щоб запустити цей крок ще раз.