Sarcini

Sarcinile sunt unitatea fundamentală de lucru în Queuety. Lansați o sarcină cu numele unui gestionar și datele utile, iar un lucrător o preia și o execută în fundal.

Lansarea sarcinilor

use Queuety\Queuety;

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

dispatch() returnează un constructor PendingJob. Puteți înlănțui opțiuni înainte de lansarea automată:

use Queuety\Enums\Priority;

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

Pentru a obține imediat ID-ul sarcinii, apelați ->id():

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

Scrierea gestionarilor

Un gestionar este orice clasă care implementează interfața 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() primește datele utile ale sarcinii și efectuează lucrarea. Metoda config() returnează valori implicite opționale:

Aceste chei de resurse fac parte din modelul mai amplu Queuety de Gestionarea resurselor, în care metadatele sarcinilor, bugetele fluxurilor de lucru și admiterea lucrătorilor funcționează împreună.

Înregistrați gestionarul:

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

Sau folosiți atributele PHP pentru înregistrare automată.

Prioritate

Joburile au patru niveluri de prioritate. Joburile cu prioritate mai mare sunt preluate primele de către workeri.

use Queuety\Enums\Priority;

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

Întârziere

Întârziați un job astfel încât să devină disponibil numai după numărul specificat de secunde:

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

Jobul este inserat imediat, dar workerii nu îl vor prelua până la expirarea întârzierii.

Joburi unice

Preveniți trimiterea duplicată pentru același handler și același payload. Dacă există deja un job în așteptare sau în curs de procesare cu același handler și payload, este returnat ID-ul jobului existent în loc să fie creat unul nou.

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

Dependențe între joburi

Un job poate depinde de un alt job. Un worker nu îl va prelua până când dependența nu este finalizată.

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

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

Trimiterea în lot

Trimiteți mai multe joburi printr-o singură instrucțiune INSERT cu mai multe rânduri, pentru un debit maxim:

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

Fiecare element poate include chei opționale: queue, priority, delay, max_attempts.

Reîncercare și întârziere progresivă

Când un job aruncă o excepție, Queuety îl readuce în pending cu o available_at întârziată, până când a fost executat de max_attempts ori în total (prima execuție se numără, așadar max_attempts( 3 ) înseamnă cel mult două reîncercări). Întârzierea dinaintea fiecărei reîncercări este stabilită de strategia de întârziere progresivă, unde attempt este numărul execuției tocmai eșuate:

Setați valoarea implicită globală în wp-config.php:

define( 'QUEUETY_RETRY_BACKOFF', 'linear' );

Sau per handler în config():

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

Sau la fiecare trimitere:

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

Sarcini în carantină

După epuizarea tuturor încercărilor de reîncercare, un job este marcat ca îngropat. Joburile îngropate rămân în baza de date pentru examinare, dar nu sunt procesate. Le puteți reîncerca manual. O reîncercare manuală face jobul să fie din nou pending fără a reseta numărul de încercări, astfel că un job îngropat reîncercat mai primește o execuție și este îngropat din nou dacă aceasta eșuează:

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

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

Sau prin CLI:

wp queuety retry 42
wp queuety retry-buried

Expirarea timpului

