Tööd

Tööd on Queuety-i põhiline tööühik. Saada töö järjekorda käsitleja nime ja koormaga ning tööline võtab selle ja käivitab taustal.

Tööde järjekorda saatmine

use Queuety\Queuety;

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

dispatch() tagastab PendingJob koosturi. Enne automaatset järjekorda saatmist saad sellele valikuid aheldada:

use Queuety\Enums\Priority;

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

Töö ID kohe kättesaamiseks kutsu välja ->id():

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

Käsitlejate kirjutamine

Käsitlejaks sobib iga klass, mis rakendab liidest 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',
        ];
    }
}

Meetod handle() saab töö koorma ja teeb töö ära. Meetod config() tagastab valikulised vaikeväärtused:

Need ressursivõtmed kuuluvad Queuety laiema Resource Management mudeli juurde, kus töö metaandmed, töövoo eelarved ja tööliste vastuvõtt toimivad koos.

Registreeri käsitleja:

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

Või kasutage automaatseks registreerimiseks PHP atribuute.

Prioriteet

Töödel on neli prioriteeditaset. Töötajad võtavad esmalt ette kõrgema prioriteediga tööd.

use Queuety\Enums\Priority;

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

Viivitus

Lükka töö edasi, et see muutuks kättesaadavaks alles määratud arvu sekundite möödudes:

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

Töö lisatakse kohe, kuid töötajad ei võta seda ette enne viivitusaja möödumist.

Unikaalsed tööd

Väldi sama käsitleja ja andmete korduvat tööle saatmist. Kui sama käsitleja ja andmetega ootel või töötlemisel töö on juba olemas, tagastatakse uue loomise asemel olemasoleva töö ID.

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

Töö sõltuvused

Töö võib sõltuda teisest tööst. Töötaja ei võta seda ette enne, kui sõltuv töö on lõpetatud.

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

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

Tööde paketina saatmine

Saada mitu tööd ühe mitmerealise INSERT-päringuga, et saavutada maksimaalne läbilaskevõime:

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

Iga üksus võib sisaldada valikulisi võtmeid: queue, priority, delay, max_attempts.

Uuesti proovimine ja viivitused

Kui töö käivitab erandi, paneb Queuety selle tagasi olekusse pending ja lisab sellele viivitusega available_at, kuni töö on käivitatud kokku max_attempts korda (esimene käivitus läheb arvesse, seega max_attempts( 3 ) tähendab kuni kahte korduskatset). Iga korduskatse eelneva viivituse määrab viivituse strateegia, kus attempt tähistab äsja ebaõnnestunud käivituse numbrit:

Määra globaalne vaikeväärtus asukohas wp-config.php:

define( 'QUEUETY_RETRY_BACKOFF', 'linear' );

Või määra see iga käsitleja jaoks asukohas config():

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

Või iga tööle saatmise korral:

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

Ootele paigutatud tööd

Kui kõik korduskatsed on ammendatud, märgitakse töö maetud olekusse. Maetud tööd jäävad andmebaasi kontrollimiseks alles, kuid neid ei töödelda. Saad neid käsitsi uuesti proovida. Käsitsi korduskatse viib töö uuesti olekusse pending, katsete arvu lähtestamata, nii et uuesti proovitud maetud töö käivitatakse veel ühe korra ja maetakse ebaõnnestumise korral taas:

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

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

Või CLI kaudu:

wp queuety retry 42
wp queuety retry-buried

Ajalimiit

Queuety käivitab pcntl_alarm() maksimaalse käitusaja jaoks ning peaks tähtaja ületamisel katse nurjama, andes TimeoutException. Seda ajalõppu praegu ei jõustata (#10): töötaja ei luba asünkroonseid signaale, mistõttu pika käitusajaga käsitleja töötab lõpuni; alarmi käivitumisel katkestatakse aga sleep()-is blokeeritud käsitleja ja märgitakse lõpetatuks. Määra ajalõpp globaalselt:

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

Tööd, mis on olekus processing kauem kui QUEUETY_STALE_TIMEOUT sekundit, taastatakse automaatselt:

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

Taastage aegunud tööd käsitsi:

wp queuety recover

Järjekorrad

Tööd saadetakse nimelisse järjekorda. Iga töötaja töötleb üht järjekorda või järjestatud järjekordade loendit (vt Töötajad):

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

wp queuety work --queue=emails

Vaikimisi järjekord on default. Töökoormuste eraldamiseks saad käitada iga järjekorra jaoks eraldi töötajaprotsessi.

Kiiruse piiramine

Piira, kui sageli saab käsitlejat kindla ajavahemiku jooksul käivitada:

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

Lisateavet leiad jaotisest Kiiruse piiramine.

Töölähetatavad tööklassid

Queuety v0.6.0 lisab klassipõhise töö API. Selle asemel et hoida käsitleja nimi, andmed ja käsitlejaklass eraldi, määrad ühe klassi, mis koondab kõik: andmed (tüübitud avalike atribuutidena), käivitamisloogika ja valikulise vahetarkvara.

Liides Contracts\Job

use Queuety\Contracts\Job;

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

Klass Job hoiab oma andmeid avalike atribuutidena ja käivitab oma loogika meetodis handle(). Eraldi meetodit config() ega käsitleja registreerimist pole vaja.

Tunnusjoon Dispatchable

Lisa tunnusjoon Dispatchable, et lisada oma tööklassile staatiline meetod 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 );
    }
}

