Työt

Työt ovat Queuety-järjestelmän perusyksikkö. Lähetä työ suoritettavaksi käsittelijän nimellä ja hyötykuormalla, niin työntekijä poimii ja suorittaa sen taustalla.

Töiden lähettäminen

use Queuety\Queuety;

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

dispatch() palauttaa PendingJob-rakentajan. Voit ketjuttaa siihen asetuksia ennen kuin se lähettää työn automaattisesti:

use Queuety\Enums\Priority;

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

Saat työn tunnuksen heti kutsumalla toimintoa ->id():

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

Käsittelijöiden kirjoittaminen

Käsittelijä on mikä tahansa luokka, joka toteuttaa Queuety\Handler-rajapinnan:

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

Metodi handle() vastaanottaa työn hyötykuorman ja suorittaa työn. Metodi config() palauttaa valinnaiset oletusarvot:

Nämä resurssiavaimet ovat osa Queuety-järjestelmän laajempaa Resurssienhallinta-mallia, jossa työn metatiedot, työnkulkubudjetit ja työntekijöiden hyväksyntä toimivat yhdessä.

Rekisteröi käsittelijä:

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

Tai käytä PHP-attribuutteja automaattiseen rekisteröintiin.

Prioriteetti

Töillä on neljä prioriteettitasoa. Työntekijät varaavat ensin korkeamman prioriteetin työt.

use Queuety\Enums\Priority;

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

Viive

Aseta työlle viive, jotta se on käytettävissä vasta määritetyn sekuntimäärän kuluttua:

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

Työ lisätään heti, mutta työntekijät eivät varaa sitä ennen kuin viive on kulunut.

Yksilölliset työt

Estä saman käsittelijän ja hyötykuorman töiden lähettäminen kahteen kertaan. Jos odottava tai käsittelyssä oleva työ samalla käsittelijällä ja hyötykuormalla on jo olemassa, palautetaan sen nykyinen tunnus uuden työn luomisen sijaan.

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

Työriippuvuudet

Työ voi riippua toisesta työstä. Työntekijä ei varaa sitä ennen kuin riippuvuus on valmis.

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

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

Erälähetys

Lähetä useita töitä yhdellä usean rivin INSERT-lauseella, jotta läpimeno on mahdollisimman suuri:

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

Jokainen kohde voi sisältää valinnaisia avaimia: queue, priority, delay, max_attempts.

Uudelleenyritys ja viive

Kun työ aiheuttaa poikkeuksen, Queuety palauttaa sen arvoon pending viivästetyllä available_at-toiminnolla, kunnes työ on suoritettu yhteensä max_attempts kertaa (ensimmäinen suorituskerta lasketaan mukaan, joten max_attempts( 3 ) tarkoittaa enintään kahta uudelleenyritystä). Kunkin uudelleenyrityksen viive määräytyy viivestrategian mukaan. attempt ilmaisee juuri epäonnistuneen suorituskerran numeron:

Aseta yleinen oletusarvo kohteessa wp-config.php:

define( 'QUEUETY_RETRY_BACKOFF', 'linear' );

Tai käsittelijäkohtaisesti kohdassa config():

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

Tai lähetyskertakohtaisesti:

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

Sivuun siirretyt työt

Kun kaikki uudelleenyritysyritykset on käytetty, työ merkitään haudatuksi. Haudatut työt säilyvät tietokannassa tarkastelua varten, mutta niitä ei käsitellä. Voit yrittää niitä uudelleen manuaalisesti. Manuaalinen uudelleenyritys asettaa työn jälleen tilaan pending nollaamatta sen yritysten määrää, joten uudelleen yritetty haudattu työ suoritetaan vielä kerran ja haudataan uudelleen, jos suoritus epäonnistuu:

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

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

Tai käyttämällä CLI:

wp queuety retry 42
wp queuety retry-buried

Aikakatkaisu

