Vmesna programska oprema

Vmesna programska oprema ovije izvajanje opravil v cevovod v obliki čebule. Vsak vmesni člen lahko izvede logiko pred opravilom in po njem, prekine izvajanje ali ujame izjeme. Queuety vključuje pet vgrajenih vmesnih členov in podpira tudi lastne prek vmesnika Contracts\Middleware.

Vmesna programska oprema je na voljo za razrede opravil , ki jih je mogoče odpremljati (razredi, ki implementirajo Contracts\Job z značko Dispatchable).

Vmesnik Contracts\Middleware

use Queuety\Contracts\Middleware;

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

Parameter $job je primerek opravila, ki se obdeluje. Pokličite $next( $job ), da nadaljujete izvajanje z naslednjim vmesnim členom v cevovodu (ali z metodo handle() opravila, če je to zadnji vmesni člen).

Dodajanje vmesne programske opreme opravilu

V razredu opravila določite metodo middleware(), ki vrne polje primerkov vmesnih členov:

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

Vrstni red izvajanja cevovoda

Vmesni členi se uporabijo po vrstnem redu v polju, pri čemer je prvi zunanji sloj. Če vaša metoda middleware() vrne [A, B, C], je vrstni red izvajanja naslednji:

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

Vsak vmesni člen ovije naslednjega. A se izvede prvi pred opravilom in zadnji po njem, zato je primeren za merjenje časa ali ovijanje z blokom try/catch.

Pisanje lastne vmesne programske opreme

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

Uporabite ga pri opravilu:

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

Kratek stik

Vmesni člen lahko v celoti preskoči opravilo, če ne pokliče $next():

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

        $next( $job );
    }
}

Vgrajena vmesna programska oprema

ThrottlesExceptions

Omejuje izjeme, ki jih sproži obravnavalnik opravil, da prepreči nenaden porast opravil. Če obravnavalnik v časovnem obdobju sproži preveč izjem, bodo nadaljnja opravila zanj zavrnjena z RuntimeException, namesto da bi se obdelala. Tako preprečite, da bi nedelujoča zunanja storitev naenkrat izčrpala vse poskuse ponovitve pri številnih opravilih.

use Queuety\Middleware\ThrottlesExceptions;

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

Število izjem se spremlja za vsak razred obravnavalnika posebej v tabeli zaklepanja Queuety in je skupno vsem delavskim procesom. Ko je dosežen prag, vmesni člen sproži RuntimeException, ne da bi poklical opravilo, delavec pa to obravnava kot vsako drugo napako: poskus se šteje v max_attempts, opravilo se ponovi z običajnim zamikom, po izčrpanju poskusov pa se premakne v arhiv.

Primer: če zunanji API ne deluje, se prvih 10 napak obdela običajno (vsaka poveča števec izjem). Nadaljnji poskusi se hitro zaključijo brez klica API, dokler 10 minut ne mine brez nove izjeme; obdobje se začne znova ob vsaki zabeleženi izjemi.

RateLimited

Zagotavlja največje število izvajanj v časovnem obdobju. Če je omejitev znotraj vmesnega člena presežena, se sproži RateLimitExceededException; delavec jo obravnava kot običajen neuspešen poskus, ki se ponovi z običajnim zamikom in šteje v max_attempts. Ko vmesni člen omejitev registrira pri omejevalniku hitrosti delavca, se poznejša opravila za obravnavalnik, dokler je omejitev dosežena, pred izvajanjem prav tako vrnejo v pending (glejte Omejevanje hitrosti).

use Queuety\Middleware\RateLimited;

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

Omejevalnik hitrosti uporablja notranji primerek RateLimiter, ki ga zagotavlja Queuety, zato omejitve veljajo za vse delavske procese.

Timeout

Nastavi časovno omejitev izvajanja posameznega opravila z pcntl_alarm in naj bi sprožil TimeoutException, če izvajanje opravila traja predolgo. Časovna omejitev trenutno ni uveljavljena (#10): Queuety ne omogoča asinhronih signalov, zato se upravljalnik alarma med izvajanjem opravila ne zažene. Vmesni člen ob vrnitvi tudi ponastavi alarm, s čimer prekliče lasten alarm delavca za časovno omejitev opravila.

use Queuety\Middleware\Timeout;

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

Zahteva razširitev pcntl PHP. Če pcntl ni na voljo, vmesni člen tiho preskoči izvajanje (časovna omejitev se ne uveljavi).

UniqueJob

Preprečuje sočasno izvajanje opravil z istim ključem. Uporablja tabelo zaklepanja v zbirki podatkov, da zagotovi izvajanje samo enega primerka hkrati. Če je zaklep že zaseden, se opravilo tiho preskoči (ne ponovi se).

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

Preprečuje prekrivanje izvajanja opravil z istim logičnim ključem in samodejno sprosti zaklep, da prepreči njegovo nedokončno zasedenost zaradi sesutih delavcev. Podobno je kot UniqueJob, vendar je zasnovano za dolgotrajna opravila, pri katerih se mora zaklep po časovni omejitvi samodejno sprostiti.

use Queuety\Middleware\WithoutOverlapping;

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

Če je zaklep že zaseden, se opravilo tiho preskoči. Potekli zaklepi se samodejno očistijo pred vsakim poskusom pridobitve zaklepa.

Primer: kombiniranje vmesne programske opreme

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

To opravilo bo:

1. Ukinjeno, če traja več kot 5 minut (Timeout)
2. Preskočeno, če se že izvaja drug uvoz za istega dobavitelja (WithoutOverlapping)
3. Omejeno na 5 uvozov na minuto za vse dobavitelje (RateLimited)