Úlohy

Úlohy sú základnou pracovnou jednotkou v Queuety. Zaraďte úlohu s názvom obslužnej triedy a dátami; pracovník ju prevezme a vykoná na pozadí.

Zaraďovanie úloh

use Queuety\Queuety;

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

dispatch() vráti zostavovač PendingJob. Pred automatickým zaradením môžete reťaziť jeho možnosti:

use Queuety\Enums\Priority;

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

Ak chcete okamžite získať ID úlohy, zavolajte ->id():

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

Písanie obslužných tried

Obslužnou triedou môže byť ľubovoľná trieda, ktorá implementuje rozhranie 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',
        ];
    }
}

Metóda handle() prijíma dáta úlohy a vykonáva prácu. Metóda config() vracia voliteľné predvolené hodnoty:

Tieto kľúče zdrojov patria do širšieho modelu QueuetySprávy zdrojov, v ktorom spolupracujú metadáta úloh, limity pracovných postupov a prijímanie úloh pracovníkmi.

Zaregistrujte obslužnú triedu:

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

Alebo na automatickú registráciu použite atribúty PHP.

Priorita

Úlohy majú štyri úrovne priority. Pracovníci najskôr preberajú úlohy s vyššou prioritou.

use Queuety\Enums\Priority;

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

Odloženie

Odložte úlohu, aby bola dostupná až po uplynutí zadaného počtu sekúnd:

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

Úloha sa vloží okamžite, no pracovníci ju prevezmú až po uplynutí odkladu.

Jedinečné úlohy

Zabráňte opakovanému zaradeniu rovnakej obslužnej triedy s rovnakými dátami. Ak už existuje čakajúca alebo spracúvaná úloha s rovnakou obslužnou triedou a dátami, namiesto vytvorenia novej sa vráti ID existujúcej úlohy.

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

Závislosti úloh

Úloha môže závisieť od inej úlohy. Pracovník ju neprevezme, kým sa závislá úloha nedokončí.

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

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

Hromadné zaraďovanie

Pre maximálnu priepustnosť zaraďte viacero úloh jediným viacriadkovým príkazom 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' ] ],
] );

Každá položka môže obsahovať voliteľné kľúče: queue, priority, delay, max_attempts.

Opakovanie a odstup

Keď úloha vyvolá výnimku, Queuety ju vráti do pending s odloženým available_at, kým sa nespustí celkovo max_attempts-krát (počíta sa aj prvé spustenie, takže max_attempts( 3 ) znamená najviac dve opakovania). Odstup pred každým opakovaním určuje stratégia odstupu, pričom attempt označuje číslo práve neúspešného spustenia:

Nastavte globálnu predvolenú hodnotu v wp-config.php:

define( 'QUEUETY_RETRY_BACKOFF', 'linear' );

Alebo ju nastavte pre konkrétnu obslužnú triedu v config():

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

Alebo pri zaraďovaní:

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

Neúspešné úlohy

Po vyčerpaní všetkých pokusov o opakovanie sa úloha označí ako odložená. Takéto úlohy zostávajú v databáze na kontrolu, ale nespracúvajú sa. Môžete ich zopakovať manuálne. Manuálne opakovanie vráti úlohu do stavu pending bez vynulovania počtu pokusov, takže odložená úloha sa spustí ešte raz a pri neúspechu sa znova odloží:

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

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

Alebo cez CLI:

wp queuety retry 42
wp queuety retry-buried

Časový limit

Queuety nastaví pcntl_alarm() na maximálny čas vykonávania a po jeho prekročení má pokus zlyhať s TimeoutException. Tento limit sa momentálne nevynucuje (#10): pracovník nepovoľuje asynchrónne signály, preto dlho bežiaca obslužná trieda dokončí vykonávanie a obslužná trieda zablokovaná v sleep() sa po zaznení alarmu preruší a označí za dokončenú. Globálny limit nastavíte takto:

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

Úlohy, ktoré zostanú v stave processing dlhšie než QUEUETY_STALE_TIMEOUT sekúnd, sa automaticky obnovia:

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

Zastarane úlohy obnovte ručne:

wp queuety recover

Fronty

Úlohy sa zaraďujú do pomenovaných frontov. Každý pracovník spracúva jeden front alebo usporiadaný zoznam frontov (pozrite časť Pracovníci):

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

wp queuety work --queue=emails

Predvolený front je default. Pre oddelenie pracovných záťaží môžete spustiť samostatné procesy pracovníkov pre každý front.

Obmedzenie frekvencie

Obmedzte, ako často sa môže obslužná trieda spustiť v rámci časového okna:

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

Podrobnosti nájdete v časti Obmedzenie frekvencie.

Triedy úloh na zaraďovanie

Queuety v0.6.0 prináša objektový API úloh. Namiesto oddelenia názvu obslužnej triedy, dát a jej triedy definujete jednu triedu, ktorá všetko zapuzdrí: dáta (ako typované verejné vlastnosti), logiku vykonania a voliteľný middleware.

Rozhranie Contracts\Job

use Queuety\Contracts\Job;

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

Trieda Job uchováva dáta vo verejných vlastnostiach a vykonáva vlastnú logiku v handle(). Nie je potrebná samostatná metóda config() ani registrácia obslužnej triedy.

Trait Dispatchable

Pridajte trait Dispatchable a získajte statickú metódu dispatch() vo svojej triede úlohy:

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

Zaraďte ju odovzdaním argumentov konštruktora:

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

Trait Dispatchable interne volá Queuety::dispatch(), takže všetky možnosti PendingJob (front, priorita, odklad, max_attempts, unique, after) fungujú rovnako.

Ako funguje serializácia

Pri zaradení inštancie Job metóda JobSerializer z Queuety získa plne kvalifikovaný názov triedy ako obslužnú triedu a všetky verejné vlastnosti ako dáta. Na strane pracovníka serializátor obnoví inštanciu priradením kľúčov dát k názvom parametrov konštruktora. Enumy s podkladovou hodnotou sa serializujú na túto hodnotu a automaticky sa obnovia.

Vlastnosti úloh

Triedy úloh môžu deklarovať verejné vlastnosti, ktorými určujú správanie pri opakovaní a časovom limite. Pracovník ich načíta pomocou reflexie a automaticky uplatní.

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...
    }
}

