Väliohjelmistot

Väliohjelmisto käärii työn suorituksen sipulimalliseen putkeen. Kukin väliohjelmisto voi suorittaa logiikkaa ennen työtä ja sen jälkeen, keskeyttää suorituksen lyhyeen tai ottaa poikkeukset kiinni. Queuety sisältää viisi valmista väliohjelmistoa ja tukee mukautettuja väliohjelmistoja Contracts\Middleware-rajapinnan kautta.

Väliohjelmistot ovat käytettävissä lähetettävissä oleville työluokille (luokat, jotka toteuttavat Contracts\Job-rajapinnan Dispatchable-ominaisuuden avulla).

Contracts\Middleware-rajapinta

use Queuety\Contracts\Middleware;

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

$job-parametri on käsiteltävä työinstanssi. Kutsu $next( $job )-metodia, jotta suoritus siirtyy putken seuraavalle väliohjelmistolle (tai työn handle()-metodille, jos tämä on viimeinen väliohjelmisto).

Väliohjelmiston lisääminen työhön

Määritä työluokalle middleware()-metodi, joka palauttaa taulukon väliohjelmisto-instansseja:

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

Putken suoritusjärjestys

Väliohjelmistot otetaan käyttöön taulukon järjestyksessä, ja ensimmäinen väliohjelmisto on uloin kerros. Jos middleware()-metodisi palauttaa arvon [A, B, C], suoritusjärjestys on seuraava:

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

Kukin väliohjelmisto käärii seuraavan. A suoritetaan ensin ennen työtä ja viimeisenä työn jälkeen, joten se sopii hyvin ajoituksen mittaamiseen tai try/catch-kääreeksi.

Mukautetun väliohjelmiston kirjoittaminen

Toteuta Contracts\Middleware-rajapinta:

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

Käytä sitä työssä näin:

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

Suorituksen keskeyttäminen

Väliohjelmisto voi ohittaa työn kokonaan jättämällä kutsumatta $next():

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

        $next( $job );
    }
}

Sisäänrakennetut väliohjelmistot

ThrottlesExceptions

Rajoittaa työn käsittelijän poikkeuksia työryöppyjen estämiseksi. Jos käsittelijä on heittänyt liian monta poikkeusta tietyn ajan sisällä, saman käsittelijän seuraavat työt hylätään RuntimeException-poikkeuksella käsittelyn sijaan. Näin estetään viallisen ulkoisen palvelun aiheuttama tilanne, jossa kaikkien töiden uudelleenyritykset kuluvat loppuun kerralla.

use Queuety\Middleware\ThrottlesExceptions;

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

Poikkeusmäärät lasketaan käsittelijäluokittain Queuety-järjestelmän lukitustaulussa, ja ne jaetaan kaikkien työntekijäprosessien kesken. Kun raja saavutetaan, väliohjelmisto heittää RuntimeException-poikkeuksen kutsumatta työtä. Työntekijä käsittelee sen kuten minkä tahansa muun virheen: yritys lasketaan mukaan max_attempts-rajaan, työ yritetään uudelleen normaalilla viiveellä ja siirretään sivuun, kun yritykset loppuvat.

Esimerkki: jos ulkoinen API ei ole käytettävissä, ensimmäiset 10 epäonnistumista käsitellään normaalisti (jokainen kasvattaa poikkeuslaskuria). Myöhemmät yritykset epäonnistuvat nopeasti kutsumatta API-käsittelijää, kunnes kuluu 10 minuuttia ilman uutta poikkeusta. Aikaikkuna alkaa alusta jokaisen kirjatun poikkeuksen yhteydessä.

RateLimited

Pakottaa suoritusten enimmäismäärän aikaikkunan aikana. Jos raja ylittyy väliohjelmistossa, heitetään RateLimitExceededException-poikkeus; työntekijä käsittelee sen tavallisena epäonnistuneena yrityksenä, joka yritetään uudelleen normaalilla viiveellä ja lasketaan mukaan max_attempts-rajaan. Kun väliohjelmisto on rekisteröinyt rajan työntekijän nopeusrajoittimeen, käsittelijän myöhemmät työt palautetaan myös pending-tilaan ennen suoritusta rajan ollessa voimassa (katso Nopeusrajoitukset).

use Queuety\Middleware\RateLimited;

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

Nopeusrajoitin käyttää Queuety-järjestelmän sisäistä RateLimiter-instanssia, joten rajoja valvotaan kaikissa työntekijäprosesseissa.

Timeout

Asettaa työkohtaisen suoritusajan aikakatkaisun pcntl_alarm-toiminnolla, jonka on tarkoitus heittää TimeoutException-poikkeus, kun työ kestää liian kauan. Aikakatkaisua ei tällä hetkellä valvota (#10): Queuety ei ota käyttöön asynkronisia signaaleja, joten hälytyskäsittelijä ei suoritu työn ollessa käynnissä. Väliohjelmisto myös nollaa hälytyksen palatessaan, mikä peruuttaa työntekijän oman työkohtaisen aikakatkaisuhälytyksen.

use Queuety\Middleware\Timeout;

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

Vaatii pcntl PHP -laajennuksen. Jos pcntl ei ole käytettävissä, väliohjelmisto ohittaa aikakatkaisun hiljaisesti (aikakatkaisua ei valvota).

UniqueJob

Estää samanaikaisten töiden suorituksen, kun niillä on sama avain. Käyttää tietokannan lukitustaulua varmistaakseen, että vain yksi instanssi suoritetaan kerrallaan. Jos lukitus on jo varattu, työ ohitetaan hiljaisesti (sitä ei yritetä uudelleen).

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

Estää saman loogisen avaimen jakavien töiden päällekkäisen suorituksen ja vanhentaa lukituksen automaattisesti, jotta kaatuneiden työntekijöiden jättämät lukitukset eivät jää voimaan. Samankaltainen kuin UniqueJob, mutta suunniteltu pitkäkestoisiin toimintoihin, joissa lukituksen tulee vapautua automaattisesti aikakatkaisun jälkeen.

use Queuety\Middleware\WithoutOverlapping;

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

Jos lukitus on jo varattu, työ ohitetaan hiljaisesti. Vanhentuneet lukitukset siivotaan automaattisesti ennen jokaista lukituksen hankintayritystä.

Esimerkki: väliohjelmistojen yhdistäminen

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ämä työ:

1. Keskeytetään, jos sen suoritus kestää yli 5 minuuttia (Timeout)
2. Ohitetaan, jos saman toimittajan toinen tuonti on jo käynnissä (WithoutOverlapping)
3. Rajoitetaan nopeuteen 5 tuontia minuutissa kaikkien toimittajien kesken (RateLimited)