Middleware

Middleware umschließt die Job-Ausführung in einer Pipeline nach dem Zwiebelprinzip. Jede Middleware kann vor und nach dem Job Logik ausführen, die Ausführung vorzeitig beenden oder Ausnahmen abfangen. Queuety liefert fünf integrierte Middleware-Komponenten und unterstützt eigene Middleware über die Schnittstelle Contracts\Middleware.

Middleware ist für dispatchbare Job-Klassen verfügbar (Klassen, die Contracts\Job mit dem Trait Dispatchable implementieren).

Die Schnittstelle Contracts\Middleware

use Queuety\Contracts\Middleware;

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

Der Parameter $job ist die Job-Instanz, die verarbeitet wird. Rufe $next( $job ) auf, um die Ausführung an die nächste Middleware in der Pipeline weiterzugeben (oder an die Methode handle() des Jobs, wenn dies die letzte Middleware ist).

Middleware zu einem Job hinzufügen

Definiere in deiner Job-Klasse eine Methode middleware(), die ein Array mit Middleware-Instanzen zurückgibt:

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

Reihenfolge der Pipeline-Ausführung

Middleware werden in der Reihenfolge des Arrays angewendet, wobei die erste Middleware die äußerste Ebene bildet. Wenn deine Methode middleware() [A, B, C] zurückgibt, ist die Ausführungsreihenfolge:

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

Jede Middleware umschließt die nächste. A wird vor dem Job zuerst und danach zuletzt ausgeführt und eignet sich daher ideal für Zeitmessungen oder Try-Catch-Wrapper.

Eigene Middleware schreiben

Implementiere die Schnittstelle 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" );
        }
    }
}

Verwende sie für einen Job:

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

Ausführung vorzeitig beenden

Eine Middleware kann den Job vollständig überspringen, indem sie $next() nicht aufruft:

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

        $next( $job );
    }
}

Integrierte Middleware

ThrottlesExceptions

Begrenzt Ausnahmen eines Job-Handlers, um eine Flut von Jobs zu verhindern. Hat ein Handler innerhalb eines Zeitfensters zu viele Ausnahmen ausgelöst, werden nachfolgende Jobs für diesen Handler mit einem RuntimeException abgelehnt, statt verarbeitet zu werden. So wird verhindert, dass ein ausgefallener externer Dienst alle Wiederholungsversuche für viele Jobs auf einmal aufbraucht.

use Queuety\Middleware\ThrottlesExceptions;

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

Ausnahmen werden pro Handler-Klasse in der Sperrtabelle von Queuety erfasst und von allen Worker-Prozessen gemeinsam genutzt. Ist der Grenzwert erreicht, löst die Middleware ein RuntimeException aus, ohne den Job aufzurufen. Der Worker behandelt dies wie jeden anderen Fehler: Der Versuch zählt für max_attempts, der Job wird mit dem üblichen Backoff erneut ausgeführt und nach Aufbrauchen aller Versuche in den Ablagebereich verschoben.

Beispiel: Ist ein externer API nicht verfügbar, werden die ersten 10 Fehler normal verarbeitet (bei jedem wird der Ausnahmezähler erhöht). Weitere Versuche schlagen sofort fehl, ohne API aufzurufen, bis 10 Minuten lang keine neue Ausnahme auftritt. Bei jeder erfassten Ausnahme beginnt das Zeitfenster erneut.

RateLimited

Erzwingt eine maximale Anzahl von Ausführungen innerhalb eines Zeitfensters. Wird das Limit innerhalb der Middleware überschritten, wird ein RateLimitExceededException ausgelöst. Der Worker behandelt es wie einen gewöhnlichen fehlgeschlagenen Versuch, wiederholt ihn mit dem üblichen Backoff und zählt ihn für max_attempts. Nachdem die Middleware das Limit beim Ratenbegrenzer des Workers registriert hat, werden spätere Jobs des Handlers bei erreichtem Limit vor der Ausführung wieder in die Warteschlange pending gestellt (siehe Ratenbegrenzung).

use Queuety\Middleware\RateLimited;

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

Der Ratenbegrenzer verwendet die interne RateLimiter-Instanz von Queuety. Dadurch gelten die Limits prozessübergreifend für alle Worker.

Timeout

Richtet mit pcntl_alarm ein Zeitlimit für die Job-Ausführung ein und soll ein TimeoutException auslösen, wenn der Job zu lange läuft. Das Zeitlimit wird derzeit nicht durchgesetzt (#10): Queuety aktiviert keine asynchronen Signale, daher wird der Alarm-Handler während der Job-Ausführung nicht ausgeführt. Außerdem setzt die Middleware den Alarm beim Beenden zurück und bricht damit den eigenen Job-Zeitlimit-Alarm des Workers ab.

use Queuety\Middleware\Timeout;

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

Erfordert die Erweiterung pcntl PHP. Ist pcntl nicht verfügbar, wird die Middleware stillschweigend übersprungen (es gilt kein Zeitlimit).

UniqueJob

Verhindert die gleichzeitige Ausführung von Jobs mit demselben Schlüssel. Eine Datenbanksperrtabelle stellt sicher, dass jeweils nur eine Instanz läuft. Ist die Sperre bereits belegt, wird der Job stillschweigend übersprungen (keine Wiederholung).

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

Verhindert überlappende Ausführungen von Jobs mit demselben logischen Schlüssel. Eine automatische Sperrfrist verhindert dauerhaft bestehende Sperren durch abgestürzte Worker. Ähnlich wie UniqueJob, aber für lang laufende Vorgänge ausgelegt, bei denen die Sperre nach einem Zeitlimit automatisch freigegeben werden soll.

use Queuety\Middleware\WithoutOverlapping;

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

Ist die Sperre bereits belegt, wird der Job stillschweigend übersprungen. Abgelaufene Sperren werden vor jedem Sperrversuch automatisch bereinigt.

Beispiel: Middleware kombinieren

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

Dieser Job wird:

1. beendet, wenn er länger als 5 Minuten läuft (Timeout)
2. übersprungen, wenn bereits ein Import desselben Lieferanten läuft (WithoutOverlapping)
3. auf 5 Importe pro Minute über alle Lieferanten hinweg begrenzt (RateLimited)