Opravila

Opravila so osnovna enota dela v Queuety. Pošljite opravilo v izvajanje z imenom obravnavalnika in vsebino, nato pa ga delavec prevzame in izvede v ozadju.

Pošiljanje opravil v izvajanje

use Queuety\Queuety;

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

dispatch() vrne gradnik PendingJob. Pred samodejnim pošiljanjem lahko verižite 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 );

Če želite takoj dobiti ID opravila, pokličite ->id():

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

Pisanje obravnavalnikov

Obravnavalnik je kateri koli razred, ki implementira vmesnik 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() prejme vsebino opravila in izvede delo. Metoda config() vrne neobvezne privzete vrednosti:

Ti ključi virov so del širšega modela upravljanja virov Queuety, v katerem skupaj delujejo metapodatki opravil, proračuni potekov dela in sprejemanje delavcev: Upravljanje virov.

Registrirajte obravnavalnik:

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

Ali pa za samodejno registracijo uporabite atribute PHP.

Prednostna raven

Opravila imajo štiri prednostne ravni. Delavci najprej prevzamejo opravila z višjo prednostno ravnjo.

use Queuety\Enums\Priority;

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

Zakasnitev

Zakasnite opravilo, da bo na voljo šele po določenem številu sekund:

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

Opravilo se vstavi takoj, vendar ga delavci ne bodo prevzeli, dokler se zakasnitev ne izteče.

Enolična opravila

Preprečite podvojeno pošiljanje za isti obravnavalnik in vsebino. Če že obstaja čakajoče ali obdelovano opravilo z istim obravnavalnikom in vsebino, se namesto ustvarjanja novega vrne ID obstoječega opravila.

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

Odvisnosti opravil

Opravilo je lahko odvisno od drugega opravila. Delavec ga ne bo prevzel, dokler odvisno opravilo ni dokončano.

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

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

Skupinsko pošiljanje

Pošljite več opravil z enim večvrstičnim INSERT za največjo prepustnost:

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

Vsak element lahko vključuje neobvezne ključe: queue, priority, delay, max_attempts.

Ponovni poskusi in eksponentno čakanje

Ko opravilo sproži izjemo, ga Queuety vrne v pending z zakasnjenim available_at, dokler se skupaj ne izvede max_attempts-krat (šteje se tudi prvi poskus, zato max_attempts( 3 ) pomeni največ dva ponovna poskusa). Zakasnitev pred vsakim ponovnim poskusom določa strategija čakanja, pri čemer je attempt številka pravkar neuspelega poskusa:

Nastavite globalno privzeto vrednost v wp-config.php:

define( 'QUEUETY_RETRY_BACKOFF', 'linear' );

Ali pa za posamezen obravnavalnik v config():

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

Ali pa za posamezno pošiljanje:

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

Arhivirana opravila

Ko so vsi poskusi ponovnega zagona izčrpani, je opravilo označeno kot zakopano. Zakopana opravila ostanejo v zbirki podatkov za pregled, vendar se ne obdelujejo. Ročno jih lahko znova poskusite izvesti. Pri ročnem ponovnem poskusu se opravilo znova pending, ne da bi se ponastavilo število poskusov, zato se zakopano opravilo izvede še enkrat in je ob neuspehu tega poskusa znova zakopano:

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

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

Ali prek CLI:

wp queuety retry 42
wp queuety retry-buried

Časovna omejitev

Queuety nastavi pcntl_alarm() za najdaljši čas izvajanja in naj bi ob prekoračitvi sprožil neuspeh poskusa z TimeoutException. Ta časovna omejitev trenutno ni uveljavljena (#10): delavec ne omogoči asinhronih signalov, zato se dolgotrajni obravnavalnik izvede do konca, obravnavalnik, blokiran v sleep(), pa se prekine, ko se sproži alarm, in je nato označen kot dokončan. Globalno nastavite časovno omejitev:

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

Opravila, ki so v stanju processing dlje kot QUEUETY_STALE_TIMEOUT sekund, se samodejno obnovijo:

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

Zastarela opravila obnovite ročno:

wp queuety recover

Čakalne vrste

Opravila se pošiljajo v poimenovane čakalne vrste. Vsak delavec obdeluje eno čakalno vrsto ali urejen seznam čakalnih vrst (glejte Delavci):

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

wp queuety work --queue=emails

Privzeta čakalna vrsta je default. Za ločevanje delovnih obremenitev lahko za vsako čakalno vrsto zaženete ločen proces delavca.

Omejevanje hitrosti

Omejite, kako pogosto se lahko obravnavalnik izvede v določenem časovnem obdobju:

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

Podrobnosti najdete v razdelku Omejevanje hitrosti.

Razredi opravil, ki jih je mogoče poslati

Queuety v0.6.0 uvaja opravila, temelječa na razredih, prek API. Namesto ločenega navajanja imena obravnavalnika, vsebine in razreda obravnavalnika določite en razred, ki združuje vse: vsebino (kot tipizirane javne lastnosti), logiko izvajanja in neobvezno vmesno programsko opremo.

Vmesnik Contracts\Job

use Queuety\Contracts\Job;

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

Razred Job hrani svojo vsebino kot javne lastnosti in izvaja lastno logiko v handle(). Ločena metoda config() ali registracija obravnavalnika nista potrebni.

Trait Dispatchable

Razredu opravila dodajte trait Dispatchable, da pridobite statično metodo 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 );
    }
}