Pole $backoff poskytuje postupne sa predlžujúce odstupy pri opakovaní. Napríklad [10, 30, 60, 120] znamená, že prvé opakovanie počká 10 sekúnd, druhé 30 sekúnd, tretie 60 sekúnd a každé ďalšie 120 sekúnd.

Vlastnosti súvisiace so zdrojmi sú užitočné, keď triedy úloh predstavujú rôznu náročnosť. Rýchla aktualizácia vyrovnávacej pamäte aj rozsiahle volanie poskytovateľa môžu byť „jednou úlohou“, no pri rozhodovaní, či pracovný postup alebo prijatie úlohy pracovníkom zostane v bezpečných medziach, by sa nemali nevyhnutne počítať rovnako.

Celý model vrátane limitov pracovných postupov a prijímania úloh na strane pracovníkov nájdete v časti Správa zdrojov.

Háčik pri zlyhaní

Keď sa úloha po vyčerpaní všetkých pokusov odloží, Queuety zavolá metódu failed() na inštancii úlohy, ak je definovaná. Využite ju na čistenie, upozornenia alebo zaznamenávanie udalostí.

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()}" );
    }
}

Metóda failed() prijíma výnimku, ktorá spôsobila konečné zlyhanie. Inštancia úlohy sa deserializuje s pôvodnými dátami, takže sú dostupné všetky verejné vlastnosti.

Podmienené zaraďovanie

Trait Dispatchable poskytuje metódy dispatch_if() a dispatch_unless() na podmienené zaraďovanie. Obe prijímajú najprv logickú podmienku a za ňou argumenty konštruktora úlohy.

// 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() zaradí úlohu, keď je podmienka true, a vráti PendingJob. Keď podmienka false, vráti null.

dispatch_unless() zaradí úlohu, keď je podmienka false, a vráti PendingJob. Keď podmienka true, vráti null.

Obe metódy pri zaraďovaní podporujú reťazenie možností PendingJob:

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

Synchrónne zaraďovanie

Použite Queuety::dispatch_sync() na okamžité vykonanie úlohy v aktuálnom procese bez jej vloženia do frontu. Hodí sa na testovanie a úlohy, ktoré sa musia vykonať priamo.

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' ] );

Metóda handle() úlohy sa zavolá priamo. Nevytvorí sa žiadny riadok v databáze, nepoužije sa pracovník a neuplatní sa middleware.

Reťazenie úloh

Pri reťazení sa úlohy vykonávajú postupne, jedna po druhej. Každá úloha v reťazci závisí od úspešného dokončenia predchádzajúcej. Ak niektorá zlyhá, nasledujúce úlohy zostanú čakať.

Použitie 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() vráti ChainBuilder s plynulými metódami na nastavenie frontu a obslužnej triedy pre zlyhanie. Metóda dispatch() vytvorí všetky úlohy so sekvenčnými vzťahmi depends_on a vráti ID prvej úlohy.

Použitie with_chain()

Trait Dispatchable poskytuje metódu with_chain(). Momentálne je ekvivalentná metóde Queuety::chain( $jobs ): trieda, na ktorej sa volá, sa do reťazca nepridá (#13), preto do poľa uveďte aj prvú úlohu:

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

Spracovanie zlyhania reťazca

Zaregistrujte obslužnú triedu pre zlyhanie pomocou catch(). Táto trieda musí mať metódu handle(). Zavolá sa, keď sa niektorá úloha v reťazci natrvalo odloží. Pracovník podporuje handle(), handle( Job $job ) a handle( Job $job, \Throwable $exception ).

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

Pridanie middleware

Triedy úloh môžu definovať metódu middleware(), ktorá vráti middleware obalujúci vykonanie:

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

Podrobnosti o vstavanom middleware a písaní vlastného nájdete v časti Middleware.

Porovnanie úloh s obslužnou triedou

Oba prístupy sú plne podporované. Existujúce úlohy s obslužnými triedami fungujú ďalej bez úprav. V jednom projekte môžete kombinovať oba štýly.

Životný cyklus úlohy

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