Middleware

Middleware indpakker jobudførelse i en løgformet pipeline. Hvert middleware-led kan udføre logik før og efter jobbet, afbryde udførelsen eller fange undtagelser. Queuety leveres med fem indbyggede middleware-led og understøtter tilpassede middleware-led via grænsefladen Contracts\Middleware.

Middleware er tilgængeligt for jobklasser, der kan sendes til behandling (klasser, der implementerer Contracts\Job med traiten Dispatchable).

Grænsefladen Contracts\Middleware

use Queuety\Contracts\Middleware;

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

Parameteren $job er den jobinstans, der behandles. Kald $next( $job ) for at sende udførelsen videre til det næste middleware-led i pipelinen (eller til jobbets metode handle(), hvis dette er det sidste middleware-led).

Føj middleware til et job

Definér en metode middleware() i din jobklasse, som returnerer et array med middlewareinstanser:

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

Rækkefølge for pipelineudførelse

Middleware anvendes i arrayrækkefølge, hvor det første middleware-led er det yderste lag. Hvis din metode middleware() returnerer [A, B, C], er udførelsesrækkefølgen:

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

Hvert middleware-led indpakker det næste. A kører først før jobbet og sidst efter jobbet, hvilket gør det velegnet til tidsmåling eller try/catch-indpakning.

Skriv tilpasset middleware

Implementér grænsefladen 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" );
        }
    }
}

Brug det på et job:

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

Afbryd før jobbet

Et middleware-led kan springe jobbet helt over ved ikke at kalde $next():

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

        $next( $job );
    }
}

Indbygget middleware

ThrottlesExceptions

Begrænser antallet af undtagelser fra en jobhandler for at forhindre jobstorme. Hvis en handler har udløst for mange undtagelser inden for et tidsinterval, afvises efterfølgende jobs for denne handler med en RuntimeException i stedet for at blive behandlet. Det forhindrer, at en ekstern tjeneste, der fejler, opbruger alle gentagelsesforsøg for mange jobs på én gang.

use Queuety\Middleware\ThrottlesExceptions;

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

Antallet af undtagelser registreres pr. handlerklasse i låsetabellen for Queuety og deles mellem alle workerprocesser. Når grænsen nås, udløser middleware-leddet en RuntimeException uden at kalde jobbet, og workeren behandler det som enhver anden fejl: forsøget tæller med i max_attempts, jobbet gentages med den normale backoff, og det lægges i depot, når alle forsøg er brugt.

Eksempel: Hvis en ekstern API er nede, behandles de første 10 fejl normalt (hver fejl øger undtagelsestælleren). Yderligere forsøg fejler straks uden at kalde API, indtil der er gået 10 minutter uden en ny undtagelse; vinduet starter forfra ved hver registreret undtagelse.

RateLimited

Håndhæver et maksimalt antal kørsler inden for et tidsinterval. Hvis grænsen overskrides inde i middleware-leddet, udløses en RateLimitExceededException; workeren håndterer den som et almindeligt mislykket forsøg, gentager det med den normale backoff og tæller det med i max_attempts. Når middleware-leddet har registreret grænsen hos workerens hastighedsbegrænser, bliver efterfølgende jobs for handleren også sat tilbage til pending, inden de køres, mens grænsen er nået (se Hastighedsbegrænsning).

use Queuety\Middleware\RateLimited;

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

Ratebegrænseren bruger Queuetys interne RateLimiter-instans, så begrænsningerne håndhæves på tværs af alle worker-processer.

Timeout

Armerer en timeout for det enkelte jobs kørsel med pcntl_alarm og skal udløse en TimeoutException, hvis jobbet kører for længe. Timeouten håndhæves ikke i øjeblikket (#10): Queuety aktiverer ikke asynkrone signaler, så alarmhandleren kører ikke, mens jobbet udføres. Middleware-softwaren nulstiller også alarmen, når den returnerer, hvilket annullerer workerens egen alarm for jobtimeout.

use Queuety\Middleware\Timeout;

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

Kræver udvidelsen pcntl PHP. Hvis pcntl ikke er tilgængelig, fortsætter middleware-softwaren lydløst (ingen timeout håndhæves).

UniqueJob

Forhindrer samtidig kørsel af jobs med samme nøgle. Bruger en låsetabel i databasen for at sikre, at kun én instans kører ad gangen. Hvis en lås allerede er taget, springes jobbet lydløst over (det forsøges ikke igen).

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

Forhindrer overlappende kørsel af jobs med samme logiske nøgle. Låsen udløber automatisk, så den ikke bliver hængende, hvis en worker går ned. Ligner UniqueJob, men er beregnet til langvarige handlinger, hvor låsen skal frigives automatisk efter en timeout.

use Queuety\Middleware\WithoutOverlapping;

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

Hvis en lås allerede er taget, springes jobbet lydløst over. Udløbne låse ryddes automatisk op før hvert forsøg på at tage en lås.

Eksempel: kombination af middleware

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

Dette job vil:

1. Blive stoppet, hvis det kører i mere end 5 minutter (Timeout)
2. Blive sprunget over, hvis en anden import for samme leverandør allerede kører (WithoutOverlapping)
3. Blive begrænset til 5 importer i minuttet på tværs af alle leverandører (RateLimited)