Úlohy

Úlohy jsou základní jednotkou práce v Queuety. Odešlete úlohu s názvem obslužné rutiny a payloadem a worker ji převezme a provede na pozadí.

Odesílání úloh

use Queuety\Queuety;

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

dispatch() vrací builder PendingJob. Než se úloha automaticky odešle, můžete zřetězit její nastavení:

use Queuety\Enums\Priority;

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

Chcete-li ihned získat ID úlohy, zavolejte ->id():

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

Psaní obslužných rutin

Obslužnou rutinou může být libovolná třída, která implementuje rozhraní 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',
        ];
    }
}

Metoda handle() přijímá payload úlohy a provádí práci. Metoda config() vrací volitelné výchozí hodnoty:

Tyto klíče zdrojů jsou součástí širšího modelu Správy zdrojů v Queuety, kde spolupracují metadata úloh, rozpočty workflow a přijímání úloh workery.

Zaregistrujte obslužnou rutinu:

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

Nebo použijte atributy PHP pro automatickou registraci.

Priorita

Úlohy mají čtyři úrovně priority. Workeři přebírají úlohy s vyšší prioritou jako první.

use Queuety\Enums\Priority;

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

Zpoždění

Odložte úlohu, aby byla dostupná až po uplynutí zadaného počtu sekund:

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

Úloha se vloží okamžitě, ale workeři ji nepřevezmou, dokud zpoždění neuplyne.

Jedinečné úlohy

Zabraňte opakovanému odeslání stejné obslužné rutiny se stejným payloadem. Pokud již existuje čekající nebo zpracovávaná úloha se stejnou obslužnou rutinou a payloadem, vrátí se její ID místo vytvoření nové úlohy.

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

Závislosti úloh

Úloha může záviset na jiné úloze. Worker ji nepřevezme, dokud závislá úloha nebude dokončena.

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

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

Odesílání dávek

Odešlete více úloh jediným příkazem INSERT s více řádky, a dosáhněte tak maximální propustnosti:

$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 obsahovat volitelné klíče: queue, priority, delay, max_attempts.

Opakované pokusy a prodlevy

Když úloha vyvolá výjimku, Queuety ji vrátí do pending s odloženým available_at, dokud nebude spuštěna celkem max_attemptskrát (počítá se i první spuštění, takže max_attempts( 3 ) znamená nejvýše dva opakované pokusy). Prodlevu před každým opakováním určuje strategie prodlev, kde attempt představuje číslo právě neúspěšného spuštění:

Nastavte globální výchozí hodnotu v wp-config.php:

define( 'QUEUETY_RETRY_BACKOFF', 'linear' );

Nebo ji nastavte pro každou obslužnou rutinu v config():

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

Nebo při odesílání:

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

Úlohy v karanténě

Po vyčerpání všech opakovaných pokusů se úloha označí jako odložená. Odložené úlohy zůstávají v databázi pro kontrolu, ale nezpracovávají se. Můžete je zkusit spustit ručně. Ruční opakování nastaví úlohu znovu jako pending, aniž by vynulovalo počet pokusů. Odložená úloha tedy dostane ještě jedno spuštění a při jeho neúspěchu se znovu odloží:

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

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

Nebo pomocí CLI:

wp queuety retry 42
wp queuety retry-buried

Časový limit

Queuety nastaví pcntl_alarm() na maximální dobu provádění a při jejím překročení má daný pokus ukončit pomocí TimeoutException. Tento limit se v současnosti nevynucuje (#10): worker nepovoluje asynchronní signály, takže dlouho běžící obslužná rutina doběhne do konce; obslužná rutina zablokovaná v sleep() se při spuštění alarmu předčasně ukončí a označí jako dokončená. Nastavte limit globálně:

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

Úlohy, které zůstávají ve stavu processing déle než QUEUETY_STALE_TIMEOUT sekund, se automaticky obnoví:

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

Obnovte zastaralé úlohy ručně:

wp queuety recover

Fronty

Úlohy se odesílají do pojmenovaných front. Každý worker zpracovává jednu frontu nebo uspořádaný seznam front (viz Workeři):

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

wp queuety work --queue=emails

Výchozí fronta je default. Pro oddělení zátěže můžete pro každou frontu spustit samostatné procesy workerů.

Omezování četnosti

Omezte, jak často může být obslužná rutina spuštěna v daném časovém okně:

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

Podrobnosti najdete v části Omezování četnosti.

Třídy úloh s vlastním odesláním

Queuety v0.6.0 představuje API úloh založený na třídách. Místo odděleného názvu obslužné rutiny, payloadu a třídy obslužné rutiny definujete jedinou třídu, která vše zapouzdřuje: payload (jako typované veřejné vlastnosti), logiku provádění a volitelný middleware.

Rozhraní Contracts\Job

use Queuety\Contracts\Job;

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

Třída Job uchovává payload ve veřejných vlastnostech a vlastní logiku provádí v handle(). Není potřeba samostatná metoda config() ani registrace obslužné rutiny.

Trait Dispatchable

Přidejte trait Dispatchable, abyste ve třídě úlohy získali statickou metodu 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 );
    }
}