Saada see tööle, edastades konstruktori argumendid:

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

Tunnusjoon Dispatchable kutsub sisemiselt välja Queuety::dispatch(), nii et kõik valikud PendingJob (järjekord, prioriteet, viivitus, max_attempts, unique, after) toimivad täpselt samamoodi.

Kuidas serialiseerimine toimib

Kui saadad tööklassi Job eksemplari tööle, eraldab Queuety JobSerializer-i abil täielikult kvalifitseeritud klassinime käsitlejana ja kõik avalikud atribuudid andmetena. Töötaja taastab eksemplari, sobitades andmete võtmed konstruktori parameetrite nimedega. Backed enumid serialiseeritakse nende aluseks olevaks väärtuseks ja taastatakse automaatselt.

Töö omadused

Tööklassid võivad korduskatsete ja ajalõpu käitumise juhtimiseks deklareerida avalikke atribuute. Töötaja loeb need peegelduse abil ja rakendab automaatselt.

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

Massiiv $backoff määrab kasvavad korduskatsete viivitused. Näiteks [10, 30, 60, 120] tähendab, et enne esimest korduskatset oodatakse 10 sekundit, enne teist 30 sekundit, enne kolmandat 60 sekundit ja enne edasisi korduskatseid 120 sekundit.

Ressursiga seotud atribuudid on kasulikud, kui tööklassid esindavad eri kuluprofiile. Kiire vahemälu värskendamine ja suur teenusepakkuja päring võivad mõlemad olla „üks töö”, kuid töövoo eelarve või töötajale lubamise otsuse puhul ei peaks neid tingimata võrdselt arvestama, kui eesmärk on hoida protsess turvalistes piirides.

Täielikku mudelit, sealhulgas töövoo eelarveid ja töötajapoolset lubamist, kirjeldab Ressursihaldus.

Ebaõnnestumise hook

Kui töö maetakse pärast kõigi korduskatsete ammendamist, kutsub Queuety töö eksemplaril meetodi failed() välja, kui see on määratud. Kasuta seda puhastuseks, teavitusteks või logimiseks.

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

Meetod failed() saab erandi, mis põhjustas lõpliku ebaõnnestumise. Töö eksemplar deserialiseeritakse algsete andmetega, seega on kõik avalikud atribuudid saadaval.

Tingimuslik tööle saatmine

Tunnusjoon Dispatchable pakub tingimuslikuks tööle saatmiseks meetodeid dispatch_if() ja dispatch_unless(). Mõlemad võtavad esimeseks argumendiks tõeväärtustingimuse, millele järgnevad töö konstruktori argumendid.

// 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() saadab tööle, kui tingimus on true, ja tagastab PendingJob. Kui tingimus on false, tagastab see null.

dispatch_unless() saadab tööle, kui tingimus on false, ja tagastab PendingJob. Kui tingimus on true, tagastab see null.

Mõlemad meetodid toetavad tööle saatmisel aheldatavaid valikuid PendingJob:

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

Sünkroonne tööle saatmine

Kasuta Queuety::dispatch_sync()-i, et käivitada töö kohe praeguses protsessis seda järjekorda lisamata. See on kasulik testimiseks ja tööde puhul, mis peavad käivituma samas protsessis.

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

Töö meetodit handle() kutsutakse otse. Andmebaasi rida ei looda, töötajat ei kaasata ja vahevara ei rakendata.

Tööde aheldamine

Tööde aheldamine käivitab tööd üksteise järel. Iga ahelas olev töö sõltub sellele eelneva töö edukast lõpetamisest. Kui mõni töö ebaõnnestub, jäävad järgnevad tööd ootele.

Kasutades 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() tagastab ChainBuilder, mille aheldatavate meetoditega saab määrata järjekorra ja ebaõnnestumise käsitleja. Meetod dispatch() loob kõik tööd järjestikuste depends_on seostega ja tagastab esimese töö ID.

Kasutades with_chain()

Tunnusjoon Dispatchable pakub meetodit with_chain(). Praegu on see samaväärne meetodiga Queuety::chain( $jobs ): klassi, millel seda kutsutakse, ahelasse ei lisata (#13), seega loetle massiivis ka esimene töö:

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

Ahela ebaõnnestumise käsitlemine

Registreeri ebaõnnestumise käsitleja klass meetodiga catch(). Käsitlejaklassil peab olema meetod handle(). See kutsutakse välja, kui mõni töö ahelas püsivalt maetakse. Töötaja toetab valikuid handle(), handle( Job $job ) ja handle( Job $job, \Throwable $exception ).

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

Vahetarkvara lisamine

Tööklassid võivad määrata meetodi middleware(), mis tagastab käivitamist ümbritseva vahetarkvara:

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

Sisseehitatud vahetarkvara ja kohandatud vahetarkvara kirjutamise üksikasjad leiad jaotisest Vahetarkvara.

Käsitlejapõhiste ja klassipõhiste tööde võrdlus

Mõlemad lähenemisviisid on täielikult toetatud. Olemasolevad käsitlejapõhised tööd toimivad muudatusteta edasi. Samas projektis saad kasutada mõlemat stiili.

Töö elutsükkel

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