Παράλληλα βήματα

Μερικές φορές πολλά βήματα μπορούν να εκτελούνται ταυτόχρονα. Η μέθοδος ->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() για να εκτελέσετε ξανά το βήμα που απέτυχε.