Междинен софтуер

Междинният софтуер обгръща изпълнението на задачата в конвейер с последователни слоеве. Всеки междинен софтуер може да изпълнява логика преди и след задачата, да прекъсне изпълнението или да прихваща изключения. 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. Бъде ограничена до 5 импорта в минута за всички доставчици (RateLimited)