Middleware

Middleware membungkus eksekusi pekerjaan dalam pipeline bergaya onion. Setiap middleware dapat menjalankan logika sebelum dan sesudah pekerjaan, menghentikan eksekusi lebih awal, atau menangkap pengecualian. Queuety menyertakan lima middleware bawaan dan mendukung middleware kustom melalui antarmuka Contracts\Middleware.

Middleware tersedia untuk kelas pekerjaan yang dapat dikirimkan (kelas yang mengimplementasikan Contracts\Job dengan trait Dispatchable).

Antarmuka Contracts\Middleware

use Queuety\Contracts\Middleware;

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

Parameter $job adalah instansi pekerjaan yang sedang diproses. Panggil $next( $job ) untuk meneruskan eksekusi ke middleware berikutnya dalam pipeline (atau ke metode handle() pekerjaan jika ini middleware terakhir).

Menambahkan middleware ke pekerjaan

Definisikan metode middleware() pada kelas pekerjaan Anda yang mengembalikan array berisi instansi middleware:

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 ),
        ];
    }
}

Urutan eksekusi pipeline

Middleware diterapkan sesuai urutan dalam array, dengan middleware pertama menjadi lapisan terluar. Jika metode middleware() Anda mengembalikan [A, B, C], urutan eksekusinya adalah:

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

Setiap middleware membungkus middleware berikutnya. A berjalan pertama sebelum pekerjaan dan terakhir setelah pekerjaan, sehingga ideal untuk pembungkus pengukuran waktu atau try/catch.

Menulis middleware kustom

Implementasikan antarmuka Contracts\Middleware:

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

Gunakan pada pekerjaan:

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

Menghentikan eksekusi lebih awal

Middleware dapat melewati pekerjaan sepenuhnya dengan tidak memanggil $next():

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

        $next( $job );
    }
}

Middleware bawaan

ThrottlesExceptions

Membatasi pengecualian dari handler pekerjaan untuk mencegah lonjakan kegagalan pekerjaan. Jika handler menimbulkan terlalu banyak pengecualian dalam rentang waktu tertentu, pekerjaan berikutnya untuk handler tersebut ditolak dengan RuntimeException alih-alih diproses. Ini mencegah layanan eksternal yang bermasalah menghabiskan semua percobaan ulang untuk banyak pekerjaan sekaligus.

use Queuety\Middleware\ThrottlesExceptions;

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

Jumlah pengecualian dilacak per kelas handler dalam tabel kunci milik Queuety dan dibagikan ke semua proses worker. Saat ambang batas tercapai, middleware menimbulkan RuntimeException tanpa memanggil pekerjaan, dan worker memperlakukannya seperti kegagalan lainnya: percobaan tersebut dihitung ke max_attempts, pekerjaan dicoba lagi dengan jeda normal, dan pekerjaan dikubur setelah semua percobaannya habis.

Contoh: jika API eksternal sedang tidak berfungsi, 10 kegagalan pertama diproses secara normal (masing-masing menambah penghitung pengecualian). Percobaan selanjutnya langsung gagal tanpa memanggil API sampai 10 menit berlalu tanpa pengecualian baru; jendela waktu dimulai ulang setiap kali pengecualian tercatat.

RateLimited

Memberlakukan jumlah maksimum eksekusi dalam rentang waktu tertentu. Jika batas terlampaui di dalam middleware, RateLimitExceededException ditimbulkan; worker menanganinya sebagai percobaan gagal biasa, mencobanya lagi dengan jeda normal, dan menghitungnya ke max_attempts. Setelah middleware mendaftarkan batas pada pembatas laju worker, pekerjaan berikutnya untuk handler tersebut juga dikembalikan ke pending sebelum dijalankan selama batas masih tercapai (lihat Pembatasan Laju).

use Queuety\Middleware\RateLimited;

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

Pembatas laju menggunakan instansi RateLimiter internal milik Queuety, sehingga batas diberlakukan di semua proses worker.

Timeout

Mengatur batas waktu eksekusi per pekerjaan dengan pcntl_alarm dan dimaksudkan untuk menimbulkan TimeoutException jika pekerjaan berjalan terlalu lama. Batas waktu saat ini tidak diberlakukan (#10): Queuety tidak mengaktifkan sinyal asinkron, sehingga handler alarm tidak berjalan saat pekerjaan dieksekusi. Middleware ini juga mengatur ulang alarm saat selesai, sehingga membatalkan alarm batas waktu pekerjaan milik worker.

use Queuety\Middleware\Timeout;

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

Memerlukan ekstensi pcntl PHP. Jika pcntl tidak tersedia, middleware tidak melakukan apa pun tanpa pemberitahuan (batas waktu tidak diberlakukan).

UniqueJob

Mencegah eksekusi bersamaan untuk pekerjaan dengan kunci yang sama. Menggunakan tabel kunci basis data untuk memastikan hanya satu instansi berjalan pada satu waktu. Jika kunci sudah dipegang, pekerjaan dilewati tanpa pemberitahuan (tidak dicoba lagi).

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

Mencegah eksekusi tumpang tindih pada pekerjaan yang berbagi kunci logis, dengan masa kedaluwarsa kunci otomatis untuk mencegah kunci macet akibat worker yang berhenti secara tidak normal. Mirip dengan UniqueJob, tetapi dirancang untuk operasi berdurasi panjang saat kunci perlu dilepas otomatis setelah batas waktu.

use Queuety\Middleware\WithoutOverlapping;

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

Jika kunci sudah dipegang, pekerjaan dilewati tanpa pemberitahuan. Kunci yang kedaluwarsa dibersihkan secara otomatis sebelum setiap upaya memperoleh kunci.

Contoh: menggabungkan middleware

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 ),
        ];
    }
}

Pekerjaan ini akan:

1. Dihentikan jika berjalan lebih dari 5 menit (Timeout)
2. Dilewati jika impor lain dari pemasok yang sama sedang berjalan (WithoutOverlapping)
3. Dibatasi hingga 5 impor per menit untuk semua pemasok (RateLimited)