Paralel Adımlar

Bazen birden çok adım aynı anda çalışabilir. ->parallel() yöntemi bir işleyici grubunu eşzamanlı olarak dağıtır. İş akışı yalnızca tüm paralel işleyiciler tamamlandığında ilerler.

->parallel() yöntemi

use Queuety\Queuety;

Queuety::workflow( 'enrich_profile' )
    ->then( FetchUserHandler::class )
    ->parallel( [
        FetchSocialProfileHandler::class,
        FetchPaymentHistoryHandler::class,
        FetchActivityLogHandler::class,
    ] )
    ->then( MergeAndSaveHandler::class )
    ->dispatch( [ 'user_id' => 42 ] );

Bu örnekte:

1. FetchUserHandler önce çalışır (sıralı)
2. Üç işleyici eşzamanlı çalışır (paralel)
3. MergeAndSaveHandler üçünün de tamamlanmasından sonra çalışır (sıralı)

Nasıl çalışır

Paralel bir adıma ulaşıldığında Queuety, gruptaki her işleyici için bir iş dağıtır. Her iş, mevcut iş akışı durumunu alır ve bağımsız çalışır. İş akışı, bekleyen paralel işlerin sayısını izler ve sonraki adıma ancak tümü tamamlandığında ilerler.

Her paralel işleyicinin dönüş değeri iş akışı durumuyla birleştirilir. Birden çok işleyici aynı anahtarı döndürürse en son tamamlanan kazanır.

Durum birleştirme

Her paralel işleyici, paylaşılan iş akışı durumuyla birleştirilen veriler döndürür:

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

Paralel adım tamamlandıktan sonra durum hem social_profile hem de payment_history değerini içerir ve sonraki sıralı adım bunların tümüne erişebilir.

Sıralı ve paralel adımları bir arada kullanma

->then() ve ->parallel() çağrılarını istediğiniz gibi bir arada kullanabilirsiniz:

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 );

Hata işleme

Paralel bir gruptaki herhangi bir işleyici başarısız olursa iş akışı, o işleyici için normal yeniden deneme mantığını izler. Tamamlanmış diğer paralel işleyiciler yeniden çalıştırılmaz. Başarısız işleyici başarılı olduğunda iş akışı devam eder. Yeniden deneme hakkı tükenir ve işleyici gömülürse iş akışı başarısız olarak işaretlenir ve kalan dal işleri gömülür; başarısız adımı yeniden çalıştırmak için retry_workflow() kullanın.