Jobs

Las tareas son la unidad básica de trabajo en Queuety. Envíe una tarea con el nombre de un controlador y una carga útil; un worker la recoge y la ejecuta en segundo plano.

Envío de tareas

use Queuety\Queuety;

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

dispatch() devuelve un generador PendingJob. Puede encadenar opciones antes de que se envíe automáticamente:

use Queuety\Enums\Priority;

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

Para obtener el ID de la tarea de inmediato, llame a ->id():

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

Escritura de controladores

Un controlador puede ser cualquier clase que implemente la interfaz 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',
        ];
    }
}

El método handle() recibe la carga útil de la tarea y ejecuta el trabajo. El método config() devuelve valores predeterminados opcionales:

Estas claves de recursos forman parte del modelo más amplio de Resource Management de Queuety, donde los metadatos de las tareas, los presupuestos de flujos de trabajo y la admisión de workers funcionan en conjunto.

Registre el controlador:

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

O use los atributos PHP para el registro automático.

Prioridad

Los trabajos tienen cuatro niveles de prioridad. Los trabajadores reclaman primero los trabajos de mayor prioridad.

use Queuety\Enums\Priority;

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

Retraso

Retrase un trabajo para que esté disponible solo después del número de segundos especificado:

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

El trabajo se inserta de inmediato, pero los trabajadores no lo reclamarán hasta que haya transcurrido el retraso.

Trabajos únicos

Evite los envíos duplicados para el mismo controlador y la misma carga útil. Si ya existe un trabajo pendiente o en curso con el mismo controlador y la misma carga útil, se devuelve el ID del trabajo existente en lugar de crear uno nuevo.

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

Dependencias entre trabajos

Un trabajo puede depender de otro. Un trabajador no lo reclamará hasta que se haya completado la dependencia.

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

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

Envío por lotes

Envíe varios trabajos en una sola instrucción INSERT de varias filas para obtener el máximo rendimiento:

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

Cada elemento puede incluir claves opcionales: queue, priority, delay, max_attempts.

Reintentos y espera progresiva

Cuando un trabajo lanza una excepción, Queuety lo devuelve a pending con un available_at retrasado hasta que se haya ejecutado max_attempts veces en total (cuenta también la primera ejecución, por lo que max_attempts( 3 ) permite hasta dos reintentos). La estrategia de espera progresiva determina el retraso antes de cada reintento, donde attempt es el número de la ejecución que acaba de fallar:

Establezca el valor predeterminado global en wp-config.php:

define( 'QUEUETY_RETRY_BACKOFF', 'linear' );

O por controlador en config():

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

O por envío:

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

Trabajos en la cola de fallidos

Tras agotar todos los intentos de reintento, un trabajo se marca como enterrado. Los trabajos enterrados permanecen en la base de datos para su inspección, pero no se procesan. Puede reintentarlos manualmente. Un reintento manual vuelve a poner el trabajo en pending sin restablecer el recuento de intentos; así, un trabajo enterrado que se reintenta se ejecuta una vez más y vuelve a quedar enterrado si esa ejecución falla:

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

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

O mediante CLI:

wp queuety retry 42
wp queuety retry-buried

Tiempo de espera

