Задачи

Задачи — основная единица работы в Queuety. Отправьте задачу с именем обработчика и данными, и воркер получит её и выполнит в фоновом режиме.

Отправка задач

use Queuety\Queuety;

Queuety::dispatch( 'send_email', [ 'to' => 'user@example.com' ] );

dispatch() возвращает построитель PendingJob. Перед автоматической отправкой можно последовательно добавить параметры:

use Queuety\Enums\Priority;

Queuety::dispatch( 'send_email', [ 'to' => 'user@example.com' ] )
    ->on_queue( 'emails' )
    ->with_priority( Priority::High )
    ->delay( 300 )
    ->max_attempts( 5 );

Чтобы сразу получить ID задачи, вызовите ->id():

$job_id = Queuety::dispatch( 'send_email', $payload )->id();

Написание обработчиков

Обработчиком может быть любой класс, реализующий интерфейс Queuety\Handler:

use Queuety\Handler;

class SendEmailHandler implements Handler {
    public function handle( array $payload ): void {
        wp_mail( $payload['to'], $payload['subject'], $payload['body'] );
    }

    public function config(): array {
        return [
            'queue'           => 'emails',
            'max_attempts'    => 5,
            'backoff'         => 'exponential',
        ];
    }
}

Метод handle() получает данные задачи и выполняет работу. Метод config() возвращает необязательные значения по умолчанию:

Эти ключи ресурсов входят в общую модель Управления ресурсами в Queuety, где метаданные задач, бюджеты рабочего процесса и допуск воркеров работают вместе.

Зарегистрируйте обработчик:

Queuety::register( 'send_email', SendEmailHandler::class );

Или используйте атрибуты PHP для автоматической регистрации.

Приоритет

У задач есть четыре уровня приоритета. Воркеры в первую очередь получают задачи с более высоким приоритетом.

use Queuety\Enums\Priority;

Queuety::dispatch( 'handler', $payload )->with_priority( Priority::Urgent );

Задержка

Задайте задержку задачи, чтобы она стала доступна только по истечении указанного числа секунд:

Queuety::dispatch( 'send_reminder', $payload )->delay( 3600 ); // 1 hour

Задача добавляется сразу, но воркеры не получат её, пока не истечёт задержка.

Уникальные задачи

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

Queuety::dispatch( 'sync_inventory', [ 'sku' => 'ABC-123' ] )->unique();

Зависимости задач

Задача может зависеть от другой задачи. Воркер не получит её, пока не завершится задача-зависимость.

$first_id = Queuety::dispatch( 'import_data', $payload )->id();

Queuety::dispatch( 'send_notification', [ 'import_id' => $first_id ] )
    ->after( $first_id );

Пакетная отправка

Отправляйте несколько задач одним многострочным запросом INSERT для максимальной пропускной способности:

$job_ids = Queuety::batch( [
    [ 'handler' => 'send_email', 'payload' => [ 'to' => 'a@example.com' ] ],
    [ 'handler' => 'send_email', 'payload' => [ 'to' => 'b@example.com' ] ],
    [ 'handler' => 'send_email', 'payload' => [ 'to' => 'c@example.com' ] ],
] );

Каждый элемент может содержать необязательные ключи: queue, priority, delay, max_attempts.

Повторные попытки и задержка

При возникновении исключения в задаче Queuety возвращает её в pending, откладывая available_at до тех пор, пока общее число запусков не достигнет max_attempts (первый запуск учитывается, поэтому max_attempts( 3 ) означает не более двух повторных попыток). Задержка перед каждой повторной попыткой определяется стратегией задержки, где attempt — номер завершившегося с ошибкой запуска:

Задайте глобальное значение по умолчанию в wp-config.php:

define( 'QUEUETY_RETRY_BACKOFF', 'linear' );

Или задайте его отдельно для каждого обработчика в config():

public function config(): array {
    return [ 'backoff' => 'exponential', 'max_attempts' => 5 ];
}

Или задайте его при отправке:

Queuety::dispatch( 'handler', $payload )->max_attempts( 10 );

Отложенные задания

После исчерпания всех попыток задача помечается как архивная. Архивные задачи остаются в базе данных для проверки, но не обрабатываются. Их можно повторить вручную. При ручном повторе задача снова pending без сброса счётчика попыток, поэтому архивная задача получает ещё один запуск и снова перемещается в архив, если этот запуск завершится с ошибкой:

// Retry a specific buried job
Queuety::retry( $job_id );

// Retry all buried jobs
$count = Queuety::retry_buried();

