Job

I job sono l'unità di lavoro fondamentale in Queuety. Inviate un job con il nome di un handler e un payload; un worker lo preleva ed esegue in background.

Invio dei job

use Queuety\Queuety;

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

dispatch() restituisce un builder PendingJob. Potete concatenare le opzioni prima che venga inviato automaticamente:

use Queuety\Enums\Priority;

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

Per ottenere subito l'ID del job, chiamate ->id():

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

Scrittura degli handler

Un handler è qualsiasi classe che implementa l'interfaccia 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',
        ];
    }
}

Il metodo handle() riceve il payload del job ed esegue il lavoro. Il metodo config() restituisce eventuali valori predefiniti:

Queste chiavi delle risorse fanno parte del più ampio modello Gestione delle risorse di Queuety, in cui i metadati dei job, i budget dei flussi di lavoro e l'ammissione dei worker operano insieme.

Registrate l'handler:

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

Oppure usate gli attributi PHP per la registrazione automatica.

Priorità

I job hanno quattro livelli di priorità. I worker acquisiscono per primi i job con priorità più alta.

use Queuety\Enums\Priority;

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

Ritardo

Ritardate un job in modo che diventi disponibile solo dopo il numero di secondi specificato:

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

Il job viene inserito immediatamente, ma i worker non lo acquisiscono finché non è trascorso il ritardo.

Job univoci

Impedite invii duplicati per lo stesso handler e payload. Se esiste già un job in attesa o in elaborazione con lo stesso handler e payload, viene restituito l'ID del job esistente invece di crearne uno nuovo.

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

Dipendenze tra job

Un job può dipendere da un altro job. Un worker non lo acquisirà finché la dipendenza non sarà stata completata.

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

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

Invio in batch

Inviate più job con un singolo INSERT su più righe per ottenere il massimo throughput:

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

Ogni elemento può includere chiavi facoltative: queue, priority, delay, max_attempts.

Nuovi tentativi e attesa progressiva

Quando un job genera un'eccezione, Queuety lo rimette in pending con un available_at ritardato, finché non è stato eseguito max_attempts volte in totale (si conta anche la prima esecuzione, quindi max_attempts( 3 ) consente fino a due nuovi tentativi). Il ritardo prima di ogni nuovo tentativo dipende dalla strategia di attesa progressiva, in cui attempt indica il numero dell'esecuzione appena fallita:

Impostate il valore predefinito globale in wp-config.php:

define( 'QUEUETY_RETRY_BACKOFF', 'linear' );

Oppure per handler in config():

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

Oppure per singolo invio:

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

Processi sepolti

Dopo aver esaurito tutti i tentativi, un job viene contrassegnato come messo in archivio. I job archiviati restano nel database per essere ispezionati, ma non vengono elaborati. Potete riprovarli manualmente. Un nuovo tentativo manuale rende di nuovo il job pending senza azzerarne il conteggio dei tentativi; pertanto, un job archiviato e riprovato viene eseguito un'altra volta e rimesso in archivio se anche quell'esecuzione fallisce:

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

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

Oppure tramite CLI:

wp queuety retry 42
wp queuety retry-buried

Timeout

