Завдання

Завдання — основна одиниця роботи в 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 (черга, пріоритет, затримка, max_attempts, unique, after) працюють так само.

Як працює серіалізація

Коли ви диспетчеризуєте екземпляр Job, Queuety витягує повне ім’я класу з JobSerializer як обробник, а всі публічні властивості — як дані. На боці воркера серіалізатор відновлює екземпляр, зіставляючи ключі даних з іменами параметрів конструктора. Enum із базовим значенням серіалізуються в це значення та автоматично відновлюються.

Властивості завдання

Класи завдань можуть оголошувати публічні властивості для керування повторними спробами й тайм-аутами. Воркер читає їх за допомогою рефлексії та автоматично застосовує.

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)