Užduotys

Užduotys yra pagrindinis darbo vienetas Queuety. Išsiųskite užduotį nurodydami apdorojimo metodą ir duomenis, o darbininkas ją paims ir atliks fone.

Užduočių siuntimas

use Queuety\Queuety;

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

dispatch() grąžina PendingJob konfigūravimo objektą. Prieš automatiškai išsiunčiant užduotį, galite susieti papildomas parinktis:

use Queuety\Enums\Priority;

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

Jei norite iškart gauti užduoties ID, iškvieskite ->id():

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

Apdorojimo metodų kūrimas

Apdorojimo metodas gali būti bet kuri klasė, įgyvendinanti sąsają 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',
        ];
    }
}

Metodas handle() gauna užduoties duomenis ir atlieka darbą. Metodas config() grąžina pasirinktines numatytąsias reikšmes:

Šie išteklių raktai yra platesnio Queuety modelio „Išteklių valdymas“ dalis. Jame kartu veikia užduočių metaduomenys, darbo eigos kvotos ir darbininkų priėmimas.

Užregistruokite apdorojimo metodą:

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

Arba naudokite atributus PHP, kad jie būtų užregistruoti automatiškai.

Prioritetas

Užduotys turi keturis prioriteto lygius. Darbininkai pirmiausia paima aukštesnio prioriteto užduotis.

use Queuety\Enums\Priority;

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

Delsa

Atidėkite užduotį, kad ji būtų pasiekiama tik praėjus nurodytam sekundžių skaičiui:

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

Užduotis įterpiama iškart, tačiau darbininkai jos nepaims, kol nepraeis delsos laikas.

Unikalios užduotys

Neleiskite pakartotinai siųsti tos pačios apdorojimo funkcijos ir duomenų užduočių. Jei tokia užduotis jau laukia arba yra apdorojama, grąžinamas esamos užduoties ID, o nauja nekuriama.

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

Užduočių priklausomybės

Užduotis gali priklausyti nuo kitos užduoties. Darbininkas jos nepaims, kol priklausoma užduotis nebus užbaigta.

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

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

Paketinis siuntimas

Vienu kelių eilučių INSERT sakiniu išsiųskite kelias užduotis, kad našumas būtų kuo didesnis:

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

Kiekvienam elementui galima nurodyti pasirinktinius raktus: queue, priority, delay, max_attempts.

Pakartotiniai bandymai ir delsos strategija

Kai užduotis sukelia išimtį, Queuety grąžina ją į pending su atidėtu available_at, kol ji bus paleista iš viso max_attempts kartų (pirmasis paleidimas įskaičiuojamas, todėl max_attempts( 3 ) reiškia, kad bus bandoma dar iki dviejų kartų). Kiekvieno pakartotinio bandymo delsą nustato delsos strategija, kurioje attempt yra ką tik nepavykusio paleidimo numeris:

Nustatykite visuotinę numatytąją reikšmę čia: wp-config.php:

define( 'QUEUETY_RETRY_BACKOFF', 'linear' );

Arba nustatykite ją kiekvienam apdorojimo metodui čia: config():

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

Arba nustatykite ją siųsdami užduotį:

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

Į nepavykusių užduočių saugyklą perkeltos užduotys

Išnaudojus visus pakartotinio bandymo mėginimus, užduotis pažymima kaip perkelta į saugyklą. Tokios užduotys lieka duomenų bazėje, kad jas būtų galima patikrinti, tačiau nebeapdorojamos. Jas galite bandyti dar kartą rankiniu būdu. Pakartojus bandymą rankiniu būdu, užduotis vėl tampa pending, tačiau jos bandymų skaičius nenustatomas iš naujo. Todėl dar kartą paleista saugykloje buvusi užduotis bus vykdoma dar vieną kartą ir, jei šis bandymas nepavyks, vėl bus perkelta į saugyklą:

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

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

Arba per CLI:

wp queuety retry 42
wp queuety retry-buried

Skirtasis laikas

