Tarpinė programinė įranga

Tarpinė programinė įranga apgaubia užduoties vykdymą sluoksniuotu konvejeriu. Kiekviena tarpinė programinė įranga gali vykdyti logiką prieš užduotį ir po jos, sustabdyti vykdymą anksčiau arba perimti išimtis. Queuety turi penkias įtaisytąsias tarpines programas ir palaiko pasirinktines per Contracts\Middleware sąsają.

Tarpinę programinę įrangą galima naudoti užduočių klasėse, kurias galima perduoti vykdyti (klasėse, įgyvendinančiose Contracts\Job ir naudojančiose savybę Dispatchable).

Sąsaja Contracts\Middleware

use Queuety\Contracts\Middleware;

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

Parametras $job yra apdorojamas užduoties egzempliorius. Iškvieskite $next( $job ), kad perduotumėte vykdymą kitai konvejerio tarpinei programinei įrangai (arba užduoties metodui handle(), jei tai paskutinė tarpinė programa).

Tarpinės programinės įrangos įtraukimas į užduotį

Užduoties klasėje apibrėžkite metodą middleware(), grąžinantį tarpinių programų egzempliorių masyvą:

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

Konvejerio vykdymo tvarka

Tarpinė programinė įranga taikoma masyvo tvarka, o pirmoji yra išorinis sluoksnis. Jei jūsų metodas middleware() grąžina [A, B, C], vykdymo tvarka yra tokia:

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

Kiekviena tarpinė programa apgaubia kitą. A pirmiausia vykdoma prieš užduotį, o paskiausiai – jai pasibaigus, todėl puikiai tinka trukmei matuoti arba try/catch apgaubai.

Pasirinktinės tarpinės programinės įrangos kūrimas

Įgyvendinkite sąsają 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" );
        }
    }
}

Naudokite ją užduotyje:

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

Vykdymo sustabdymas

Tarpinė programa gali visiškai praleisti užduotį, jei neiškviečia $next():

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

        $next( $job );
    }
}

Integruota tarpinė programinė įranga

ThrottlesExceptions

Apriboja užduoties apdorojimo funkcijos išimčių skaičių, kad būtų išvengta užduočių antplūdžio. Jei apdorojimo funkcijoje per tam tikrą laikotarpį įvyko per daug išimčių, vėlesnės jos užduotys atmetamos su RuntimeException, o ne apdorojamos. Taip neleidžiama neveikiančiai išorinei paslaugai vienu metu išeikvoti visų pakartotinių bandymų daugelyje užduočių.

use Queuety\Middleware\ThrottlesExceptions;

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

Išimčių skaičius sekamas pagal apdorojimo funkcijos klasę Queuety užraktų lentelėje ir bendras visiems vykdytojo procesams. Pasiekus ribą, tarpinė programa išmeta RuntimeException neiškvietusi užduoties, o vykdytojas tai apdoroja kaip bet kurią kitą klaidą: bandymas įskaičiuojamas į max_attempts, užduotis pakartojama taikant įprastą delsą, o pasibaigus bandymams perkeliama į nepavykusių užduočių saugyklą.

Pavyzdys: jei neveikia išorinė API, pirmieji 10 nesėkmingų bandymų apdorojami įprastai (kiekvienas padidina išimčių skaitiklį). Kiti bandymai iškart nepavyksta neiškvietus API, kol 10 minučių neįvyksta nauja išimtis; langas prasideda iš naujo po kiekvienos užregistruotos išimties.

RateLimited

Užtikrina didžiausią vykdymų skaičių per laikotarpį. Jei tarpinėje programinėje įrangoje riba viršijama, išmetama RateLimitExceededException; vykdytojas ją apdoroja kaip įprastą nesėkmingą bandymą, pakartoja taikydamas įprastą delsą ir įskaičiuoja į max_attempts. Tarpinei programai užregistravus ribą vykdytojo spartos ribotuve, vėlesnės apdorojimo funkcijos užduotys, kol galioja riba, taip pat grąžinamos į pending dar prieš jas vykdant (žr. Spartos ribojimas).

use Queuety\Middleware\RateLimited;

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

Spartos ribotuvas naudoja vidinį RateLimiter egzempliorių, priklausantį Queuety, todėl ribos taikomos visiems vykdytojo procesams.

Timeout

Nustato kiekvienos užduoties vykdymo laiko limitą naudodamas pcntl_alarm; numatyta išmesti TimeoutException, jei užduotis trunka per ilgai. Šiuo metu laiko limitas netaikomas (#10): Queuety neįjungia asinchroninių signalų, todėl žadintuvo apdorojimo funkcija nevykdoma, kol vykdoma užduotis. Be to, tarpinė programinė įranga grąžindama nustato žadintuvą iš naujo ir atšaukia paties vykdytojo užduočiai skirtą žadintuvą.

use Queuety\Middleware\Timeout;

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

Reikia plėtinio pcntl PHP. Jei pcntl nepasiekiamas, tarpinė programinė įranga tyliai praleidžiama (laiko limitas netaikomas).

UniqueJob

Neleidžia vienu metu vykdyti užduočių su tuo pačiu raktu. Naudoja duomenų bazės užraktų lentelę, kad vienu metu veiktų tik vienas egzempliorius. Jei užraktas jau laikomas, užduotis tyliai praleidžiama (nekartojama).

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

Neleidžia vienu metu vykdyti užduočių, kurioms naudojamas tas pats loginis raktas. Užraktas automatiškai nustoja galioti, kad apsaugotų nuo užstrigusių užraktų, jei vykdytojas netikėtai nutrūksta. Panašu į UniqueJob, tačiau skirta ilgai trunkančioms operacijoms, kurių užraktas turi būti automatiškai atleidžiamas pasibaigus skirtajam laikui.

use Queuety\Middleware\WithoutOverlapping;

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

Jei užraktas jau laikomas, užduotis tyliai praleidžiama. Prieš kiekvieną bandymą užrakinti automatiškai pašalinami nebegaliojantys užraktai.

Pavyzdys: tarpinių programų derinimas

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

Ši užduotis:

1. Bus nutraukta, jei vykdoma ilgiau nei 5 minutes (Timeout)
2. Bus praleista, jei jau vykdomas to paties tiekėjo importavimas (WithoutOverlapping)
3. Bus apribota iki 5 importavimų per minutę tarp visų tiekėjų (RateLimited)