Ara Katman Yazılımı

Ara katman yazılımı, iş yürütmesini soğan biçimli bir işlem hattına sarar. Her ara katman, işten önce ve sonra mantık çalıştırabilir, yürütmeyi kısa devre edebilir veya istisnaları yakalayabilir. Queuety, yerleşik beş ara katmanla gelir ve Contracts\Middleware arayüzü üzerinden özel ara katmanları destekler.

Ara katman yazılımı, dağıtılabilir iş sınıfları için kullanılabilir (Contracts\Job uygulayan ve Dispatchable trait'ini kullanan sınıflar).

Contracts\Middleware arayüzü

use Queuety\Contracts\Middleware;

interface Middleware {
    public function handle( object $job, \Closure $next ): void;
}

$job parametresi, işlenen iş örneğidir. İşlem hattındaki sonraki ara katmana (veya son ara katmansa işin handle() yöntemine) yürütmeyi aktarmak için $next( $job ) çağrısını yapın.

Bir işe ara katman yazılımı ekleme

İş sınıfınızda, ara katman örneklerinden oluşan bir dizi döndüren middleware() yöntemi tanımlayın:

use Queuety\Contracts\Job;
use Queuety\Dispatchable;
use Queuety\Middleware\RateLimited;
use Queuety\Middleware\Timeout;

class SyncInventoryJob implements Job {
    use Dispatchable;

    public function __construct(
        public readonly string $sku,
    ) {}

    public function handle(): void {
        // Sync inventory for this SKU...
    }

    public function middleware(): array {
        return [
            new Timeout( seconds: 30 ),
            new RateLimited( max: 100, window: 60 ),
        ];
    }
}

İşlem hattı yürütme sırası

Ara katmanlar dizi sırasına göre uygulanır; ilk ara katman en dış katmandır. middleware() yönteminiz [A, B, C] döndürürse yürütme sırası şöyledir:

A::handle() -> B::handle() -> C::handle() -> $job->handle()

Her ara katman bir sonrakini sarar. A işten önce ilk, işten sonra son çalışır; bu da onu zamanlama veya try/catch sarmalayıcıları için ideal kılar.

Özel ara katman yazma

Contracts\Middleware arayüzünü uygulayın:

use Queuety\Contracts\Middleware;

class LogExecution implements Middleware {
    public function handle( object $job, \Closure $next ): void {
        $class = get_class( $job );
        error_log( "[Queuety] Starting: {$class}" );

        $start = microtime( true );

        try {
            $next( $job );
        } finally {
            $elapsed = round( microtime( true ) - $start, 3 );
            error_log( "[Queuety] Finished: {$class} in {$elapsed}s" );
        }
    }
}

Bir işte kullanın:

public function middleware(): array {
    return [
        new LogExecution(),
    ];
}

Kısa devreye alma

Bir ara katman, $next() çağrısını yapmayarak işi tamamen atlayabilir:

class SkipOnMaintenanceMode implements Middleware {
    public function handle( object $job, \Closure $next ): void {
        if ( wp_is_maintenance_mode() ) {
            return; // Job silently skipped.
        }

        $next( $job );
    }
}

Yerleşik ara katman yazılımları

ThrottlesExceptions

İşleyiciden kaynaklanan istisnaları sınırlayarak iş dalgalarını önler. Bir işleyici belirli bir zaman aralığında çok fazla istisna fırlatmışsa, o işleyiciye ait sonraki işler işlenmek yerine RuntimeException ile reddedilir. Böylece başarısız bir harici hizmetin birçok işteki tüm yeniden deneme girişimlerini aynı anda tüketmesi önlenir.

use Queuety\Middleware\ThrottlesExceptions;

public function middleware(): array {
    return [
        new ThrottlesExceptions( max_attempts: 10, decay_minutes: 10 ),
    ];
}

İstisna sayıları, Queuety kilitler tablosunda işleyici sınıfına göre izlenir ve tüm işçi süreçlerinde paylaşılır. Eşiğe ulaşıldığında ara katman, işi çağırmadan RuntimeException fırlatır. İşçi bunu diğer hatalar gibi ele alır: deneme max_attempts sayısına eklenir, iş normal geri çekilme süresiyle yeniden denenir ve deneme hakkı bittiğinde gömülür.

Örnek: Harici bir API çalışmıyorsa ilk 10 başarısızlık normal şekilde işlenir (her biri istisna sayacını artırır). Yeni bir istisna olmadan 10 dakika geçene kadar sonraki denemeler API çağrılmadan hızla başarısız olur; pencere, kaydedilen her istisnayla yeniden başlar.

RateLimited

Bir zaman aralığında izin verilen en yüksek yürütme sayısını uygular. Ara katman içindeki sınır aşılırsa RateLimitExceededException fırlatılır; işçi bunu normal bir başarısız deneme gibi ele alır, normal geri çekilme süresiyle yeniden dener ve max_attempts sayısına ekler. Ara katman sınırı işçinin hız sınırlayıcısına kaydettikten sonra, sınır doluyken işleyiciye ait sonraki işler çalıştırılmadan pending kuyruğuna geri konur (bkz. Hız Sınırlama).

use Queuety\Middleware\RateLimited;

public function middleware(): array {
    return [
        new RateLimited( max: 60, window: 60 ), // 60 per minute
    ];
}

Hız sınırlayıcı, Queuety dahili RateLimiter örneğini kullanır; bu nedenle sınırlar tüm işçi süreçlerinde uygulanır.

Timeout

pcntl_alarm ile iş başına yürütme zaman aşımı başlatır ve iş çok uzun sürerse TimeoutException fırlatması amaçlanır. Zaman aşımı şu anda uygulanmıyor (#10): Queuety eşzamansız sinyalleri etkinleştirmediğinden, iş yürütülürken alarm işleyicisi çalışmaz. Ara katman ayrıca döndüğünde alarmı sıfırlar ve böylece işçinin kendi iş zaman aşımı alarmını iptal eder.

use Queuety\Middleware\Timeout;

public function middleware(): array {
    return [
        new Timeout( seconds: 30 ),
    ];
}

pcntl PHP uzantısını gerektirir. pcntl kullanılabilir değilse ara katman sessizce devam eder (zaman aşımı uygulanmaz).

UniqueJob

Aynı anahtara sahip işlerin eşzamanlı yürütülmesini önler. Aynı anda yalnızca bir örneğin çalışmasını sağlamak için veritabanı kilit tablosu kullanır. Kilit zaten tutuluyorsa iş sessizce atlanır (yeniden denenmez).

use Queuety\Middleware\UniqueJob;

public function middleware(): array {
    return [
        new UniqueJob(),                         // Lock key defaults to the job class name
        new UniqueJob( key: 'sync_inventory' ),   // Custom lock key
    ];
}

WithoutOverlapping

Aynı mantıksal anahtarı paylaşan işlerin çakışan yürütmelerini, çöken işçilerden kaynaklanan kilitlerin kalıcı olmasını önlemek için otomatik kilit süresi dolumuyla engeller. UniqueJob ile benzerdir ancak kilidin zaman aşımından sonra otomatik olarak bırakılması gereken uzun süreli işlemler için tasarlanmıştır.

use Queuety\Middleware\WithoutOverlapping;

public function middleware(): array {
    return [
        new WithoutOverlapping( key: 'import_products', release_after: 600 ),
    ];
}

Kilit zaten tutuluyorsa iş sessizce atlanır. Süresi dolan kilitler, her kilit alma girişiminden önce otomatik olarak temizlenir.

Örnek: ara katmanları birleştirme

use Queuety\Contracts\Job;
use Queuety\Dispatchable;
use Queuety\Middleware\RateLimited;
use Queuety\Middleware\Timeout;
use Queuety\Middleware\WithoutOverlapping;

class ImportProductsJob implements Job {
    use Dispatchable;

    public function __construct(
        public readonly int $supplier_id,
    ) {}

    public function handle(): void {
        // Fetch and import products from the supplier...
    }

    public function middleware(): array {
        return [
            new Timeout( seconds: 300 ),
            new WithoutOverlapping(
                key: "import_supplier_{$this->supplier_id}",
                release_after: 600,
            ),
            new RateLimited( max: 5, window: 60 ),
        ];
    }
}

Bu iş şunları yapar:

1. 5 dakikadan uzun sürerse sonlandırılır (Timeout)
2. Aynı tedarikçi için başka bir içe aktarma zaten çalışıyorsa atlanır (WithoutOverlapping)
3. Tüm tedarikçiler genelinde dakikada 5 içe aktarmayla sınırlandırılır (RateLimited)