Uzdevumi

Uzdevumi ir Queuety darba pamatvienība. Nosūtiet uzdevumu, norādot apdarinātāja nosaukumu un datus; darba process to paņems un izpildīs fonā.

Uzdevumu nosūtīšana

use Queuety\Queuety;

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

dispatch() atgriež PendingJob veidotāju. Varat savienot opcijas, pirms tas automātiski nosūta uzdevumu:

use Queuety\Enums\Priority;

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

Lai uzreiz iegūtu uzdevuma ID, izsauciet ->id():

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

Apdarinātāju rakstīšana

Apdarinātājs ir jebkura klase, kas implementē saskarni 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',
        ];
    }
}

Metode handle() saņem uzdevuma datus un veic darbu. Metode config() atgriež neobligātās noklusējuma vērtības:

Šie resursu taustiņi ir daļa no plašākā Resursu pārvaldības modeļa, kurā kopā darbojas uzdevumu metadati, darbplūsmu budžeti un darba procesu piekļuves kontrole Queuety.

Reģistrējiet apdarinātāju:

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

Vai izmantojiet PHP atribūtus automātiskai reģistrēšanai.

Prioritāte

Uzdevumiem ir četri prioritātes līmeņi. Darba procesi vispirms paņem augstākas prioritātes uzdevumus.

use Queuety\Enums\Priority;

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

Aizkave

Aizkavējiet uzdevumu, lai tas kļūtu pieejams tikai pēc norādītā sekunžu skaita:

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

Uzdevums tiek ievietots uzreiz, taču darba procesi to nepaņems, līdz būs pagājis aizkaves laiks.

Unikāli uzdevumi

Novērsiet viena apdarinātāja un datu dublētu nosūtīšanu. Ja jau pastāv gaidošs vai apstrādājams uzdevums ar tādu pašu apdarinātāju un datiem, jauna uzdevuma vietā tiek atgriezts esošā uzdevuma ID.

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

Uzdevumu atkarības

Uzdevums var būt atkarīgs no cita uzdevuma. Darba process to nepaņems, līdz būs pabeigts atkarīgais uzdevums.

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

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

Uzdevumu nosūtīšana partijā

Nosūtiet vairākus uzdevumus vienā vairāku rindu INSERT, lai sasniegtu maksimālu caurlaidspēju:

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

Katram elementam var norādīt neobligātās atslēgas: queue, priority, delay, max_attempts.

Atkārtoti mēģinājumi un aizkaves palielināšana

Ja uzdevums rada izņēmumu, Queuety to ievieto atpakaļ rindā pending ar aizkavētu available_at, līdz tas kopumā būs izpildīts max_attempts reizes (skaitot pirmo izpildes reizi; tātad max_attempts( 3 ) nozīmē, ka būs līdz diviem atkārtotiem mēģinājumiem). Aizkavi pirms katra atkārtotā mēģinājuma nosaka atkāpšanās stratēģija, kur attempt ir tikko neveiksmīgās izpildes kārtas numurs:

Iestatiet globālo noklusējuma vērtību šeit: wp-config.php:

define( 'QUEUETY_RETRY_BACKOFF', 'linear' );

Vai katram apstrādātājam config():

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

Vai katrai izsūtīšanai:

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

Apbedītie darbi

Kad visi atkārtotie mēģinājumi ir izsmelti, uzdevumam tiek piešķirts statuss buried. Šādi uzdevumi paliek datubāzē pārbaudei, bet netiek apstrādāti. Tos var mēģināt izpildīt vēlreiz manuāli. Manuāls atkārtots mēģinājums atjauno uzdevuma statusu uz pending, neatiestatot mēģinājumu skaitu, tāpēc šādam uzdevumam būs vēl viena izpilde, un kļūmes gadījumā tas atkal tiks apglabāts:

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

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

Vai ar CLI:

wp queuety retry 42
wp queuety retry-buried

Noildze