Queuety activa una pcntl_alarm() para el tiempo máximo de ejecución y está pensado para hacer fallar el intento con un TimeoutException cuando se supera ese límite. Actualmente, ese tiempo de espera no se aplica (#10): el trabajador no habilita las señales asíncronas, por lo que un controlador de larga duración se ejecuta hasta el final, mientras que un controlador bloqueado en sleep() se interrumpe cuando suena la alarma y luego se marca como completado. Configure el tiempo de espera globalmente:

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

Los trabajos que permanecen en estado processing durante más de QUEUETY_STALE_TIMEOUT segundos se recuperan automáticamente:

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

Recupere manualmente los trabajos obsoletos:

wp queuety recover

Colas

Los trabajos se envían a colas con nombre. Cada trabajador procesa una cola o una lista ordenada de colas (consulte Trabajadores):

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

wp queuety work --queue=emails

La cola predeterminada es default. Puede ejecutar procesos de trabajador independientes para cada cola y así aislar las cargas de trabajo.

Limitación de frecuencia

Limite la frecuencia con la que puede ejecutarse un controlador dentro de un intervalo de tiempo:

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

Consulte Limitación de frecuencia para obtener más detalles.

Clases de trabajo enviables

Queuety v0.6.0 incorpora un trabajo basado en clases API. En lugar de separar el nombre del controlador, la carga útil y la clase del controlador, se define una sola clase que lo encapsula todo: la carga útil (como propiedades públicas tipadas), la lógica de ejecución y el middleware opcional.

La interfaz Contracts\Job

use Queuety\Contracts\Job;

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

Una clase Job almacena su carga útil en propiedades públicas y ejecuta su propia lógica en handle(). No se requiere un método config() ni registrar un controlador por separado.

El rasgo Dispatchable

Añada el rasgo Dispatchable para obtener un método estático dispatch() en su clase de trabajo:

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

Envíelo pasando los argumentos del constructor:

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

El rasgo Dispatchable llama internamente a Queuety::dispatch(), por lo que todas las opciones de PendingJob (cola, prioridad, retraso, max_attempts, unique, after) funcionan exactamente igual.

Cómo funciona la serialización

Al enviar una instancia de Job, el Queuety de JobSerializer extrae el nombre completo de la clase como controlador y todas las propiedades públicas como carga útil. En el lado del trabajador, el serializador reconstruye la instancia haciendo coincidir las claves de la carga útil con los nombres de los parámetros del constructor. Los enumerados con valor asociado se serializan con su valor subyacente y se restauran automáticamente.

Propiedades de las tareas

Las clases de trabajo pueden declarar propiedades públicas para controlar los reintentos y el tiempo de espera. El trabajador las lee mediante reflexión y las aplica automáticamente.

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

La matriz $backoff proporciona retrasos de reintento progresivos. Por ejemplo, [10, 30, 60, 120] significa que el primer reintento espera 10 segundos, el segundo 30 segundos, el tercero 60 segundos y los siguientes, 120 segundos.

Las propiedades relacionadas con los recursos son útiles cuando las clases de trabajo representan distintos perfiles de coste. Una actualización rápida de la caché y una llamada extensa a un proveedor pueden ser «un trabajo» cada una, pero no deberían contabilizarse necesariamente igual cuando un presupuesto de flujo de trabajo o una decisión de admisión de trabajadores intenta mantener el proceso dentro de unos límites seguros.

Consulte Gestión de recursos para conocer el modelo completo, incluidos los presupuestos de flujo de trabajo y la admisión en el lado del trabajador.

Hook de fallo

Cuando un trabajo queda enterrado tras agotar todos los intentos de reintento, Queuety llama a un método failed() en la instancia del trabajo, si está definido. Úselo para tareas de limpieza, alertas o registro.

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

El método failed() recibe la excepción que causó el fallo definitivo. La instancia del trabajo se deserializa con su carga útil original, por lo que están disponibles todas las propiedades públicas.

Envío condicional

El rasgo Dispatchable proporciona dispatch_if() y dispatch_unless() para realizar envíos condicionales. Ambos aceptan una condición booleana como primer argumento, seguida de los argumentos del constructor del trabajo.

// 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() realiza el envío cuando la condición es true y devuelve PendingJob. Devuelve null cuando la condición es false.

dispatch_unless() realiza el envío cuando la condición es false y devuelve PendingJob. Devuelve null cuando la condición es true.

Ambos métodos admiten opciones PendingJob encadenadas cuando realizan el envío:

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

Envío síncrono

Use Queuety::dispatch_sync() para ejecutar un trabajo de inmediato en el proceso actual, sin añadirlo a la cola. Es útil para hacer pruebas y para trabajos que deben ejecutarse en línea.

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

Se llama directamente al método handle() del trabajo. No se crea ninguna fila en la base de datos, no interviene ningún worker ni se aplica middleware.

Encadenamiento de trabajos

El encadenamiento de trabajos ejecuta una serie de trabajos uno tras otro. Cada trabajo de la cadena depende de que el anterior se complete correctamente. Si algún trabajo falla, los siguientes de la cadena quedan pendientes.

Uso de 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() devuelve un ChainBuilder con métodos encadenables para establecer la cola y un controlador de fallos. El método dispatch() crea todos los trabajos con relaciones depends_on secuenciales y devuelve el ID del primer trabajo.

Uso de with_chain()

El rasgo Dispatchable proporciona un método with_chain(). Actualmente, equivale a Queuety::chain( $jobs ): la clase desde la que se llama no se añade a la cadena (#13), así que incluya también el primer trabajo en la matriz:

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

Gestión de fallos en la cadena

Registre una clase controladora de fallos mediante catch(). La clase controladora debe tener un método handle(). Se llama cuando cualquier trabajo de la cadena queda enterrado de forma permanente. El trabajador admite handle(), handle( Job $job ) y handle( Job $job, \Throwable $exception ).

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

Añadir middleware

Las clases de trabajo pueden definir un método middleware() que devuelva middleware para envolver la ejecución:

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

Consulte Middleware para obtener más información sobre el middleware integrado y cómo escribir middleware personalizado.

Comparación con trabajos basados en controladores

Ambos enfoques son totalmente compatibles. Los trabajos basados en controladores existentes siguen funcionando sin cambios. Puede combinar ambos estilos en el mismo proyecto.

Ciclo de vida del trabajo

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