Zadania

Zadania to podstawowa jednostka pracy w Queuety. Zleć zadanie, podając nazwę procedury obsługi i dane, a proces roboczy pobierze je i wykona w tle.

Zlecanie zadań

use Queuety\Queuety;

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

dispatch() zwraca obiekt konstruktora PendingJob. Możesz łączyć opcje przed automatycznym zleceniem zadania:

use Queuety\Enums\Priority;

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

Aby od razu pobrać identyfikator zadania, wywołaj ->id():

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

Tworzenie procedur obsługi

Procedurą obsługi może być dowolna klasa implementująca interfejs 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() otrzymuje dane zadania i wykonuje pracę. Metoda config() zwraca opcjonalne wartości domyślne:

Te klucze zasobów są częścią szerszego modelu Zarządzania zasobami w Queuety, w którym współdziałają metadane zadań, budżety przepływów pracy i przyjmowanie zadań przez procesy robocze.

Zarejestruj procedurę obsługi:

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

Lub użyj atrybutów PHP do automatycznej rejestracji.

Priorytet

Zadania mają cztery poziomy priorytetu. Pracownicy najpierw przejmują zadania o wyższym priorytecie.

use Queuety\Enums\Priority;

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

Opóźnienie

Opóźnij zadanie, aby stało się dostępne dopiero po upływie określonej liczby sekund:

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

Zadanie jest dodawane natychmiast, ale pracownicy nie przejmą go, dopóki nie upłynie czas opóźnienia.

Unikatowe zadania

Zapobiegaj wielokrotnemu wysyłaniu tego samego zadania z tym samym programem obsługi i ładunkiem. Jeśli istnieje już oczekujące lub przetwarzane zadanie z tym samym programem obsługi i ładunkiem, zwracany jest identyfikator istniejącego zadania zamiast tworzenia nowego.

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

Zależności między zadaniami

Zadanie może zależeć od innego zadania. Pracownik nie przejmie go, dopóki zadanie, od którego zależy, nie zostanie ukończone.

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

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

Wysyłanie partii zadań

Wysyłaj wiele zadań w pojedynczej instrukcji INSERT obejmującej wiele wierszy, aby osiągnąć maksymalną przepustowość:

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

Każdy element może zawierać opcjonalne klucze: queue, priority, delay, max_attempts.

Ponawianie prób i opóźnienia

Gdy zadanie zgłosi wyjątek, Queuety dodaje je z powrotem do pending z opóźnionym available_at, aż zostanie uruchomione łącznie max_attempts razy (liczy się także pierwsze uruchomienie, więc max_attempts( 3 ) oznacza maksymalnie dwie ponowne próby). Opóźnienie przed każdą ponowną próbą określa strategia opóźnień, w której attempt oznacza numer właśnie nieudanego uruchomienia:

Ustaw globalną wartość domyślną w wp-config.php:

define( 'QUEUETY_RETRY_BACKOFF', 'linear' );

Lub dla danego programu obsługi w config():

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

Lub przy wysyłaniu zadania:

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

Zadania w archiwum

Po wyczerpaniu wszystkich prób ponowienia zadanie otrzymuje status zagrzebane. Takie zadania pozostają w bazie danych do wglądu, ale nie są przetwarzane. Można ponowić je ręcznie. Ręczne ponowienie zmienia status zadania na pending, ale nie zeruje licznika prób, więc ponowione zagrzebane zadanie zostanie uruchomione jeszcze raz i ponownie zagrzebane, jeśli ta próba się nie powiedzie:

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

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

Lub przez CLI:

wp queuety retry 42
wp queuety retry-buried

Limit czasu

