Medzivrstva

Medzivrstva obaľuje vykonávanie úlohy do viacvrstvového procesu. Každá vrstva môže pred úlohou aj po nej spustiť vlastnú logiku, zastaviť vykonávanie alebo zachytiť výnimky. Queuety obsahuje päť vstavaných vrstiev a podporuje vlastné vrstvy prostredníctvom rozhrania Contracts\Middleware.

Medzivrstva je dostupná pre úlohy, ktoré možno odoslať (triedy implementujúce Contracts\Job s vlastnosťou Dispatchable).

Rozhranie Contracts\Middleware

use Queuety\Contracts\Middleware;

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

Parameter $job je spracovávaná inštancia úlohy. Zavolajte $next( $job ), aby ste odovzdali vykonávanie ďalšej vrstve v procese (alebo metóde handle() úlohy, ak ide o poslednú vrstvu).

Pridanie medzivrstvy k úlohe

V triede úlohy definujte metódu middleware(), ktorá vráti pole inštancií medzivrstiev:

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

Poradie vykonávania v procese

Medzivrstvy sa použijú v poradí poľa, pričom prvá tvorí najvonkajšiu vrstvu. Ak metóda middleware() vráti [A, B, C], poradie vykonávania bude takéto:

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

Každá medzivrstva obalí nasledujúcu. A sa spustí ako prvá pred úlohou a ako posledná po nej, preto sa hodí na meranie času alebo na obalenie bloku try/catch.

Písanie vlastnej medzivrstvy

Implementujte rozhranie 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žite ju v úlohe:

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

Predčasné ukončenie

Medzvrstva môže úlohu úplne preskočiť 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 );
    }
}

Vstavaný middleware

ThrottlesExceptions

Obmedzuje výnimky z obslužnej rutiny úlohy, aby zabránila náporu úloh. Ak obslužná rutina vyvolá v určitom časovom období priveľa výnimiek, ďalšie úlohy pre túto rutinu sa odmietnu s výnimkou RuntimeException namiesto spracovania. Tým sa zabráni tomu, aby zlyhávajúca externá služba naraz spotrebovala všetky opakované pokusy mnohých úloh.

use Queuety\Middleware\ThrottlesExceptions;

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

Počty výnimiek sa sledujú podľa triedy obslužnej rutiny v tabuľke zámkov patriacej Queuety a zdieľajú sa medzi všetkými procesmi pracovníkov. Po dosiahnutí limitu medzivrstva vyvolá RuntimeException bez zavolania úlohy a pracovník s tým naloží ako s každým iným zlyhaním: pokus sa započíta do max_attempts, úloha sa zopakuje s bežným oneskorením a po vyčerpaní pokusov sa uloží medzi neúspešné úlohy.

Príklad: ak je externé API mimo prevádzky, prvých 10 zlyhaní sa spracuje štandardne (každé zvýši počítadlo výnimiek). Ďalšie pokusy zlyhajú okamžite bez zavolania API, kým neuplynie 10 minút bez novej výnimky; pri každej zaznamenanej výnimke sa časové okno spustí znova.

RateLimited

Vynucuje maximálny počet vykonaní v časovom okne. Ak sa limit v medzivrstve prekročí, vyvolá sa RateLimitExceededException; pracovník to spracuje ako bežný neúspešný pokus, zopakuje ho s obvyklým oneskorením a započíta do max_attempts. Keď medzivrstva zaregistruje limit v obmedzovači frekvencie pracovníka, ďalšie úlohy pre danú obslužnú rutinu sa pred spustením vrátia do pending, ak je limit stále vyčerpaný (pozri Obmedzovanie frekvencie).

use Queuety\Middleware\RateLimited;

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

Obmedzovač frekvencie používa internú inštanciu RateLimiter patriacu Queuety, takže limity sa uplatňujú vo všetkých procesoch pracovníkov.

Timeout

Nastavuje časový limit vykonávania každej úlohy pomocou pcntl_alarm a má vyvolať TimeoutException, ak úloha trvá príliš dlho. Časový limit sa momentálne nevynucuje (#10): Queuety nepovoľuje asynchrónne signály, takže obslužná rutina alarmu sa počas vykonávania úlohy nespustí. Medzivrstva pri návrate tiež resetuje alarm, čím zruší vlastný alarm časového limitu úlohy pracovníka.

use Queuety\Middleware\Timeout;

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

Vyžaduje rozšírenie pcntl PHP. Ak pcntl nie je dostupné, medzivrstva potichu pokračuje bez uplatnenia časového limitu.

UniqueJob

Zabraňuje súbežnému vykonávaniu úloh s rovnakým kľúčom. Pomocou tabuľky databázových zámkov zabezpečí, že naraz beží iba jedna inštancia. Ak je zámok už obsadený, úloha sa potichu preskočí a nebude sa opakovať.

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 prekrývaniu úloh so spoločným logickým kľúčom a automaticky uvoľní zámok, aby nezostal zaseknutý po páde pracovníka. Podobá sa UniqueJob, ale je určený pre dlhotrvajúce operácie, pri ktorých sa má zámok po uplynutí limitu automaticky uvoľniť.

use Queuety\Middleware\WithoutOverlapping;

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

Ak je zámok už obsadený, úloha sa potichu preskočí. Platnosť uplynutých zámkov sa automaticky kontroluje pred každým pokusom o získanie zámku.

Príklad: kombinovanie medzivrstiev

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

Táto úloha:

1. Bude ukončená, ak pobeží dlhšie ako 5 minút (Timeout)
2. Bude preskočená, ak už prebieha iný import od rovnakého dodávateľa (WithoutOverlapping)
3. Bude obmedzená na 5 importov za minútu naprieč všetkými dodávateľmi (RateLimited)