Vahevara

Vahevara ümbritseb töö täitmist kihilise konveierina. Iga vahevara saab käivitada loogikat enne ja pärast tööd, täitmise enneaegselt lõpetada või püüda erandeid. Queuety sisaldab viit sisseehitatud vahevara ning toetab kohandatud vahevara liidese Contracts\Middleware kaudu.

Vahevara on saadaval väljakutsutavate tööklasside jaoks (klassid, mis rakendavad Contracts\Job ja kasutavad traiti Dispatchable).

Liides Contracts\Middleware

use Queuety\Contracts\Middleware;

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

Parameeter $job on töödeldav tööobjekt. Käivitage $next( $job ), et anda täitmine edasi konveieri järgmisele vahevarale (või viimase vahevara korral töö meetodile handle()).

Vahevara lisamine tööle

Määratlege oma tööklassis meetod middleware(), mis tagastab vahevara eksemplaride massiivi:

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

Konveieri täitmise järjekord

Vahevarad rakendatakse massiivi järjekorras, esimene vahevara on välimine kiht. Kui meetod middleware() tagastab [A, B, C], on täitmise järjekord järgmine:

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

Iga vahevara ümbritseb järgmist. A käivitub enne tööd esimesena ja pärast tööd viimasena, mistõttu sobib see hästi aja mõõtmiseks või try/catch-ümbriseks.

Kohandatud vahevara kirjutamine

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

Kasutage seda töö puhul:

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

Täitmise enneaegne lõpetamine

Vahevara võib töö täielikult vahele jätta, kui see ei käivita $next():

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

        $next( $job );
    }
}

Sisseehitatud vahevara

ThrottlesExceptions

Piirab töötlusfunktsiooni erandite arvu, et vältida tööde tulva. Kui töötlusfunktsioon on ajavahemiku jooksul liiga palju erandeid tekitanud, lükatakse selle töötleja järgmised tööd tagasi RuntimeException-iga, selle asemel et neid töödelda. Nii ei kuluta tõrkuv väline teenus korraga paljude tööde kõiki korduskatseid.

use Queuety\Middleware\ThrottlesExceptions;

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

Erandite arvu jälgitakse töötlejaklassi kaupa Queuety-i lukustustabelis ja see on ühine kõigile töötajaprotsessidele. Lävendi saavutamisel tekitab vahevara RuntimeException-i ilma tööd käivitamata ning töötaja käsitleb seda nagu iga muud tõrget: katse läheb arvesse max_attempts puhul, töö proovitakse uuesti tavapärase viivitusega ja pärast katsete ammendumist paigutatakse see ootele.

Näide: kui väline API ei tööta, käsitletakse esimest 10 tõrget tavapäraselt (igaüks suurendab erandite loendurit). Järgmised katsed ebaõnnestuvad kiiresti ega käivita API-i, kuni möödub 10 minutit ilma uue erandita; ajavahemik algab iga registreeritud erandi korral uuesti.

RateLimited

Kehtestab ajavahemiku jooksul tehtavate täitmiste ülempiiri. Kui vahevara sees piir ületatakse, tekitatakse RateLimitExceededException; töötaja käsitleb seda tavapärase ebaõnnestunud katsena, proovib uuesti tavapärase viivitusega ja arvestab selle hulka max_attempts puhul. Kui vahevara on piirangu töötaja päringupiirajale registreerinud, suunatakse selle töötleja hilisemad tööd piirangu kehtimise ajal enne käivitamist tagasi olekusse pending (vt Päringupiirang).

use Queuety\Middleware\RateLimited;

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

Päringupiiraja kasutab Queuety sisemist eksemplari RateLimiter, nii et piirangud kehtivad kõigis töötajaprotsessides.

Timeout

Seadistab tööpõhise täitmise ajalõpu, kasutades pcntl_alarm, ja peaks töö liiga pika kestuse korral tekitama TimeoutException. Ajalõppu praegu ei rakendata (#10): Queuety ei luba asünkroonseid signaale, mistõttu alarmihaldur ei tööta töö täitmise ajal. Vahevara lähtestab naastes ka alarmi, tühistades töötaja enda töö ajalõpu alarmi.

use Queuety\Middleware\Timeout;

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

Vajab laiendust pcntl PHP. Kui pcntl pole saadaval, jätkab vahevara vaikselt tööd (ajalõppu ei rakendata).

UniqueJob

Takistab sama võtmega tööde samaaegset täitmist. Kasutab andmebaasi lukustustabelit, et tagada korraga vaid ühe eksemplari käivitamine. Kui lukk on juba hõivatud, jäetakse töö vaikselt vahele (seda ei proovita uuesti).

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

Takistab sama loogilise võtmega tööde kattuvat täitmist ning vabastab luku automaatselt, et vältida krahhinud töötajate tekitatud püsivaid lukke. Sarnaneb UniqueJob-iga, kuid on mõeldud kaua kestvateks toiminguteks, mille lukk peaks pärast ajalõppu automaatselt vabanema.

use Queuety\Middleware\WithoutOverlapping;

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

Kui lukk on juba hõivatud, jäetakse töö vaikselt vahele. Aegunud lukud koristatakse automaatselt enne iga hõivamiskatset.

Näide: vahevarade kombineerimine

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

See töö:

1. Lõpetatakse, kui see kestab kauem kui 5 minutit (Timeout)
2. Jäetakse vahele, kui sama tarnija import juba käib (WithoutOverlapping)
3. Piiratakse 5 impordiga minutis kõigi tarnijate peale kokku (RateLimited)