Queuety ustawia pcntl_alarm() na maksymalny czas wykonania; po jego przekroczeniu ma zakończyć próbę niepowodzeniem z TimeoutException. Obecnie limit czasu nie jest egzekwowany (#10): pracownik nie włącza sygnałów asynchronicznych, więc długotrwały program obsługi kończy działanie, a program obsługi zablokowany w sleep() zostaje przerwany po wywołaniu alarmu i oznaczony jako ukończony. Skonfiguruj limit czasu globalnie:

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

Zadania, które pozostają w stanie processing dłużej niż QUEUETY_STALE_TIMEOUT sekund, są automatycznie odzyskiwane:

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

Odzyskaj nieaktualne zadania ręcznie:

wp queuety recover

Kolejki

Zadania są wysyłane do nazwanych kolejek. Każdy pracownik przetwarza jedną kolejkę lub uporządkowaną listę kolejek (zobacz Pracownicy):

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

wp queuety work --queue=emails

Domyślna kolejka to default. Możesz uruchomić osobne procesy pracowników dla każdej kolejki, aby odizolować obciążenia.

Ograniczanie częstotliwości

Ogranicz częstotliwość wykonywania programu obsługi w określonym przedziale czasu:

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

Szczegóły znajdziesz w sekcji Ograniczanie częstotliwości.

Klasy zadań do wysłania

Queuety v0.6.0 wprowadza klasę zadania API. Zamiast oddzielać nazwę programu obsługi, ładunek i klasę programu obsługi, definiujesz jedną klasę, która obejmuje wszystko: ładunek (w postaci typowanych właściwości publicznych), logikę wykonania i opcjonalne oprogramowanie pośredniczące.

Interfejs Contracts\Job

use Queuety\Contracts\Job;

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

Klasa Job przechowuje ładunek we właściwościach publicznych i wykonuje własną logikę w handle(). Nie jest wymagane osobne wywołanie metody config() ani rejestrowanie programu obsługi.

Cecha Dispatchable

Dodaj cechę Dispatchable, aby udostępnić statyczną metodę dispatch() w klasie zadania:

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

Wyślij je, przekazując argumenty konstruktora:

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

Cecha Dispatchable wewnętrznie wywołuje Queuety::dispatch(), więc wszystkie opcje PendingJob (kolejka, priorytet, opóźnienie, max_attempts, unique, after) działają tak samo.

Jak działa serializacja

Gdy wysyłasz instancję Job, Queuety funkcji JobSerializer pobiera w pełni kwalifikowaną nazwę klasy jako program obsługi, a wszystkie właściwości publiczne jako ładunek. Po stronie pracownika serializator odtwarza instancję, dopasowując klucze ładunku do nazw parametrów konstruktora. Enumy z wartościami bazowymi są serializowane do ich wartości bazowej i automatycznie odtwarzane.

Właściwości zadania

Klasy zadań mogą deklarować właściwości publiczne, aby kontrolować ponawianie prób i zachowanie limitu czasu. Pracownik odczytuje je przez refleksję i automatycznie je stosuje.

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

Tablica $backoff określa rosnące opóźnienia między ponownymi próbami. Na przykład [10, 30, 60, 120] oznacza, że pierwsza ponowna próba następuje po 10 sekundach, druga po 30 sekundach, trzecia po 60 sekundach, a każda kolejna po 120 sekundach.

Właściwości związane z zasobami są przydatne, gdy klasy zadań reprezentują różne profile kosztów. Szybkie odświeżenie pamięci podręcznej i duże wywołanie dostawcy mogą być „jednym zadaniem”, ale niekoniecznie powinny być liczone tak samo, gdy budżet przepływu pracy lub decyzja o dopuszczeniu zadania przez pracownika ma utrzymać proces w bezpiecznych granicach.

Pełny model, obejmujący budżety przepływu pracy i dopuszczanie zadań po stronie pracownika, opisano w sekcji Zarządzanie zasobami.

Procedura wywoływana po niepowodzeniu

Gdy zadanie zostaje zagrzebane po wyczerpaniu wszystkich prób, Queuety wywołuje metodę failed() na instancji zadania, jeśli została zdefiniowana. Użyj jej do sprzątania, wysyłania alertów lub rejestrowania zdarzeń.

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() otrzymuje wyjątek, który spowodował ostateczne niepowodzenie. Instancja zadania jest deserializowana z pierwotnym ładunkiem, więc wszystkie właściwości publiczne są dostępne.

Warunkowe wysyłanie

Cecha Dispatchable udostępnia metody dispatch_if() i dispatch_unless() do warunkowego wysyłania zadań. Obie przyjmują warunek logiczny jako pierwszy argument, a następnie argumenty konstruktora zadania.

// 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() wysyła zadanie, gdy warunek jest true, i zwraca PendingJob. Gdy warunek jest false, zwraca null.

dispatch_unless() wysyła zadanie, gdy warunek jest false, i zwraca PendingJob. Gdy warunek jest true, zwraca null.

Obie metody obsługują łańcuchowe dodawanie opcji PendingJob podczas wysyłania:

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

Wysyłanie synchroniczne

Użyj Queuety::dispatch_sync(), aby natychmiast wykonać zadanie w bieżącym procesie bez dodawania go do kolejki. Przydaje się to podczas testowania oraz w przypadku zadań, które muszą być wykonywane w linii.

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() zadania jest wywoływana bezpośrednio. Nie jest tworzony wiersz w bazie danych, nie uczestniczy w tym żaden proces roboczy, a oprogramowanie pośredniczące nie jest stosowane.

Łączenie zadań

Łączenie zadań umożliwia wykonywanie ich kolejno, jedno po drugim. Każde zadanie w łańcuchu zależy od pomyślnego ukończenia poprzedniego. Jeśli którekolwiek zadanie się nie powiedzie, kolejne pozostają w oczekiwaniu.

Używanie 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() zwraca obiekt ChainBuilder z metodami łańcuchowymi do ustawiania kolejki i procedury obsługi błędów. Metoda dispatch() tworzy wszystkie zadania z sekwencyjnymi relacjami depends_on i zwraca identyfikator pierwszego zadania.

Używanie with_chain()

Cecha Dispatchable udostępnia metodę with_chain(). Obecnie jej działanie jest równoważne Queuety::chain( $jobs ): klasa, na której ją wywołano, nie jest dodawana do łańcucha (#13), więc uwzględnij w tablicy także pierwsze zadanie:

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

Obsługa błędów w łańcuchu

Zarejestruj klasę obsługującą błędy za pomocą catch(). Klasa obsługi musi mieć metodę handle(). Jest ona wywoływana, gdy dowolne zadanie w łańcuchu zostanie trwale zagrzebane. Pracownik obsługuje handle(), handle( Job $job ) i handle( Job $job, \Throwable $exception ).

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

Dodawanie oprogramowania pośredniczącego

Klasy zadań mogą definiować metodę middleware(), która zwraca oprogramowanie pośredniczące opakowujące wykonanie:

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

Szczegóły dotyczące wbudowanego oprogramowania pośredniczącego i tworzenia własnego znajdziesz w sekcji Oprogramowanie pośredniczące.

Porównanie z zadaniami opartymi na programach obsługi

Oba podejścia są w pełni obsługiwane. Istniejące zadania oparte na programach obsługi nadal działają bez zmian. Możesz łączyć oba style w tym samym projekcie.

Cykl życia zadania

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