Starpprogrammatūra

Starpprogrammatūra aptver darbu izpildi caur sīpola tipa konveijeru. Katra starpprogrammatūra var izpildīt loģiku pirms un pēc darba, pārtraukt izpildi vai uztvert izņēmumus. Queuety ir iekļautas piecas iebūvētas starpprogrammatūras, un pielāgotas starpprogrammatūras tiek atbalstītas, izmantojot Contracts\Middleware saskarni.

Starpprogrammatūra ir pieejama nosūtāmajām darbu klasēm (klasēm, kas implementē Contracts\Job ar Dispatchable trait).

Saskarne Contracts\Middleware

use Queuety\Contracts\Middleware;

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

Parametrs $job ir apstrādājamā darba instance. Izsauciet $next( $job ), lai nodotu izpildi nākamajai starpprogrammatūrai konveijerā (vai darba metodei handle(), ja šī ir pēdējā starpprogrammatūra).

Starpprogrammatūras pievienošana darbam

Definējiet metodi middleware() darba klasē, kas atgriež starpprogrammatūras instanču masīvu:

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

Konveijera izpildes secība

Starpprogrammatūras tiek lietotas masīva secībā, un pirmā starpprogrammatūra ir ārējais slānis. Ja jūsu metode middleware() atgriež [A, B, C], izpildes secība ir šāda:

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

Katra starpprogrammatūra aptver nākamo. A tiek izpildīta pirmā pirms darba un pēdējā pēc darba, tāpēc tā ir piemērota laika mērīšanai vai try/catch aptverošajai loģikai.

Pielāgotas starpprogrammatūras rakstīšana

Implementējiet saskarni 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" );
        }
    }
}

Izmantojiet to darbā:

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

Izpildes pārtraukšana

Starpprogrammatūra var pilnībā izlaist darbu, neizsaucot $next():

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

        $next( $job );
    }
}

Iebūvētā starpprogrammatūra

ThrottlesExceptions

Ierobežo izņēmumus, ko rada darba apdarinātājs, lai novērstu darbu vētras. Ja apdarinātājs noteiktā laika periodā ir radījis pārāk daudz izņēmumu, turpmākie šī apdarinātāja darbi tiek noraidīti ar RuntimeException, nevis apstrādāti. Tas neļauj kļūdainam ārējam pakalpojumam vienlaikus iztērēt visus atkārtotas mēģināšanas mēģinājumus daudziem darbiem.

use Queuety\Middleware\ThrottlesExceptions;

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

Izņēmumu skaits tiek uzskaitīts katrai apdarinātāja klasei Queuety bloķēšanas tabulā un ir kopīgs visiem darbinieku procesiem. Sasniedzot slieksni, starpprogrammatūra izmet RuntimeException, neizsaucot darbu, un darbinieks to apstrādā kā jebkuru citu kļūmi: mēģinājums tiek pieskaitīts pie max_attempts, darbs tiek mēģināts vēlreiz ar parasto aizturi un pēc visu mēģinājumu izbeigšanās tiek ievietots apbedīto darbu sarakstā.

Piemērs: ja ārējais API nedarbojas, pirmās 10 kļūmes tiek apstrādātas kā parasti (katra palielina izņēmumu skaitītāju). Turpmākie mēģinājumi ātri neizdodas, neizsaucot API, līdz 10 minūtes nav reģistrēts neviens jauns izņēmums; periods sākas no jauna ar katru reģistrēto izņēmumu.

RateLimited

Piespiedu kārtā ierobežo maksimālo izpildes reižu skaitu noteiktā laika periodā. Ja starpprogrammatūras ietvaros limits tiek pārsniegts, tiek izsviests RateLimitExceededException; darbinieks to apstrādā kā parastu neveiksmīgu mēģinājumu, kas tiek atkārtots ar parasto aizturi un pieskaitīts pie max_attempts. Kad starpprogrammatūra ir reģistrējusi limitu darbinieka ātruma ierobežotājā, turpmākie šī apdarinātāja darbi arī tiek ievietoti atpakaļ pending, pirms tie tiek izpildīti, kamēr limits ir sasniegts (skatiet Ātruma ierobežošana).

use Queuety\Middleware\RateLimited;

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

Ātruma ierobežotājs izmanto Queuety iekšējo instanci RateLimiter, tāpēc limiti tiek piemēroti visos darbinieku procesos.

Timeout

Iestata katra darba izpildes taimautu, izmantojot pcntl_alarm, un ir paredzēts, lai, ja darbs ieilgst, tiktu izsviests TimeoutException. Taimauts pašlaik netiek piemērots (#10): Queuety neiespējo asinhronos signālus, tāpēc trauksmes apdarinātājs darba izpildes laikā netiek palaists. Starpprogrammatūra arī atiestata trauksmi, kad atgriežas, un tādējādi atceļ darbinieka paša darba taimauta trauksmi.

use Queuety\Middleware\Timeout;

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

Nepieciešams paplašinājums pcntl PHP. Ja pcntl nav pieejams, starpprogrammatūra klusējot nedara neko (taimauts netiek piemērots).

UniqueJob

Novērš vienas atslēgas darbu vienlaicīgu izpildi. Izmanto datubāzes bloķēšanas tabulu, lai vienlaikus darbotos tikai viena instance. Ja bloķētājs jau ir aizņemts, darbs tiek klusi izlaists (netiek mēģināts vēlreiz).

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

Novērš tādu darbu pārklājošu izpildi, kuriem ir kopīga loģiskā atslēga, un automātiski atbrīvo bloķētāju, lai nepieļautu iestrēgušus bloķētājus, ko radījuši avarējuši darbinieki. Līdzīga UniqueJob, bet paredzēta ilgstošām darbībām, kur bloķētājam automātiski jāatbrīvojas pēc taimauta.

use Queuety\Middleware\WithoutOverlapping;

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

Ja bloķētājs jau ir aizņemts, darbs tiek klusi izlaists. Beidzies bloķētāju derīguma termiņš tiek automātiski notīrīts pirms katra bloķēšanas mēģinājuma.

Piemērs: starpprogrammatūru apvienošana

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

Šis darbs tiks:

1. Apturēts, ja izpilde ilgst vairāk par 5 minūtēm (Timeout)
2. Izlaists, ja jau notiek cita tā paša piegādātāja importēšana (WithoutOverlapping)
3. Ierobežots līdz 5 importēšanas reizēm minūtē visiem piegādātājiem (RateLimited)