Queuety nustatytam maksimaliam vykdymo laikui paleidžia pcntl_alarm() ir, jam pasibaigus, turėtų nutraukti bandymą su TimeoutException. Ši riboto laiko funkcija šiuo metu neveikia (#10): darbininkas neįjungia asinchroninių signalų, todėl ilgai veikiantis apdorojimo metodas vykdomas iki galo, o sleep() blokuojamas metodas nutraukiamas suskambus signalui ir pažymimas kaip užbaigtas. Visuotinę ribą nustatykite čia:

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

Jei užduotis būsenoje processing užstringa ilgiau nei QUEUETY_STALE_TIMEOUT sekundžių, ji atkuriama automatiškai:

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

Atkurkite pasenusias užduotis rankiniu būdu:

wp queuety recover

Eilės

Užduotys siunčiamos į pavadintas eiles. Kiekvienas darbininkas apdoroja vieną eilę arba eilės tvarka nurodytų eilių sąrašą (žr. skiltį Darbininkai):

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

wp queuety work --queue=emails

Numatytoji eilė yra default. Galite paleisti atskirus darbininkų procesus kiekvienai eilei ir taip atskirti darbo krūvius.

Dažnio ribojimas

Apribokite, kaip dažnai apdorojimo metodas gali būti vykdomas per tam tikrą laikotarpį:

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

Daugiau informacijos rasite skiltyje Dažnio ribojimas.

Išsiunčiamų užduočių klasės

Queuety v0.6.0 pristato klasių pagrindu veikiančių užduočių API. Užuot atskirai nurodę apdorojimo metodo pavadinimą, duomenis ir klasę, apibrėžiate vieną klasę, kurioje viskas sutelkta: duomenys (kaip tipizuotos viešosios savybės), vykdymo logika ir pasirinktinė tarpinė programinė įranga.

Sąsaja Contracts\Job

use Queuety\Contracts\Job;

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

Job klasė saugo duomenis viešosiose savybėse ir vykdo savo logiką metode handle(). Atskiro metodo config() ar apdorojimo metodo registruoti nereikia.

Trait Dispatchable

Pridėkite trait Dispatchable, kad užduoties klasėje būtų statinis metodas 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 );
    }
}

Išsiųskite ją perduodami konstruktoriaus argumentus:

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

Trait Dispatchable viduje iškviečia Queuety::dispatch(), todėl visos parinktys PendingJob (eilė, prioritetas, delsa, max_attempts, unikalumas, after) veikia taip pat.

Kaip veikia serializavimas

Kai išsiunčiate Job egzempliorių, Queuety JobSerializer ištraukia pilną klasės pavadinimą ir naudoja jį kaip apdorojimo metodą, o visas viešąsias savybes – kaip duomenis. Darbininko pusėje serializatorius atkuria egzempliorių, susiedamas duomenų raktus su konstruktoriaus parametrų pavadinimais. Enum reikšmės su pagrindine reikšme serializuojamos į atitinkamą reikšmę ir atkuriamos automatiškai.

Užduočių savybės

Užduočių klasėse galima apibrėžti viešąsias savybes, kuriomis valdoma pakartotinių bandymų ir laiko limito elgsena. Darbininkas jas nuskaito naudodamas refleksiją ir pritaiko automatiškai.

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

Masyvas $backoff nustato vis didesnes delsas tarp pakartotinių bandymų. Pavyzdžiui, [10, 30, 60, 120] reiškia, kad prieš pirmą pakartotinį bandymą laukiama 10 sekundžių, prieš antrą – 30 sekundžių, prieš trečią – 60 sekundžių, o prieš visus kitus – 120 sekundžių.

Su ištekliais susijusios savybės naudingos, kai užduočių klasės atlieka skirtingos apimties darbą. Spartus talpyklos atnaujinimas ir didelis kreipinys į teikėją gali abu būti „viena užduotis“, tačiau apskaičiuojant darbo eigos kvotą ar priimant sprendimą dėl darbininko priėmimo, jiems nebūtinai reikėtų skirti vienodą svorį, jei siekiama neviršyti saugių proceso ribų.

