Mellanprogramvara

Mellanprogramvara omsluter jobbkörningen i en pipeline utformad som en lök. Varje mellanprogramvara kan köra logik före och efter jobbet, avbryta körningen i förtid eller fånga undantag. Queuety levereras med fem inbyggda mellanprogramvaror och stöder anpassade mellanprogramvaror via gränssnittet Contracts\Middleware.

Mellanprogramvara kan användas för distribuerbara jobbklasser (klasser som implementerar Contracts\Job med egenskapen Dispatchable).

Gränssnittet Contracts\Middleware

use Queuety\Contracts\Middleware;

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

Parametern $job är jobbinstansen som behandlas. Anropa $next( $job ) för att skicka körningen vidare till nästa mellanprogramvara i pipelinen (eller till jobbets metod handle() om detta är den sista mellanprogramvaran).

Lägga till mellanprogramvara i ett jobb

Definiera en metod middleware() i jobbklassen som returnerar en array med instanser av mellanprogramvara:

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

Körordning i pipelinen

Mellanprogramvarorna tillämpas i arrayordning, där den första utgör det yttersta lagret. Om metoden middleware() returnerar [A, B, C] ser körordningen ut så här:

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

Varje mellanprogramvara omsluter nästa. A körs först före jobbet och sist efter jobbet, vilket gör den idealisk för tidtagning eller omslutande try/catch-logik.

Skriva anpassad mellanprogramvara

Implementera gränssnittet 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" );
        }
    }
}

Använd den i ett jobb:

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

Avbryta körningen i förtid

En mellanprogramvara kan hoppa över jobbet helt genom att inte anropa $next():

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

        $next( $job );
    }
}

Inbyggd middleware

ThrottlesExceptions

Begränsar antalet undantag från en jobbhanterare för att förhindra jobbstormar. Om en hanterare har kastat för många undantag inom ett tidsintervall avvisas efterföljande jobb för den hanteraren med ett RuntimeException i stället för att behandlas. Detta hindrar en felande extern tjänst från att förbruka alla omförsök för många jobb på en gång.

use Queuety\Middleware\ThrottlesExceptions;

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

Undantag räknas per hanterarklass i låstabellen för Queuety, och räknaren delas mellan alla arbetarprocesser. När gränsen nås kastar mellanprogramvaran ett RuntimeException utan att anropa jobbet. Arbetaren behandlar det som vilket annat fel som helst: försöket räknas mot max_attempts, jobbet görs om med normal väntetid och läggs i karantän när försöken är slut.

Exempel: om en extern API ligger nere behandlas de första 10 felen normalt (vart och ett ökar undantagsräknaren). Ytterligare försök misslyckas direkt utan att anropa API, tills 10 minuter har gått utan ett nytt undantag. Fönstret startar om vid varje registrerat undantag.

RateLimited

Begränsar antalet körningar inom ett tidsintervall. Om gränsen överskrids i mellanprogramvaran kastas ett RateLimitExceededException. Arbetaren hanterar det som ett vanligt misslyckat försök: det görs om med normal väntetid och räknas mot max_attempts. När mellanprogramvaran har registrerat gränsen hos arbetarens hastighetsbegränsare skickas senare jobb för hanteraren också tillbaka till pending innan de körs medan gränsen nåtts (se Hastighetsbegränsning).

use Queuety\Middleware\RateLimited;

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

Hastighetsbegränsaren använder den interna RateLimiter-instansen i Queuety, så gränserna gäller för alla arbetarprocesser.

Timeout

Startar en tidsgräns per jobb med pcntl_alarm och är avsedd att kasta ett TimeoutException om jobbet körs för länge. Tidsgränsen tillämpas inte för närvarande (#10): Queuety aktiverar inte asynkrona signaler, så alarmhanteraren körs inte medan jobbet körs. Mellanprogramvaran återställer också alarmet när den returnerar, vilket avbryter arbetarens eget alarm för jobbtidsgränsen.

use Queuety\Middleware\Timeout;

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

Kräver tillägget pcntl PHP. Om pcntl inte är tillgängligt gör mellanprogramvaran inget utan att meddela det (ingen tidsgräns tillämpas).

UniqueJob

Förhindrar samtidig körning av jobb med samma nyckel. Använder en låstabell i databasen för att säkerställa att endast en instans körs åt gången. Om ett lås redan finns hoppas jobbet över utan meddelande (det görs inget nytt försök).

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

Förhindrar överlappande körningar av jobb med samma logiska nyckel och låter låset upphöra automatiskt för att undvika lås som blir kvar efter att arbetare kraschat. Liknar UniqueJob, men är utformad för långvariga åtgärder där låset ska frigöras automatiskt efter en tidsgräns.

use Queuety\Middleware\WithoutOverlapping;

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

Om ett lås redan finns hoppas jobbet över utan meddelande. Utgångna lås rensas automatiskt före varje försök att skaffa ett lås.

Exempel: kombinera mellanprogramvaror

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

Det här jobbet kommer att:

1. Avbrytas om det körs längre än 5 minuter (Timeout)
2. Hoppas över om en annan import för samma leverantör redan körs (WithoutOverlapping)
3. Begränsas till 5 importer per minut för alla leverantörer (RateLimited)