Промежуточное ПО

Промежуточное ПО оборачивает выполнение задания в конвейер с последовательной обработкой. Каждый компонент может выполнять логику до и после задания, останавливать выполнение или перехватывать исключения. Queuety включает пять встроенных компонентов промежуточного ПО и поддерживает пользовательские компоненты через интерфейс Contracts\Middleware.

Промежуточное ПО доступно для классов заданий с возможностью диспетчеризации (классов, реализующих Contracts\Job с трейтом Dispatchable).

Интерфейс Contracts\Middleware

use Queuety\Contracts\Middleware;

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

Параметр $job — экземпляр обрабатываемого задания. Вызовите $next( $job ), чтобы передать выполнение следующему компоненту в конвейере (или методу handle() задания, если это последний компонент).

Добавление промежуточного ПО к заданию

Определите метод middleware() в классе задания, возвращающий массив экземпляров промежуточного ПО:

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

Порядок выполнения конвейера

Компоненты промежуточного ПО применяются в порядке массива; первый компонент образует внешний слой. Если метод middleware() возвращает [A, B, C], порядок выполнения будет таким:

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

Каждый компонент оборачивает следующий. A выполняется первым перед заданием и последним после него, поэтому подходит для замеров времени или обёрток try/catch.

Создание пользовательского промежуточного ПО

Реализуйте интерфейс 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" );
        }
    }
}

Используйте его в задании:

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

Прерывание выполнения

Компонент промежуточного ПО может полностью пропустить выполнение задания, если не вызвать $next():

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

        $next( $job );
    }
}

Встроенное промежуточное ПО

ThrottlesExceptions

Ограничивает исключения, возникающие в обработчике задания, чтобы предотвратить массовые сбои заданий. Если обработчик выбросил слишком много исключений за заданный период, последующие задания для этого обработчика отклоняются с RuntimeException вместо выполнения. Это не позволяет сбоям внешней службы израсходовать все попытки повтора сразу для множества заданий.

use Queuety\Middleware\ThrottlesExceptions;

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

Счётчики исключений ведутся отдельно для каждого класса обработчика в таблице блокировок Queuety и доступны всем рабочим процессам. При достижении порога компонент выбрасывает RuntimeException, не вызывая задание, а рабочий процесс обрабатывает его как любую другую ошибку: попытка учитывается в max_attempts, задание повторяется с обычной задержкой и перемещается в архив после исчерпания попыток.

Пример: если внешняя служба API недоступна, первые 10 ошибок обрабатываются обычным образом (каждая увеличивает счётчик исключений). Последующие попытки быстро завершаются с ошибкой, не вызывая API, пока не пройдут 10 минут без нового исключения; при каждом зарегистрированном исключении отсчёт окна начинается заново.

RateLimited

Ограничивает число выполнений за заданный период. Если ограничение превышено внутри промежуточного ПО, выбрасывается RateLimitExceededException; рабочий процесс обрабатывает его как обычную неудачную попытку, повторяет задание с обычной задержкой и учитывает попытку в max_attempts. После регистрации ограничения в ограничителе частоты рабочего процесса последующие задания этого обработчика также возвращаются в pending до запуска, пока ограничение действует (см. раздел Ограничение частоты).

use Queuety\Middleware\RateLimited;

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

Ограничитель частоты использует внутренний экземпляр Queuety RateLimiter, поэтому ограничения действуют для всех рабочих процессов.

Timeout

Устанавливает тайм-аут выполнения отдельного задания с помощью pcntl_alarm и предназначен для выбрасывания TimeoutException, если задание выполняется слишком долго. Сейчас тайм-аут не применяется (#10): Queuety не включает асинхронные сигналы, поэтому обработчик сигнала тревоги не запускается во время выполнения задания. Кроме того, по завершении компонент сбрасывает сигнал тревоги, отменяя собственный сигнал тайм-аута задания рабочего процесса.

use Queuety\Middleware\Timeout;

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

Требуется расширение pcntl PHP. Если pcntl недоступно, промежуточное ПО молча продолжает выполнение (тайм-аут не применяется).

UniqueJob

Запрещает одновременное выполнение заданий с одинаковым ключом. Использует таблицу блокировок базы данных, чтобы одновременно выполнялся только один экземпляр. Если блокировка уже установлена, задание молча пропускается (без повтора).

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

Не допускает одновременного выполнения заданий с одинаковым логическим ключом и автоматически снимает блокировку, чтобы избежать зависших блокировок при сбое рабочего процесса. Похоже на UniqueJob, но предназначено для длительных операций, где блокировка должна сниматься автоматически по тайм-ауту.

use Queuety\Middleware\WithoutOverlapping;

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

Если блокировка уже установлена, задание молча пропускается. Просроченные блокировки автоматически очищаются перед каждой попыткой установки блокировки.

Пример: сочетание компонентов промежуточного ПО

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

Это задание будет:

1. Принудительно завершено, если работает дольше 5 минут (Timeout)
2. Пропущено, если уже выполняется другой импорт от того же поставщика (WithoutOverlapping)
3. Ограничено пятью импортами в минуту от всех поставщиков (RateLimited)