Middleware

Middleware obaluje zpracování úloh do vrstevnatého řetězce. Každý middleware může před úlohou i po ní spustit vlastní logiku, zastavit další zpracování nebo zachytit výjimky. Queuety obsahuje pět vestavěných middleware a podporuje vlastní prostřednictvím rozhraní Contracts\Middleware.

Middleware je k dispozici pro třídy úloh, které lze spustit (třídy implementující Contracts\Job s vlastností Dispatchable).

Rozhraní Contracts\Middleware

use Queuety\Contracts\Middleware;

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

Parametr $job je instance zpracovávané úlohy. Zavolejte $next( $job ), chcete-li předat zpracování dalšímu middleware v řetězci (nebo metodě handle() dané úlohy, pokud jde o poslední middleware).

Přidání middleware k úloze

Ve třídě úlohy definujte metodu middleware(), která vrací pole instancí 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 ),
        ];
    }
}

Pořadí zpracování v řetězci

Middleware se použijí v pořadí uvedeném v poli, přičemž první middleware tvoří vnější vrstvu. Pokud vaše metoda middleware() vrací [A, B, C], pořadí zpracování je následující:

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

Každý middleware obaluje následující. A se spustí jako první před úlohou a jako poslední po ní, takže se hodí pro měření času nebo obalení pomocí try/catch.

Vytvoření vlastního middleware

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

Použijte ho u úlohy:

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

Přeskočení zpracování

Middleware může úlohu zcela přeskočit tím, že nezavolá $next():

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

        $next( $job );
    }
}

Vestavěný middleware

ThrottlesExceptions

Omezuje počet výjimek z obslužné rutiny úlohy, aby se zabránilo hromadnému selhání úloh. Pokud obslužná rutina vyvolá během časového okna příliš mnoho výjimek, následující úlohy této rutiny budou odmítnuty s RuntimeException místo zpracování. Tím se zabrání tomu, aby selhávající externí služba vyčerpala všechny pokusy o opakování u mnoha úloh najednou.

use Queuety\Middleware\ThrottlesExceptions;

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

Počty výjimek se sledují pro každou třídu obslužné rutiny v tabulce zámků Queuety a sdílejí se mezi všemi procesy workerů. Po dosažení limitu middleware vyvolá RuntimeException, aniž by zavolal úlohu. Worker s ní zachází stejně jako s jakýmkoli jiným selháním: pokus se započítá do max_attempts, úloha se znovu zařadí s běžnou prodlevou a po vyčerpání pokusů se odloží.

Příklad: pokud externí API nefunguje, prvních 10 selhání se zpracuje běžným způsobem (každé zvýší počítadlo výjimek). Další pokusy selžou okamžitě, aniž by se zavolalo API, dokud neuplyne 10 minut bez nové výjimky; s každou zaznamenanou výjimkou se časové okno spustí znovu.

RateLimited

Vynucuje maximální počet spuštění během časového okna. Pokud je limit uvnitř middleware překročen, vyvolá se RateLimitExceededException; worker ji zpracuje jako běžný neúspěšný pokus, znovu jej zařadí s běžnou prodlevou a započítá do max_attempts. Jakmile middleware zaregistruje limit u omezovače rychlosti workeru, další úlohy této obslužné rutiny se při dosažení limitu před spuštěním také vrátí do pending (viz Omezování rychlosti).

use Queuety\Middleware\RateLimited;

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

Omezovač rychlosti používá interní instanci Queuety RateLimiter, takže se limity uplatňují napříč všemi procesy workerů.

Timeout

Nastaví časový limit spuštění pro každou úlohu pomocí pcntl_alarm a při příliš dlouhém běhu úlohy má vyvolat TimeoutException. Časový limit se v současnosti neuplatňuje (#10): Queuety nepovoluje asynchronní signály, takže se obslužná rutina alarmu během provádění úlohy nespustí. Middleware také při návratu alarm resetuje, čímž ruší vlastní alarm časového limitu úlohy workeru.

use Queuety\Middleware\Timeout;

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

Vyžaduje rozšíření pcntl PHP. Pokud pcntl není k dispozici, middleware se tiše přeskočí (časový limit se neuplatní).

UniqueJob

Zabraňuje souběžnému spuštění úloh se stejným klíčem. K zajištění, že vždy běží jen jedna instance, používá databázovou tabulku zámků. Pokud je zámek již obsazen, úloha se tiše přeskočí (bez opakování).

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

Zabraňuje překrývání spuštění úloh se stejným logickým klíčem a automaticky nechává zámek vypršet, aby se zabránilo uváznutí při pádu workerů. Podobá se UniqueJob, ale je určený pro dlouhotrvající operace, u kterých se má zámek po uplynutí časového limitu automaticky uvolnit.

use Queuety\Middleware\WithoutOverlapping;

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

Pokud je zámek již obsazen, úloha se tiše přeskočí. Zámky, jejichž platnost vypršela, se automaticky vyčistí před každým pokusem o získání zámku.

Příklad: kombinace 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 ),
        ];
    }
}

Tato úloha:

1. Bude ukončena, pokud poběží déle než 5 minut (Timeout)
2. Bude přeskočena, pokud již běží jiný import stejného dodavatele (WithoutOverlapping)
3. Bude omezena na 5 importů za minutu napříč všemi dodavateli (RateLimited)