Или через CLI:

wp queuety retry 42
wp queuety retry-buried

Тайм-аут

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

define( 'QUEUETY_MAX_EXECUTION_TIME', 300 ); // 5 minutes

Задачи, которые остаются в статусе processing дольше QUEUETY_STALE_TIMEOUT секунд, автоматически восстанавливаются:

define( 'QUEUETY_STALE_TIMEOUT', 600 ); // 10 minutes

Восстановите устаревшие задачи вручную:

wp queuety recover

Очереди

Задачи отправляются в именованные очереди. Каждый воркер обрабатывает одну очередь или упорядоченный список очередей (см. Воркеры):

Queuety::dispatch( 'handler', $payload )->on_queue( 'emails' );

wp queuety work --queue=emails

Очередь по умолчанию — default. Для изоляции нагрузок можно запустить отдельные процессы воркеров для каждой очереди.

Ограничение частоты запросов

Ограничьте частоту выполнения обработчика в заданном временном интервале:

Queuety::dispatch( 'call_api', $payload )
    ->rate_limit( 10, 60 ); // max 10 executions per 60 seconds

Подробности см. в разделе Ограничение частоты запросов.

Классы задач с поддержкой отправки

Queuety v0.6.0 добавляет основанный на классах механизм отправки API задач. Вместо того чтобы разносить имя обработчика, данные и класс обработчика, вы создаёте один класс, содержащий всё необходимое: данные (в виде типизированных публичных свойств), логику выполнения и необязательное промежуточное ПО.

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

use Queuety\Contracts\Job;

interface Job {
    public function handle(): void;
}

Класс Job хранит данные в виде публичных свойств и выполняет собственную логику в handle(). Отдельный метод config() или регистрация обработчика не требуются.

Трейт Dispatchable

Добавьте трейт Dispatchable, чтобы получить статический метод dispatch() в классе задачи:

use Queuety\Contracts\Job;
use Queuety\Dispatchable;

class SendEmailJob implements Job {
    use Dispatchable;

    public function __construct(
        public readonly string $to,
        public readonly string $subject,
        public readonly string $body,
    ) {}

    public function handle(): void {
        wp_mail( $this->to, $this->subject, $this->body );
    }
}

Отправьте задачу, передав аргументы конструктора:

SendEmailJob::dispatch( 'user@example.com', 'Welcome!', 'Hello there.' )
    ->on_queue( 'emails' )
    ->with_priority( Priority::High );

Трейт Dispatchable внутри вызывает Queuety::dispatch(), поэтому все параметры PendingJob (queue, priority, delay, max_attempts, unique, after) работают точно так же.

Как работает сериализация

При отправке экземпляра Job метод JobSerializer в Queuety извлекает полное имя класса в качестве обработчика и все публичные свойства в качестве данных. На стороне воркера сериализатор восстанавливает экземпляр, сопоставляя ключи данных с именами параметров конструктора. Перечисления с базовым типом сериализуются в базовое значение и автоматически восстанавливаются.

Свойства задания

Классы задач могут объявлять публичные свойства для настройки повторных попыток и тайм-аутов. Воркер считывает их с помощью рефлексии и применяет автоматически.

use Queuety\Contracts\Job;
use Queuety\Dispatchable;

class ImportProductsJob implements Job {
    use Dispatchable;

    public int $tries = 5;
    public int $timeout = 120;
    public array $backoff = [ 10, 30, 60, 120 ];
    public string $concurrency_group = 'imports';
    public int $concurrency_limit = 2;
    public int $cost_units = 4;

    public function __construct(
        public readonly int $supplier_id,
    ) {}

    public function handle(): void {
        // Import products...
    }
}

Массив $backoff задаёт увеличивающиеся задержки между повторными попытками. Например, [10, 30, 60, 120] означает, что первая повторная попытка начнётся через 10 секунд, вторая — через 30 секунд, третья — через 60 секунд, а все последующие — через 120 секунд.

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

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

Хук сбоя

Когда задача перемещается в архив после исчерпания всех попыток, Queuety вызывает метод failed() у экземпляра задачи, если он определён. Используйте его для очистки, отправки оповещений или ведения журнала.

use Queuety\Contracts\Job;
use Queuety\Dispatchable;

class ProcessPaymentJob implements Job {
    use Dispatchable;

    public int $tries = 3;

    public function __construct(
        public readonly int $order_id,
    ) {}

    public function handle(): void {
        // Process the payment...
    }