Pošljite ga tako, da posredujete argumente konstruktorja:

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

Trait Dispatchable interno pokliče Queuety::dispatch(), zato vse možnosti PendingJob (queue, priority, delay, max_attempts, unique, after) delujejo enako.

Kako deluje serializacija

Ko pošljete primerek Job, Queuety iz JobSerializer pridobi polno kvalificirano ime razreda kot obravnavalnik in vse javne lastnosti kot vsebino. Na strani delavca serializator obnovi primerek tako, da ključe vsebine poveže z imeni parametrov konstruktorja. Naštevanja s podprto vrednostjo se serializirajo v osnovno vrednost in samodejno obnovijo.

Lastnosti opravil

Razredi opravil lahko določijo javne lastnosti za nadzor ponovnih poskusov in časovne omejitve. Delavec jih prebere z refleksijo in samodejno uporabi.

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

Polje $backoff določa naraščajoče zakasnitve med ponovnimi poskusi. Na primer, [10, 30, 60, 120] pomeni, da je treba pred prvim ponovnim poskusom počakati 10 sekund, pred drugim 30 sekund, pred tretjim 60 sekund, pred vsemi nadaljnjimi pa 120 sekund.

Lastnosti, povezane z viri, so uporabne, kadar razredi opravil predstavljajo različne stroškovne profile. Hitro osveževanje predpomnilnika in obsežen klic ponudnika sta lahko oba »eno opravilo«, vendar se ne bi smela nujno šteti enako, ko proračun delovnega toka ali odločitev o sprejemu opravila poskuša proces ohraniti znotraj varnih omejitev.

Celoten model, vključno s proračuni delovnih tokov in sprejemanjem opravil na strani delavca, je opisan v razdelku Upravljanje virov.

Kavelj ob napaki

Ko je opravilo zakopano po izčrpanju vseh poskusov, Queuety pokliče metodo failed() na primerku opravila, če je ta določena. Uporabite jo za čiščenje, opozarjanje ali beleženje.

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() prejme izjemo, ki je povzročila končni neuspeh. Primerek opravila se deserializira z izvirno vsebino, zato so na voljo vse javne lastnosti.

Pogojno pošiljanje

Trait Dispatchable omogoča dispatch_if() in dispatch_unless() za pogojno pošiljanje. Obe sprejmeta logični pogoj kot prvi argument, za njim pa argumente konstruktorja opravila.

// 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() pošlje opravilo, ko je pogoj true, in vrne PendingJob. Vrne null, ko je pogoj false.

dispatch_unless() pošlje opravilo, ko je pogoj false, in vrne PendingJob. Vrne null, ko je pogoj true.

Obe metodi podpirata veriženje možnosti PendingJob pri pošiljanju:

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

Sinhrono pošiljanje

Uporabite Queuety::dispatch_sync() za takojšnje izvajanje opravila v trenutnem procesu, ne da bi ga dodali v čakalno vrsto. To je uporabno za testiranje in opravila, ki se morajo izvesti neposredno.

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 handle() opravila se pokliče neposredno. Vrstica v zbirki podatkov se ne ustvari, delavec ni vključen, vmesna programska oprema pa se ne uporabi.

Veriga opravil

Veriženje opravil izvede zaporedje opravil enega za drugim. Vsako opravilo v verigi je odvisno od uspešnega zaključka prejšnjega. Če katero koli opravilo spodleti, naslednja opravila v verigi ostanejo čakajoča.

Uporaba 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() vrne ChainBuilder s tekočimi metodami za nastavitev čakalne vrste in obravnavalnika napak. Metoda dispatch() ustvari vsa opravila z zaporednimi povezavami depends_on in vrne ID prvega opravila.

Uporaba with_chain()

Trait Dispatchable ponuja metodo with_chain(). Trenutno je enakovredna metodi Queuety::chain( $jobs ): razred, na katerem je poklicana, ni dodan v verigo (#13), zato v polje vključite tudi prvo opravilo:

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

Obravnava napak v verigi

Registrirajte razred obravnavalnika napak z uporabo catch(). Razred obravnavalnika mora imeti metodo handle(). Pokliče se, ko je katero koli opravilo v verigi trajno zakopano. Delavec podpira handle(), handle( Job $job ) in handle( Job $job, \Throwable $exception ).

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

Dodajanje vmesne programske opreme

Razredi opravil lahko določijo metodo middleware(), ki vrne vmesno programsko opremo za ovijanje izvajanja:

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 vgrajeni vmesni programski opremi in pisanju lastne najdete v razdelku Vmesna programska oprema.

Primerjava opravil, ki temeljijo na obravnavalniku

Oba pristopa sta v celoti podprta. Obstoječa opravila, ki temeljijo na obravnavalniku, še naprej delujejo brez sprememb. V istem projektu lahko kombinirate oba sloga.

Življenjski cikel opravila

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