Oprogramowanie pośredniczące

Oprogramowanie pośredniczące otacza wykonywanie zadań potokiem przypominającym cebulę. Każdy element może uruchamiać logikę przed zadaniem i po nim, przerywać wykonanie lub przechwytywać wyjątki. Queuety zawiera pięć wbudowanych elementów oprogramowania pośredniczącego i obsługuje własne elementy za pośrednictwem interfejsu Contracts\Middleware.

Oprogramowanie pośredniczące jest dostępne dla klas zadań , które można uruchomić (klas implementujących Contracts\Job z użyciem cechy Dispatchable).

Interfejs Contracts\Middleware

use Queuety\Contracts\Middleware;

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

Parametr $job to przetwarzana instancja zadania. Wywołaj $next( $job ), aby przekazać wykonanie do kolejnego elementu potoku (lub do metody handle() zadania, jeśli jest to ostatni element).

Dodawanie oprogramowania pośredniczącego do zadania

Zdefiniuj metodę middleware() w klasie zadania, która zwraca tablicę instancji oprogramowania pośredniczącego:

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

Kolejność wykonywania potoku

Oprogramowanie pośredniczące stosowane jest w kolejności tablicy, a pierwszy element stanowi jej najbardziej zewnętrzną warstwę. Jeśli metoda middleware() zwraca [A, B, C], kolejność wykonania jest następująca:

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

Każdy element otacza kolejny. A uruchamia się jako pierwszy przed zadaniem i jako ostatni po nim, dzięki czemu nadaje się do pomiaru czasu lub obsługi wyjątków za pomocą try/catch.

Pisanie własnego oprogramowania pośredniczącego

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

Użyj go w zadaniu:

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

Przerywanie wykonania

Element oprogramowania pośredniczącego może całkowicie pominąć zadanie, nie wywołując $next():

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

        $next( $job );
    }
}

Wbudowane oprogramowanie pośredniczące

ThrottlesExceptions

Ogranicza wyjątki z modułu obsługi zadań, aby zapobiegać lawinowemu wzrostowi liczby zadań. Jeśli w danym przedziale czasu moduł obsługi zgłosi zbyt wiele wyjątków, kolejne zadania dla tego modułu zostaną odrzucone za pomocą RuntimeException zamiast przetworzenia. Zapobiega to wyczerpaniu wszystkich prób ponowienia przez niedziałającą usługę zewnętrzną w wielu zadaniach naraz.

use Queuety\Middleware\ThrottlesExceptions;

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

Liczba wyjątków jest śledzona dla każdej klasy modułu obsługi w tabeli blokad Queuety i współdzielona przez wszystkie procesy robocze. Po osiągnięciu progu oprogramowanie pośredniczące zgłasza RuntimeException bez wywoływania zadania, a proces roboczy traktuje to jak każde inne niepowodzenie: próba jest wliczana do max_attempts, zadanie ponawia się ze standardowym opóźnieniem, a po wyczerpaniu prób trafia do archiwum zadań.

Przykład: jeśli zewnętrzne API nie działa, pierwsze 10 niepowodzeń jest obsługiwanych normalnie (każde zwiększa licznik wyjątków). Kolejne próby kończą się szybko, bez wywoływania API, dopóki przez 10 minut nie wystąpi nowy wyjątek; okno rozpoczyna się od nowa po każdym zarejestrowanym wyjątku.

RateLimited

Wymusza maksymalną liczbę wykonań w danym przedziale czasu. Jeśli limit zostanie przekroczony w oprogramowaniu pośredniczącym, zgłaszany jest RateLimitExceededException; proces roboczy traktuje go jako zwykłą nieudaną próbę, ponawianą ze standardowym opóźnieniem i wliczaną do max_attempts. Gdy oprogramowanie pośredniczące zarejestruje limit w ograniczniku liczby żądań procesu roboczego, kolejne zadania tego modułu obsługi zostaną odłożone z powrotem do pending przed uruchomieniem, gdy limit będzie osiągnięty (zob. Ograniczanie liczby żądań).

use Queuety\Middleware\RateLimited;

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

Ogranicznik liczby żądań korzysta z wewnętrznej instancji RateLimiter należącej do Queuety, więc limity są egzekwowane we wszystkich procesach roboczych.

Timeout

Ustawia limit czasu wykonywania zadania za pomocą pcntl_alarm i ma zgłaszać TimeoutException, gdy zadanie trwa zbyt długo. Obecnie limit nie jest egzekwowany (#10): Queuety nie włącza sygnałów asynchronicznych, więc procedura obsługi alarmu nie działa podczas wykonywania zadania. Oprogramowanie pośredniczące resetuje też alarm po zakończeniu działania, co anuluje własny alarm limitu czasu zadania procesu roboczego.

use Queuety\Middleware\Timeout;

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

Wymaga rozszerzenia pcntl PHP. Jeśli pcntl jest niedostępne, oprogramowanie pośredniczące pomija ten etap bez komunikatu (limit czasu nie jest egzekwowany).

UniqueJob

Zapobiega równoczesnemu wykonywaniu zadań z tym samym kluczem. Korzysta z tabeli blokad bazy danych, aby zapewnić, że jednocześnie działa tylko jedna instancja. Jeśli blokada jest już zajęta, zadanie jest po cichu pomijane (bez ponawiania).

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

Zapobiega nakładającemu się wykonywaniu zadań współdzielących klucz logiczny; automatycznie wygasa blokada, aby uniknąć pozostawienia jej po awarii procesu roboczego. Działa podobnie jak UniqueJob, ale jest przeznaczone do długotrwałych operacji, w których blokada powinna zostać automatycznie zwolniona po upływie określonego czasu.

use Queuety\Middleware\WithoutOverlapping;

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

Jeśli blokada jest już zajęta, zadanie jest po cichu pomijane. Wygasłe blokady są automatycznie czyszczone przed każdą próbą ich uzyskania.

Przykład: łączenie oprogramowania pośredniczącego

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

To zadanie będzie:

1. Przerywane, jeśli będzie działać dłużej niż 5 minut (Timeout)
2. Pomijane, jeśli trwa już inny import od tego samego dostawcy (WithoutOverlapping)
3. Ograniczone do 5 importów na minutę dla wszystkich dostawców (RateLimited)