Queuety attiva un pcntl_alarm() per la durata massima di esecuzione e dovrebbe far fallire il tentativo con un TimeoutException se il limite viene superato. Al momento il timeout non viene applicato (#10): il worker non abilita i segnali asincroni, quindi un handler di lunga durata viene eseguito fino al completamento, mentre un handler bloccato in sleep() viene interrotto quando scatta l'allarme e poi contrassegnato come completato. Configurate il timeout globale:

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

I job che restano nello stato processing per più di QUEUETY_STALE_TIMEOUT secondi vengono recuperati automaticamente:

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

Ripristinate manualmente i job obsoleti:

wp queuety recover

Code

I job vengono inviati a code denominate. Ogni worker elabora una coda oppure un elenco ordinato di code (consultate Worker):

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

wp queuety work --queue=emails

La coda predefinita è default. Potete eseguire processi worker separati per ogni coda, così da isolare i carichi di lavoro.

Limitazione della frequenza

Limitate la frequenza con cui un handler può essere eseguito nell'arco di un intervallo di tempo:

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

Per i dettagli, consultate Limitazione della frequenza.

Classi di job inviabili

Queuety v0.6.0 introduce una classe API per i job. Invece di separare il nome dell'handler, il payload e la classe dell'handler, definite un'unica classe che racchiude tutto: il payload (come proprietà pubbliche tipizzate), la logica di esecuzione e il middleware facoltativo.

L'interfaccia Contracts\Job

use Queuety\Contracts\Job;

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

Una classe Job contiene il proprio payload in proprietà pubbliche ed esegue la propria logica in handle(). Non è necessario un metodo config() separato né registrare un handler.

Il trait Dispatchable

Aggiungete il trait Dispatchable per ottenere un metodo statico dispatch() nella classe del job:

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

Invitatelo passando gli argomenti del costruttore:

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

Il trait Dispatchable chiama internamente Queuety::dispatch(), quindi tutte le opzioni PendingJob (queue, priority, delay, max_attempts, unique, after) funzionano esattamente allo stesso modo.

Come funziona la serializzazione

Quando inviate un'istanza di Job, Queuety estrae il nome completo della classe come handler e tutte le proprietà pubbliche come payload tramite il suo JobSerializer. Sul worker, il serializzatore ricostruisce l'istanza associando le chiavi del payload ai nomi dei parametri del costruttore. Gli enum con valore di supporto vengono serializzati nel valore sottostante e ripristinati automaticamente.

Proprietà del job

Le classi dei job possono dichiarare proprietà pubbliche per controllare il comportamento dei nuovi tentativi e dei timeout. Il worker le legge tramite reflection e le applica automaticamente.

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

L'array $backoff definisce ritardi crescenti tra i nuovi tentativi. Per esempio, [10, 30, 60, 120] indica che il primo nuovo tentativo attende 10 secondi, il secondo 30 secondi, il terzo 60 secondi e gli eventuali ulteriori tentativi attendono 120 secondi.

Le proprietà relative alle risorse sono utili quando le classi dei job rappresentano profili di costo diversi. Un aggiornamento rapido della cache e una chiamata di grandi dimensioni a un provider possono essere entrambi «un job», ma non dovrebbero necessariamente avere lo stesso peso quando un budget del flusso di lavoro o una decisione di ammissione dei worker mirano a mantenere il processo entro limiti sicuri.

Per il modello completo, compresi i budget dei flussi di lavoro e l'ammissione lato worker, consultate Gestione delle risorse.

Hook di errore

Quando un job viene messo in archivio dopo aver esaurito tutti i tentativi, Queuety chiama un metodo failed() sull'istanza del job, se definito. Usatelo per la pulizia, gli avvisi o la registrazione nei log.

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

Il metodo failed() riceve l'eccezione che ha causato l'errore finale. L'istanza del job viene deserializzata con il payload originale, quindi tutte le proprietà pubbliche sono disponibili.

Invio condizionale

Il trait Dispatchable fornisce dispatch_if() e dispatch_unless() per l'invio condizionale. Entrambi accettano come primo argomento una condizione booleana, seguita dagli argomenti del costruttore del job.

// 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() invia il job se la condizione è true e restituisce PendingJob. Se la condizione è false, restituisce null.

dispatch_unless() invia il job se la condizione è false e restituisce PendingJob. Se la condizione è true, restituisce null.

Entrambi i metodi supportano la concatenazione delle opzioni PendingJob al momento dell'invio:

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

Invio sincrono

Usate Queuety::dispatch_sync() per eseguire subito un job nel processo corrente senza aggiungerlo alla coda. È utile per i test e per i job che devono essere eseguiti inline.

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

Il metodo handle() del job viene chiamato direttamente. Non viene creata alcuna riga nel database, non interviene alcun worker e non si applica il middleware.

Concatenamento dei job

Il concatenamento dei job esegue una sequenza di job uno dopo l'altro. Ogni job della catena dipende dal completamento corretto del precedente. Se un job non riesce, i job successivi della catena restano in attesa.

Uso di 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() restituisce un ChainBuilder con metodi fluenti per impostare la coda e un handler degli errori. Il metodo dispatch() crea tutti i job con relazioni depends_on sequenziali e restituisce l'ID del primo job.

Uso di with_chain()

Il trait Dispatchable fornisce un metodo with_chain(). Al momento equivale a Queuety::chain( $jobs ): la classe su cui viene chiamato non viene aggiunta alla catena (#13), quindi includete anche il primo job nell'array:

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

Gestione degli errori della catena

Registrate una classe handler degli errori usando catch(). La classe handler deve avere un metodo handle(). Viene chiamato quando un qualsiasi job della catena viene definitivamente messo in archivio. Il worker supporta handle(), handle( Job $job ) e handle( Job $job, \Throwable $exception ).

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

Aggiunta di middleware

Le classi dei job possono definire un metodo middleware() per restituire middleware che avvolge l'esecuzione:

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

Per i dettagli sul middleware integrato e su come scriverne di personalizzato, consultate Middleware.

Confronto con i job basati su handler

Entrambi gli approcci sono pienamente supportati. I job esistenti basati su handler continuano a funzionare senza modifiche. Potete combinare entrambi gli stili nello stesso progetto.

Ciclo di vita dei job

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