Queuety iestata pcntl_alarm() uz maksimālo izpildes laiku, un, to pārsniedzot, mēģinājums ir jāatzīmē kā neveiksmīgs ar TimeoutException. Pašlaik šis taimauts netiek piemērots (#10): darbinieks neiespējo asinhronos signālus, tāpēc ilgstoši izpildāms apstrādātājs darbojas līdz galam, savukārt apstrādātāja darbība, kas gaida sleep(), tiek pārtraukta, kad nostrādā signāls, un pēc tam atzīmēta kā pabeigta. Konfigurējiet taimautu globāli:

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

Uzdevumi, kas statusā processing iestrēguši ilgāk par QUEUETY_STALE_TIMEOUT sekundēm, tiek automātiski atjaunoti:

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

Novecojušos darbus var atkopt manuāli:

wp queuety recover

Rindas

Uzdevumi tiek nosūtīti uz nosauktām rindām. Katrs darbinieks apstrādā vienu rindu vai sakārtotu rindu sarakstu (skatiet Darbinieki):

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

wp queuety work --queue=emails

Noklusējuma rinda ir default. Varat palaist atsevišķus darbinieku procesus katrai rindai, lai nodalītu darba slodzes.

Ātruma ierobežošana

Ierobežojiet, cik bieži apstrādātājs var darboties noteiktā laika periodā:

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

Plašāku informāciju skatiet sadaļā Ātruma ierobežošana.

Nosūtāmas uzdevumu klases

Queuety v0.6.0 ievieš uz klasēm balstītus uzdevumus, izmantojot API. Tā vietā, lai atsevišķi norādītu apstrādātāja nosaukumu, datus un apstrādātāja klasi, definējiet vienu klasi, kas ietver visu: datus (kā tipizētas publiskās īpašības), izpildes loģiku un neobligāto starpprogrammatūru.

Interfeiss Contracts\Job

use Queuety\Contracts\Job;

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

Job klase glabā datus publiskās īpašībās un izpilda savu loģiku metodē handle(). Atsevišķa metode config() vai apstrādātāja reģistrācija nav nepieciešama.

Dispatchable traits

Pievienojiet Dispatchable trait, lai uzdevuma klasei būtu statiska metode 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 );
    }
}

Nosūtiet uzdevumu, padodot konstruktora argumentus:

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

Dispatchable traits iekšēji izsauc Queuety::dispatch(), tāpēc visas PendingJob opcijas (rinda, prioritāte, aizkave, max_attempts, unique, after) darbojas tieši tāpat.

Kā darbojas serializācija

Nosūtot Job instanci, Queuety JobSerializer iegūst pilno klases nosaukumu, ko izmanto kā apstrādātāju, un visas publiskās īpašības, ko izmanto kā datus. Darbinieka pusē serializētājs atjauno instanci, sasaistot datu atslēgas ar konstruktora parametru nosaukumiem. Enum vērtības ar pamatā esošu vērtību tiek serializētas uz šo vērtību un automātiski atjaunotas.

Uzdevumu rekvizīti

Uzdevumu klases var deklarēt publiskās īpašības, lai noteiktu atkārtotu mēģinājumu un taimauta darbību. Darbinieks tās nolasa, izmantojot refleksiju, un piemēro automātiski.

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

Masīvs $backoff nosaka pieaugošas aizkaves starp atkārtotiem mēģinājumiem. Piemēram, [10, 30, 60, 120] nozīmē, ka pirms pirmā atkārtotā mēģinājuma tiek gaidītas 10 sekundes, pirms otrā — 30 sekundes, pirms trešā — 60 sekundes, bet pirms katra nākamā mēģinājuma — 120 sekundes.

Ar resursiem saistītās īpašības noder, ja uzdevumu klases atbilst dažādiem izmaksu profiliem. Ātra kešatmiņas atjaunošana un liels izsaukums pakalpojumu sniedzējam abi var būt "viens uzdevums", taču tos nevajadzētu obligāti skaitīt vienādi, kad darbplūsmas budžets vai darbinieku piekļuves kontroles lēmums cenšas noturēt procesu drošās robežās.