Queuety käynnistää pcntl_alarm()-ajastimen enimmäissuoritusajalle, ja sen on tarkoitus epäonnistua yrityksessä TimeoutException-virheellä, jos aika ylittyy. Aikakatkaisua ei tällä hetkellä valvota (#10): työntekijä ei ota asynkronisia signaaleja käyttöön, joten pitkään suoritettava käsittelijä jatkaa loppuun asti, ja sleep()-kutsuun jumiutunut käsittelijä keskeytetään hälytyksen lauetessa ja merkitään sitten suoritetuksi. Määritä aikakatkaisu yleisesti:

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

Työt, joiden tila on processing yli QUEUETY_STALE_TIMEOUT sekuntia, palautetaan automaattisesti:

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

Palauta vanhentuneet työt manuaalisesti:

wp queuety recover

Jonot

Työt lähetetään nimettyihin jonoihin. Kukin työntekijä käsittelee yhtä jonoa tai järjestettyä jonolistaa (katso Työntekijät):

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

wp queuety work --queue=emails

Oletusjono on default. Voit eristää työkuormia suorittamalla kullekin jonolle omaa työntekijäprosessia.

Nopeuden rajoittaminen

Rajoita, kuinka usein käsittelijä voidaan suorittaa tietyn aikavälin aikana:

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

Lisätietoja on kohdassa Nopeuden rajoittaminen.

Lähetettävät työluokat

Queuety v0.6.0 esittelee luokkapohjaisen työn API. Erillisten käsittelijän nimen, hyötykuorman ja käsittelijäluokan sijaan määrittelet yhden luokan, joka kokoaa kaiken yhteen: hyötykuorman (tyypitettyinä julkisina ominaisuuksina), suorituslogiikan ja valinnaiset väliohjelmistot.

Contracts\Job-rajapinta

use Queuety\Contracts\Job;

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

Job-luokka sisältää hyötykuormansa julkisina ominaisuuksina ja suorittaa oman logiikkansa kohdassa handle(). Erillistä config()-metodia tai käsittelijän rekisteröintiä ei tarvita.

Dispatchable-trait

Lisää Dispatchable-trait saadaksesi staattisen dispatch()-metodin työluokkaasi:

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

Lähetä työ välittämällä konstruktorin argumentit:

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

Dispatchable-trait kutsuu sisäisesti Queuety::dispatch()-metodia, joten kaikki PendingJob-asetukset (jono, prioriteetti, viive, max_attempts, unique, after) toimivat täsmälleen samalla tavalla.

Sarjallistamisen toiminta

Kun lähetät Job-esiintymän, Queuety-luokan JobSerializer poimii täysin määritetyn luokan nimen käsittelijäksi ja kaikki julkiset ominaisuudet hyötykuormaksi. Työntekijän puolella sarjallistaja muodostaa esiintymän uudelleen yhdistämällä hyötykuorman avaimet konstruktorin parametrien nimiin. Enum-arvot, joilla on tausta-arvo, sarjallistetaan niiden perustana olevaksi arvoksi ja palautetaan automaattisesti.

Työn ominaisuudet

Työluokat voivat määrittää julkisia ominaisuuksia uudelleenyritys- ja aikakatkaisutoiminnan ohjaamiseksi. Työntekijä lukee ne reflektiolla ja soveltaa niitä automaattisesti.

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

$backoff-taulukko määrittää kasvavat uudelleenyritysviiveet. Esimerkiksi [10, 30, 60, 120] tarkoittaa, että ensimmäistä uudelleenyritystä odotetaan 10 sekuntia, toista 30 sekuntia ja kolmatta 60 sekuntia. Kaikissa myöhemmissä yrityksissä odotetaan 120 sekuntia.

Resurssiominaisuuksista on hyötyä, kun työluokat edustavat erilaisia kustannusprofiileja. Nopea välimuistin päivitys ja suuri palveluntarjoajakutsu voivat kumpikin olla ”yksi työ”, mutta niitä ei välttämättä pidä laskea samalla tavalla, kun työnkulun budjetti tai työntekijän pääsyä koskeva päätös pyrkii pitämään prosessin turvallisissa rajoissa.

Katso koko mallin kuvaus, mukaan lukien työnkulun budjetit ja työntekijäpuolen pääsynhallinta, kohdasta Resurssien hallinta.

Epäonnistuneen työn koukku

Kun työ haudataan kaikkien uudelleenyritysyritysten epäonnistuttua, Queuety kutsuu työesiintymän failed()-metodia, jos sellainen on määritelty. Käytä sitä siivoukseen, hälytyksiin tai lokitukseen.

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

failed()-metodi saa parametrikseen lopullisen epäonnistumisen aiheuttaneen poikkeuksen. Työesiintymä sarjallistetaan alkuperäisestä hyötykuormastaan, joten kaikki julkiset ominaisuudet ovat käytettävissä.

Ehdollinen lähetys

Dispatchable-trait tarjoaa ehdolliseen lähettämiseen metodit dispatch_if() ja dispatch_unless(). Molemmat saavat ensimmäisenä argumenttinaan totuusarvoisen ehdon ja sen jälkeen työn konstruktorin argumentit.

// 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() lähettää työn, kun ehto on true, ja palauttaa PendingJob. Se palauttaa arvon null, kun ehto on false.

dispatch_unless() lähettää työn, kun ehto on false, ja palauttaa PendingJob. Se palauttaa arvon null, kun ehto on true.

Molemmat metodit tukevat ketjutettavia PendingJob-asetuksia työn lähetyksen yhteydessä:

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

Synkroninen lähetys

Käytä Queuety::dispatch_sync()-metodia suorittaaksesi työn heti nykyisessä prosessissa lisäämättä sitä jonoon. Tämä sopii testaamiseen ja töihin, jotka on suoritettava suoraan.

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

Työn handle()-metodia kutsutaan suoraan. Tietokantariviä ei luoda, työntekijää ei käytetä eikä väliohjelmistoa sovelleta.

Töiden ketjutus

Töiden ketjutuksessa tehtävät suoritetaan järjestyksessä yksi toisensa jälkeen. Kukin ketjun työ riippuu siitä, että edellinen työ valmistuu onnistuneesti. Jos jokin työ epäonnistuu, ketjun seuraavat työt jäävät odottamaan.

Queuety::chain()-metodin käyttö

use Queuety\Queuety;

$first_job_id = Queuety::chain( [
    new FetchDataJob( $url ),
    new ProcessDataJob(),
    new NotifyCompleteJob(),
] )
    ->on_queue( 'pipeline' )
    ->catch( ChainFailedHandler::class )
    ->dispatch();

Queuety::chain() palauttaa arvon ChainBuilder, jossa on sujuvan rajapinnan metodit jonon ja epäonnistumisen käsittelijän määrittämiseen. dispatch()-metodi luo kaikki työt peräkkäisinä depends_on-suhteina ja palauttaa ensimmäisen työn tunnuksen.

with_chain()-metodin käyttö

Dispatchable-trait tarjoaa with_chain()-metodin. Se vastaa tällä hetkellä Queuety::chain( $jobs )-metodia: luokkaa, jolle metodia kutsutaan, ei lisätä ketjuun (#13), joten lisää myös taulukon ensimmäinen työ:

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

Ketjun epäonnistumisen käsittely

Rekisteröi epäonnistumisen käsittelijäluokka käyttämällä catch()-metodia. Käsittelijäluokassa on oltava handle()-metodi. Sitä kutsutaan, kun jokin ketjun työ haudataan pysyvästi. Työntekijä tukee arvoja handle(), handle( Job $job ) ja handle( Job $job, \Throwable $exception ).

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

Väliohjelmiston lisääminen

Työluokat voivat määrittää middleware()-metodin, joka palauttaa suorituksen ympärille kietoutuvan väliohjelmiston:

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

Lisätietoja sisäänrakennetuista väliohjelmistoista ja omien väliohjelmistojen kirjoittamisesta on kohdassa Väliohjelmistot.

Vertailu käsittelijäpohjaisiin töihin

Molempia tapoja tuetaan täysin. Nykyiset käsittelijäpohjaiset työt toimivat edelleen ilman muutoksia. Voit käyttää molempia tyylejä samassa projektissa.

Työn elinkaari

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