Étapes parallèles

Il arrive que plusieurs étapes puissent s’exécuter simultanément. La méthode ->parallel() répartit un groupe de gestionnaires à exécuter en parallèle. Le flux de travail ne poursuit que lorsque tous les gestionnaires parallèles ont terminé.

La méthode ->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 ] );

Dans cet exemple :

1. FetchUserHandler s’exécute en premier (séquentiellement)
2. Trois gestionnaires s’exécutent simultanément (en parallèle)
3. MergeAndSaveHandler s’exécute après la fin des trois gestionnaires (séquentiellement)

Fonctionnement

Lorsqu’une étape parallèle est atteinte, Queuety répartit une tâche par gestionnaire du groupe. Chaque tâche reçoit l’état actuel du flux de travail et s’exécute indépendamment. Le flux suit le nombre de tâches parallèles en attente et ne passe à l’étape suivante que lorsqu’elles sont toutes terminées.

La valeur renvoyée par chaque gestionnaire parallèle est fusionnée dans l’état du flux de travail. Si plusieurs gestionnaires renvoient la même clé, la dernière réponse reçue l’emporte.

Fusion de l’état

Chaque gestionnaire parallèle renvoie des données qui sont fusionnées dans l’état partagé du flux de travail :

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

Une fois l’étape parallèle terminée, l’état contient social_profile et payment_history, et l’étape séquentielle suivante peut accéder à l’ensemble de ces données.

Combiner séquentiel et parallèle

Vous pouvez combiner librement les appels ->then() et ->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 );

Gestion des erreurs

Si un gestionnaire d’un groupe parallèle échoue, le workflow applique la logique normale de nouvelle tentative à ce gestionnaire. Les autres gestionnaires parallèles déjà terminés ne sont pas relancés. Une fois le gestionnaire défaillant exécuté avec succès, le workflow reprend. S’il épuise ses tentatives et est mis en quarantaine, le workflow est marqué comme ayant échoué et les tâches restantes de la branche sont mises en quarantaine ; utilisez retry_workflow() pour relancer l’étape défaillante.