Задачи

Задачите са основната единица работа в 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, уникалност, след) работят по същия начин.

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

Когато разпределите екземпляр на Job, Queuety извлича пълното име на класа като обработчик и всички публични свойства като полезен товар чрез JobSerializer. От страната на работника сериализаторът възстановява екземпляра, като съпоставя ключовете на полезния товар с имената на параметрите на конструктора. Енумите с базова стойност се сериализират до тази стойност и се възстановяват автоматично.

Свойства на задачите

Класовете на задачите могат да декларират публични свойства за управление на повторните опити и времевото ограничение. Работникът ги прочита чрез рефлексия и ги прилага автоматично.

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. Връща null, ако условието е false.

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

И двата метода поддържат свързване на опции 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() на задачата се извиква директно. Не се създава ред в базата данни, не се включва worker и не се прилага междинен софтуер.

Свързване на задачи

Свързването на задачи изпълнява поредица от задачи една след друга. Всяка задача във веригата зависи от успешното завършване на предишната. Ако някоя задача се провали, следващите задачи във веригата остават в изчакване.

Използване на 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)