Pilnu modeli, tostarp darbplūsmu budžetus un darbinieku puses piekļuves kontroli, skatiet sadaļā Resursu pārvaldība.

Kļūmes āķis

Kad pēc visu atkārtoto mēģinājumu izsmelšanas uzdevums tiek apglabāts, Queuety izsauc uzdevuma instances metodi failed(), ja tā ir definēta. Izmantojiet to uzkopšanai, brīdinājumu nosūtīšanai vai reģistrēšanai.

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

Metode failed() saņem izņēmumu, kas izraisīja pēdējo kļūmi. Uzdevuma instance tiek deserializēta ar tās sākotnējiem datiem, tāpēc ir pieejamas visas publiskās īpašības.

Nosacījuma izpilde

Dispatchable traits nodrošina dispatch_if() un dispatch_unless() nosacījuma izpildei. Abas metodes pieņem Būla tipa nosacījumu kā pirmo argumentu, kam seko uzdevuma konstruktora argumenti.

// 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() nosūta uzdevumu, ja nosacījums ir true, un atgriež PendingJob. Ja nosacījums ir false, tiek atgriezts null.

dispatch_unless() nosūta uzdevumu, ja nosacījums ir false, un atgriež PendingJob. Ja nosacījums ir true, tiek atgriezts null.

Abas metodes, nosūtot uzdevumu, atbalsta PendingJob opciju ķēdēšanu:

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

Sinhronā izpilde

Izmantojiet Queuety::dispatch_sync(), lai nekavējoties izpildītu uzdevumu pašreizējā procesā, nepievienojot to rindai. Tas noder testēšanai un uzdevumiem, kas jāizpilda uzreiz.

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

Uzdevuma metode handle() tiek izsaukta tieši. Netiek izveidota datubāzes rinda, netiek iesaistīts darbplūsmu izpildītājs un starpprogrammatūra netiek lietota.

Uzdevumu ķēdēšana

Uzdevumu ķēdēšana izpilda uzdevumu virkni citu pēc cita. Katra uzdevuma izpilde ir atkarīga no iepriekšējā uzdevuma sekmīgas pabeigšanas. Ja kāds uzdevums neizdodas, turpmākie ķēdes uzdevumi paliek gaidīšanas statusā.

Izmantojot 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() atgriež ChainBuilder ar metodēm rindas un kļūmes apstrādātāja iestatīšanai. Metode dispatch() izveido visus uzdevumus ar secīgām depends_on attiecībām un atgriež pirmā uzdevuma ID.

Izmantojot with_chain()

Dispatchable traits nodrošina metodi with_chain(). Pašlaik tā ir līdzvērtīga Queuety::chain( $jobs ): klase, kurai metode tiek izsaukta, ķēdei netiek pievienota (#13), tāpēc masīvā norādiet arī pirmo uzdevumu:

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

Ķēdes kļūmju apstrāde

Reģistrējiet kļūmju apstrādātāja klasi, izmantojot catch(). Apstrādātāja klasei jābūt metodei handle(). Tā tiek izsaukta, kad jebkurš ķēdes uzdevums tiek galīgi apglabāts. Darbinieks atbalsta handle(), handle( Job $job ) un handle( Job $job, \Throwable $exception ).

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

Starpprogrammatūras pievienošana

Uzdevumu klases var definēt metodi middleware(), kas atgriež izpildi aptverošu starpprogrammatūru:

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

Plašāku informāciju par iebūvēto starpprogrammatūru un pielāgotas starpprogrammatūras izveidi skatiet sadaļā Starpprogrammatūra.

Salīdzinājums ar apstrādātājos balstītiem uzdevumiem

Abas pieejas ir pilnībā atbalstītas. Esošie apstrādātājos balstītie uzdevumi turpina darboties bez izmaiņām. Vienā projektā varat izmantot abus stilus.

Uzdevuma dzīves cikls

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