    public function failed( \Throwable $exception ): void {
        // Notify the team that payment processing failed permanently.
        error_log( "Payment failed for order {$this->order_id}: {$exception->getMessage()}" );
    }
}

Метод failed() получает исключение, вызвавшее окончательный сбой. Экземпляр задачи десериализуется с исходными данными, поэтому доступны все публичные свойства.

Условная отправка

Трейт Dispatchable предоставляет методы dispatch_if() и dispatch_unless() для условной отправки. Оба принимают логическое условие первым аргументом, а затем аргументы конструктора задачи.

// Only dispatch if the user has opted in to notifications.
SendEmailJob::dispatch_if( $user->wants_notifications, $user->email, 'Update', $body );

// Dispatch unless the feature flag is disabled.
SyncInventoryJob::dispatch_unless( $maintenance_mode, $sku );

dispatch_if() отправляет задачу, если условие равно true, и возвращает PendingJob. Если условие равно false, возвращает null.

dispatch_unless() отправляет задачу, если условие равно false, и возвращает PendingJob. Если условие равно true, возвращает null.

Оба метода поддерживают добавление параметров PendingJob при отправке:

SendEmailJob::dispatch_if( $should_send, $to, $subject, $body )
    ?->on_queue( 'emails' )
    ?->with_priority( Priority::High );

Синхронная отправка

Используйте Queuety::dispatch_sync(), чтобы сразу выполнить задачу в текущем процессе, не добавляя её в очередь. Это полезно при тестировании и для задач, которые должны выполняться непосредственно.

use Queuety\Queuety;

// With a Job instance
Queuety::dispatch_sync( new SendEmailJob( to: 'user@example.com', subject: 'Hello', body: 'Hi!' ) );

// With a handler name and payload
Queuety::dispatch_sync( 'send_email', [ 'to' => 'user@example.com' ] );

Метод handle() задачи вызывается напрямую. Строка в базе данных не создаётся, рабочий процесс не запускается, промежуточные обработчики не применяются.

Цепочки задач

Цепочка задач последовательно выполняет задачи одну за другой. Каждая задача в цепочке зависит от успешного завершения предыдущей. Если задача завершается с ошибкой, последующие задачи в цепочке остаются в ожидании.

Использование Queuety::chain()

use Queuety\Queuety;

$first_job_id = Queuety::chain( [
    new FetchDataJob( $url ),
    new ProcessDataJob(),
    new NotifyCompleteJob(),
] )
    ->on_queue( 'pipeline' )
    ->catch( ChainFailedHandler::class )
    ->dispatch();

Queuety::chain() возвращает ChainBuilder с цепочкой методов для настройки очереди и обработчика ошибок. Метод dispatch() создаёт все задачи с последовательными связями depends_on и возвращает ID первой задачи.

Использование with_chain()

Трейт Dispatchable предоставляет метод with_chain(). Сейчас он эквивалентен Queuety::chain( $jobs ): вызываемый класс не добавляется в цепочку (#13), поэтому укажите в массиве и первую задачу:

FetchDataJob::with_chain( [
    new FetchDataJob(),
    new ProcessDataJob(),
    new NotifyCompleteJob(),
] )
    ->on_queue( 'pipeline' )
    ->dispatch();

Обработка ошибок цепочки

Зарегистрируйте класс обработчика ошибок с помощью catch(). В классе обработчика должен быть метод handle(). Он вызывается, когда любая задача в цепочке окончательно перемещается в архив. Воркер поддерживает handle(), handle( Job $job ) и handle( Job $job, \Throwable $exception ).

class ChainFailedHandler {
    public function handle(): void {
        // Alert, clean up, etc.
    }
}

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

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

use Queuety\Contracts\Job;
use Queuety\Dispatchable;
use Queuety\Middleware\RateLimited;
use Queuety\Middleware\Timeout;

class CallOpenAIJob implements Job {
    use Dispatchable;

    public function __construct(
        public readonly string $prompt,
    ) {}

    public function handle(): void {
        // Call the API...
    }

    public function middleware(): array {
        return [
            new RateLimited( max: 60, window: 60 ),
            new Timeout( seconds: 120 ),
        ];
    }
}

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

Сравнение с задачами на основе обработчиков

Оба подхода полностью поддерживаются. Существующие задания на основе обработчиков продолжают работать без изменений. В одном проекте можно сочетать оба стиля.

Жизненный цикл задания

dispatch() -> pending -> processing -> completed
                  |           |
                  |           +-> exception -> pending (retry after backoff)
                  |                       |
                  |                       +-> buried (max attempts reached)
                  |
                  +-> delayed (available_at in the future)