Išsamų modelio aprašą, įskaitant darbo eigos kvotas ir darbininkų priėmimą, rasite skiltyje Išteklių valdymas.

Nesėkmingo užbaigimo metodas

Kai išnaudojus visus pakartotinio bandymo mėginimus užduotis perkeliama į saugyklą, Queuety iškviečia užduoties egzemplioriaus metodą failed(), jei toks apibrėžtas. Naudokite jį valymui, įspėjimams ar registravimui žurnale.

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

Metodas failed() gauna išimtį, dėl kurios įvyko galutinė klaida. Užduoties egzempliorius deserializuojamas su pradiniais duomenimis, todėl prieinamos visos viešosios savybės.

Sąlyginis siuntimas

Trait Dispatchable suteikia metodus dispatch_if() ir dispatch_unless() sąlyginiam siuntimui. Abiem metodams pirmasis argumentas yra loginė sąlyga, po kurios pateikiami užduoties konstruktoriaus argumentai.

// 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() išsiunčia užduotį, kai sąlyga yra true, ir grąžina PendingJob. Kai sąlyga yra false, grąžinama null.

dispatch_unless() išsiunčia užduotį, kai sąlyga yra false, ir grąžina PendingJob. Kai sąlyga yra true, grąžinama null.

Išsiunčiant užduotį, abiejų metodų parinktis PendingJob galima susieti su kitomis parinktimis:

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

Sinchroninis siuntimas

Naudokite Queuety::dispatch_sync(), kad užduotis būtų įvykdyta iškart dabartiniame procese, neįtraukiant jos į eilę. Tai naudinga testuojant ir vykdant užduotis, kurios turi būti atliktos tiesiogiai.

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

Užduoties metodas handle() iškviečiamas tiesiogiai. Duomenų bazės eilutė nesukuriama, vykdytojas nedalyvauja, o tarpinė programinė įranga netaikoma.

Užduočių grandinė

Užduočių grandinė paeiliui vykdo užduotis vieną po kitos. Kiekviena grandinės užduotis priklauso nuo sėkmingo ankstesnės užbaigimo. Jei kuri nors užduotis nepavyksta, vėlesnės grandinės užduotys lieka laukimo būsenoje.

Naudojant 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() grąžina ChainBuilder su sklandžiai susiejamais metodais eilei ir nesėkmės apdorojimo metodui nustatyti. Metodas dispatch() sukuria visas užduotis su nuosekliais depends_on ryšiais ir grąžina pirmosios užduoties ID.

Naudojant with_chain()

Trait Dispatchable pateikia metodą with_chain(). Šiuo metu jis prilygsta Queuety::chain( $jobs ): klasė, kurioje jis iškviečiamas, į grandinę neįtraukiama (#13), todėl į masyvą įtraukite ir pirmąją užduotį:

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

Grandinės nesėkmės apdorojimas

Užregistruokite nesėkmės apdorojimo klasę naudodami catch(). Šioje klasėje turi būti metodas handle(). Jis iškviečiamas, kai kuri nors grandinės užduotis visam laikui perkeliama į saugyklą. Darbininkas palaiko handle(), handle( Job $job ) ir handle( Job $job, \Throwable $exception ).

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

Tarpinės programinės įrangos pridėjimas

Užduočių klasėse galima apibrėžti metodą middleware(), kuris grąžina vykdymą apgaubiančią tarpinę programinę įrangą:

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

Daugiau informacijos apie integruotą tarpinę programinę įrangą ir pasirinktinės kūrimą rasite skiltyje Tarpinė programinė įranga.

Apdorojimo metodais pagrįstų užduočių palyginimas

Palaikomi abu būdai. Esamos apdorojimo metodais pagrįstos užduotys toliau veiks be pakeitimų. Tame pačiame projekte galite naudoti abu būdus.

Užduoties gyvavimo ciklas

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