Odešlete ji předáním argumentů konstruktoru:

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

Trait Dispatchable interně volá Queuety::dispatch(), takže všechny možnosti PendingJob (fronta, priorita, zpoždění, max_attempts, unique, after) fungují stejně.

Jak funguje serializace

Když odešlete instanci Job, Queuety pomocí JobSerializer získá plně kvalifikovaný název třídy jako obslužnou rutinu a všechny veřejné vlastnosti jako payload. Na straně workeru serializér zrekonstruuje instanci tak, že přiřadí klíče payloadu názvům parametrů konstruktoru. Enumy s podkladovou hodnotou se serializují na tuto hodnotu a automaticky se obnoví.

Vlastnosti úlohy

Třídy úloh mohou deklarovat veřejné vlastnosti, které řídí opakované pokusy a časové limity. Worker je načte pomocí reflexe a automaticky použije.

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 určuje postupně se prodlužující prodlevy mezi opakovanými pokusy. Například [10, 30, 60, 120] znamená, že první opakovaný pokus počká 10 sekund, druhý 30 sekund, třetí 60 sekund a každý další 120 sekund.

Vlastnosti související se zdroji jsou užitečné, když třídy úloh představují různé nároky na zdroje. Rychlé obnovení cache i náročné volání poskytovatele mohou být „jednou úlohou“, ale při dodržování rozpočtu workflow nebo rozhodování o přijímání úloh workery by neměly mít nutně stejnou váhu.

Úplný model včetně rozpočtů workflow a přijímání úloh workery najdete v části Správa zdrojů.

Neúspěšný hook

Když se úloha po vyčerpání všech opakovaných pokusů odloží, Queuety zavolá na instanci úlohy metodu failed(), pokud je definována. Použijte ji k úklidu, upozornění nebo zápisu do protokolu.

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

Metoda failed() přijímá výjimku, která způsobila konečné selhání. Instance úlohy se deserializuje s původním payloadem, takže jsou dostupné všechny veřejné vlastnosti.

Podmíněné odesílání

Trait Dispatchable poskytuje metody dispatch_if() a dispatch_unless() pro podmíněné odesílání. Obě přijímají jako první argument podmínku typu boolean, za níž následují argumenty konstruktoru ú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() odešle úlohu, pokud je podmínka true, a vrátí PendingJob. Pokud je podmínka false, vrátí null.

dispatch_unless() odešle úlohu, pokud je podmínka false, a vrátí PendingJob. Pokud je podmínka true, vrátí null.

Obě metody podporují při odeslání řetězení možností PendingJob:

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

Synchronní odeslání

Pomocí Queuety::dispatch_sync() provedete úlohu okamžitě v aktuálním procesu, aniž byste ji přidali do fronty. Hodí se to při testování a u úloh, které se musí spustit přímo.

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

Metoda úlohy handle() se zavolá přímo. Nevytvoří se žádný řádek v databázi, nepoužije se worker a middleware se neaplikuje.

Řetězení úloh

Řetězení úloh spouští posloupnost úloh jednu po druhé. Každá úloha v řetězci závisí na úspěšném dokončení předchozí úlohy. Pokud některá úloha selže, následující úlohy v řetězci zůstávají čekající.

Použití 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() vrací ChainBuilder s řetězenými metodami pro nastavení fronty a obsluhy selhání. Metoda dispatch() vytvoří všechny úlohy se sekvenčními vztahy depends_on a vrátí ID první úlohy.

Použití with_chain()

Trait Dispatchable poskytuje metodu with_chain(). V současnosti je ekvivalentní Queuety::chain( $jobs ): třída, na které je metoda zavolána, se do řetězce nepřidá (#13), proto do pole uveďte také první úlohu:

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

Zpracování selhání řetězce

Zaregistrujte třídu obsluhující selhání pomocí catch(). Třída obsluhy musí mít metodu handle(). Zavolá se, když je některá úloha v řetězci trvale přesunuta do stavu buried. Worker podporuje handle(), handle( Job $job ) a handle( Job $job, \Throwable $exception ).

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

Přidání middleware

Třídy úloh mohou definovat metodu middleware(), která vrací middleware obalující vykonání:

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 vestavěném middleware a tvorbě vlastního middleware najdete v části Middleware.

Porovnání úloh založených na obslužných třídách

Oba přístupy jsou plně podporovány. Existující úlohy založené na obslužných třídách fungují dál beze změn. V jednom projektu můžete oba styly kombinovat.

Životní cyklus úlohy

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