Queuety pornește o pcntl_alarm() pentru timpul maxim de execuție și ar trebui să facă încercarea să eșueze cu o TimeoutException dacă limita este depășită. Această limită nu este aplicată momentan (#10): workerul nu activează semnalele asincrone, așadar un handler cu execuție îndelungată rulează până la capăt, iar un handler blocat în sleep() este oprit când se declanșează alarma și apoi marcat ca finalizat. Configurați limita global:

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

Joburile care rămân blocate în starea processing mai mult de QUEUETY_STALE_TIMEOUT secunde sunt recuperate automat:

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

Recuperați manual joburile expirate:

wp queuety recover

Cozi

Joburile sunt trimise în cozi cu nume. Fiecare worker procesează o singură coadă sau o listă ordonată de cozi (consultați Workeri):

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

wp queuety work --queue=emails

Coada implicită este default. Puteți rula procese worker separate pentru fiecare coadă, pentru a izola sarcinile.

Limitarea ratei

Limitați frecvența cu care poate fi executat un handler într-un interval de timp:

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

Consultați Limitarea ratei pentru detalii.

Clase de joburi care pot fi trimise

Queuety v0.6.0 introduce o API de joburi bazată pe clase. În loc să separați numele handlerului, payload-ul și clasa handlerului, definiți o singură clasă care încapsulează totul: payload-ul (sub formă de proprietăți publice tipizate), logica de execuție și middleware-ul opțional.

Interfața Contracts\Job

use Queuety\Contracts\Job;

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

O clasă Job păstrează payload-ul în proprietăți publice și își execută propria logică în handle(). Nu este necesară o metodă config() separată sau înregistrarea handlerului.

Trait-ul Dispatchable

Adăugați trait-ul Dispatchable pentru a obține o metodă statică dispatch() în clasa jobului:

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

Trimiteți-l pasând argumentele constructorului:

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

Trait-ul Dispatchable apelează intern Queuety::dispatch(), astfel că toate opțiunile PendingJob (queue, priority, delay, max_attempts, unique, after) funcționează la fel.

Cum funcționează serializarea

Când trimiteți o instanță Job, Queuety extrage din JobSerializer numele complet calificat al clasei, pe care îl folosește ca handler, și toate proprietățile publice, pe care le folosește ca payload. Pe partea workerului, serializatorul reconstruiește instanța asociind cheile payload-ului cu numele parametrilor constructorului. Enumerațiile cu valori asociate sunt serializate în valoarea de bază și restaurate automat.

Proprietățile sarcinii

Clasele joburilor pot declara proprietăți publice pentru a controla comportamentul de reîncercare și de expirare. Workerul le citește prin reflecție și le aplică automat.

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

Tabloul $backoff oferă întârzieri progresive între reîncercări. De exemplu, [10, 30, 60, 120] înseamnă că prima reîncercare așteaptă 10 secunde, a doua 30 de secunde, a treia 60 de secunde, iar orice reîncercare ulterioară așteaptă 120 secunde.

Proprietățile legate de resurse sunt utile când clasele joburilor reprezintă profiluri de cost diferite. O reîmprospătare rapidă a cache-ului și un apel amplu către un furnizor pot fi ambele „un job”, dar nu ar trebui neapărat să fie contabilizate la fel când un buget al fluxului de lucru sau o decizie de admitere a workerilor încearcă să mențină procesul în limite sigure.

Pentru modelul complet, inclusiv bugetele fluxurilor de lucru și admiterea pe partea workerului, consultați Gestionarea resurselor.

Hook la eșec

Când un job este îngropat după epuizarea tuturor încercărilor de reîncercare, Queuety apelează o metodă failed() pe instanța jobului, dacă este definită. Folosiți-o pentru curățare, alerte sau înregistrarea în jurnal.

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() primește excepția care a cauzat eșecul final. Instanța jobului este deserializată cu payload-ul inițial, astfel încât toate proprietățile publice sunt disponibile.

Trimitere condiționată

Trait-ul Dispatchable oferă dispatch_if() și dispatch_unless() pentru trimitere condiționată. Ambele acceptă o condiție booleană ca prim argument, urmată de argumentele constructorului jobului.

// 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() trimite jobul când condiția este true și returnează PendingJob. Returnează null când condiția este false.

dispatch_unless() trimite jobul când condiția este false și returnează PendingJob. Returnează null când condiția este true.

Ambele metode acceptă înlănțuirea opțiunilor PendingJob atunci când trimit joburi:

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

Trimitere sincronă

Folosiți Queuety::dispatch_sync() pentru a executa un job imediat în procesul curent, fără a-l adăuga în coadă. Este util pentru testare și pentru joburile care trebuie executate pe loc.

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() a sarcinii este apelată direct. Nu se creează niciun rând în baza de date, nu este implicat niciun worker și nu se aplică middleware.

Înșiruirea joburilor

Înșiruirea joburilor execută o succesiune de joburi, unul după altul. Fiecare job din înșiruire depinde de finalizarea cu succes a jobului precedent. Dacă un job eșuează, joburile următoare din înșiruire rămân în așteptare.

Folosind 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() returnează un ChainBuilder cu metode fluente pentru setarea cozii și a unui handler de eșec. Metoda dispatch() creează toate joburile cu relații depends_on secvențiale și returnează ID-ul primului job.

Folosind with_chain()

Trait-ul Dispatchable oferă o metodă with_chain(). Momentan, aceasta este echivalentă cu Queuety::chain( $jobs ): clasa pe care este apelată nu este adăugată în înșiruire (#13), așadar includeți și primul job din tablou:

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

Gestionarea eșecurilor din înșiruire

Înregistrați o clasă de gestionare a eșecurilor folosind catch(). Clasa handlerului trebuie să aibă o metodă handle(). Aceasta este apelată când orice job din înșiruire este îngropat definitiv. Workerul acceptă handle(), handle( Job $job ) și handle( Job $job, \Throwable $exception ).

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

Adăugarea middleware-ului

Clasele joburilor pot defini o metodă middleware() care returnează middleware-ul ce încadrează execuția:

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

Consultați Middleware pentru detalii despre middleware-ul integrat și scrierea unui middleware personalizat.

Comparație cu joburile bazate pe handler

Ambele abordări sunt acceptate integral. Joburile existente bazate pe handler continuă să funcționeze fără modificări. Puteți combina ambele stiluri în același proiect.

Ciclul de viață al unui job

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