Middleware

A middleware hagymahéjszerű folyamatba csomagolja a feladat végrehajtását. Minden middleware a feladat előtt és után is futtathat logikát, megszakíthatja a végrehajtást, vagy elkaphatja a kivételeket. A Queuety öt beépített middleware-t tartalmaz, és az Contracts\Middleware interfészen keresztül támogatja az egyéni middleware-eket.

A middleware használható diszpécselhető feladatosztályokhoz (olyan osztályokhoz, amelyek megvalósítják a Contracts\Job elemet a Dispatchable trait használatával).

A Contracts\Middleware interfész

use Queuety\Contracts\Middleware;

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

A $job paraméter a feldolgozás alatt álló feladatpéldány. Hívja meg a $next( $job ) függvényt, hogy a végrehajtás a folyamat következő middleware-ére kerüljön (vagy ha ez az utolsó middleware, a feladat handle() metódusára).

Middleware hozzáadása egy feladathoz

Definiáljon egy middleware() metódust a feladatosztályban, amely middleware-példányok tömbjét adja vissza:

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

A folyamat végrehajtási sorrendje

A middleware-ek a tömb sorrendjében lépnek működésbe, az első middleware a legkülső réteg. Ha az middleware() metódus a következőt adja vissza: [A, B, C], a végrehajtás sorrendje a következő:

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

Minden middleware a következőt burkolja be. A A fut le elsőként a feladat előtt, és utolsóként utána, így ideális időmérésre vagy try/catch burkolóként.

Egyéni middleware írása

Implementálja a Contracts\Middleware interfészt:

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

Használja egy feladatnál:

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

A végrehajtás kihagyása

A middleware teljesen kihagyhatja a feladatot, ha nem hívja meg a $next() függvényt:

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

        $next( $job );
    }
}

Beépített köztesrétegek

ThrottlesExceptions

Korlátozza a feladatkezelő által dobott kivételeket, hogy megelőzze a feladatrohamokat. Ha egy kezelő túl sok kivételt dob egy időablakon belül, az adott kezelő további feladatait a rendszer feldolgozás helyett RuntimeException értékkel utasítja el. Ez megakadályozza, hogy egy hibás külső szolgáltatás egyszerre több feladat összes újrapróbálási kísérletét felhasználja.

use Queuety\Middleware\ThrottlesExceptions;

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

A kivételek számát kezelőosztályonként a Queuety zárolási táblája követi, és az összes munkafolyamat között megosztott. A küszöb elérésekor a middleware RuntimeException értéket dob a feladat meghívása nélkül, a munkafolyamat pedig ezt a többi hibához hasonlóan kezeli: a próbálkozás beleszámít a max_attempts értékébe, a feladat normál várakozással újrapróbálkozik, majd az összes próbálkozás kimerülésekor a sikertelen feladatok közé kerül.

Példa: ha egy külső API nem működik, az első 10 hibát a rendszer normál módon dolgozza fel (mindegyik növeli a kivételszámlálót). A további próbálkozások az API meghívása nélkül azonnal meghiúsulnak, amíg 10 percig nem történik újabb kivétel; az időablak minden rögzített kivételnél újraindul.

RateLimited

Egy időablakon belül korlátozza a végrehajtások maximális számát. Ha a korlátot a middleware-ben túllépik, a rendszer RateLimitExceededException értéket dob; a munkafolyamat ezt normál sikertelen próbálkozásként kezeli, normál várakozással újrapróbálja, és beszámítja a max_attempts értékébe. Miután a middleware regisztrálta a korlátot a munkafolyamat sebességkorlátozójánál, a kezelő későbbi feladatait is visszahelyezi a pending értékre futtatás előtt, amíg a korlát fennáll (lásd: Sebességkorlátozás).

use Queuety\Middleware\RateLimited;

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

A sebességkorlátozó a Queuety belső RateLimiter példányát használja, ezért a korlátok az összes munkafolyamatra érvényesek.

Timeout

pcntl_alarm-val feladatvégrehajtási időkorlátot állít be, és a feladat túl hosszú futásakor TimeoutException kivételt kellene dobnia. Az időkorlát jelenleg nincs érvényesítve (#10): a Queuety nem engedélyez aszinkron jelzéseket, ezért a riasztáskezelő nem fut le a feladat végrehajtása közben. A middleware visszatéréskor a riasztást is visszaállítja, ami törli a munkafolyamat saját feladat-időkorlátját.

use Queuety\Middleware\Timeout;

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

A pcntl PHP bővítmény szükséges hozzá. Ha a pcntl nem érhető el, a middleware csendben kihagyja ezt a részt (nem érvényesít időkorlátot).

UniqueJob

Megakadályozza az azonos kulccsal rendelkező feladatok párhuzamos végrehajtását. Adatbázis-zárolási táblát használ annak biztosítására, hogy egyszerre csak egy példány fusson. Ha egy zárolás már foglalt, a feladat csendben kimarad (újrapróbálás nélkül).

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

Megakadályozza az azonos logikai kulccsal rendelkező feladatok átfedő végrehajtását, és automatikusan lejáró zárolást használ, hogy az összeomlott munkafolyamatok ne hagyjanak hátra zárolást. Hasonló a UniqueJob megoldáshoz, de olyan hosszan futó műveletekhez készült, ahol a zárolásnak időkorlát után automatikusan fel kell oldódnia.

use Queuety\Middleware\WithoutOverlapping;

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

Ha egy zárolás már foglalt, a feladat csendben kimarad. A lejárt zárolások minden új zárolási kísérlet előtt automatikusan törlődnek.

Példa: middleware-ek kombinálása

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

Ez a feladat:

1. Leáll, ha 5 percnél tovább fut (Timeout)
2. Kimarad, ha már fut egy másik import ugyanattól a beszállítótól (WithoutOverlapping)
3. Percenként legfeljebb 5 importot hajt végre az összes beszállítótól (RateLimited)