Tarefas

As tarefas são a unidade central de trabalho em Queuety. Despache uma tarefa com o nome de um manipulador e uma carga útil; um worker a coleta e a executa em segundo plano.

Despachando tarefas

use Queuety\Queuety;

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

dispatch() retorna um construtor PendingJob. Você pode encadear opções antes do despacho automático:

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 obter o ID da tarefa imediatamente, chame ->id():

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

Escrevendo manipuladores

Um manipulador é qualquer classe que implementa a interface 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',
        ];
    }
}

O método handle() recebe a carga útil da tarefa e executa o trabalho. O método config() retorna valores padrão opcionais:

Essas chaves de recursos fazem parte do modelo mais amplo de QueuetyGerenciamento de recursos, em que metadados de tarefas, orçamentos de fluxos de trabalho e admissão de workers atuam em conjunto.

Registre o manipulador:

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

Ou use os atributos PHP para registro automático.

Prioridade

As tarefas têm quatro níveis de prioridade. Os workers obtêm primeiro as tarefas de prioridade mais alta.

use Queuety\Enums\Priority;

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

Atraso

Atrase uma tarefa para que fique disponível somente após o número de segundos especificado:

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

A tarefa é inserida imediatamente, mas os workers só a obterão quando o atraso terminar.

Tarefas únicas

Evite despachos duplicados para o mesmo manipulador e carga útil. Se já existir uma tarefa pendente ou em processamento com o mesmo manipulador e a mesma carga útil, o ID da tarefa existente será retornado em vez de criar uma nova.

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

Dependências entre tarefas

Uma tarefa pode depender de outra. Um worker só a obterá quando a dependência for concluída.

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

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

Despacho em lote

Despache várias tarefas em um único INSERT de várias linhas para obter o máximo de processamento:

$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 item pode incluir chaves opcionais: queue, priority, delay, max_attempts.

Novas tentativas e espera progressiva

Quando uma tarefa lança uma exceção, Queuety a devolve a pending com um available_at atrasado até que tenha sido executada max_attempts vezes no total (a primeira execução conta, portanto, max_attempts( 3 ) significa até duas novas tentativas). O atraso antes de cada nova tentativa é determinado pela estratégia de espera progressiva, em que attempt é o número da execução que acabou de falhar:

Defina o padrão global em wp-config.php:

define( 'QUEUETY_RETRY_BACKOFF', 'linear' );

Ou por manipulador em config():

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

Ou por despacho:

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

Jobs em espera

Depois de esgotar todas as tentativas, uma tarefa é marcada como arquivada. As tarefas arquivadas permanecem no banco de dados para inspeção, mas não são processadas. Você pode tentar executá-las novamente manualmente. Uma nova tentativa manual torna a tarefa pending novamente, sem redefinir a contagem de tentativas; assim, uma tarefa arquivada que seja tentada novamente terá mais uma execução e será arquivada de novo se essa execução falhar:

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

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

Ou por meio de CLI:

wp queuety retry 42
wp queuety retry-buried

Timeout

Queuety ativa um pcntl_alarm() pelo tempo máximo de execução e deve fazer a tentativa falhar com um TimeoutException quando esse limite for excedido. Esse tempo limite não está sendo aplicado no momento (#10): o worker não habilita sinais assíncronos, portanto, um manipulador de longa duração executa até o fim, enquanto um manipulador bloqueado em sleep() é interrompido quando o alarme dispara e, em seguida, marcado como concluído. Configure o tempo limite globalmente:

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

Tarefas que permanecem no status processing por mais de QUEUETY_STALE_TIMEOUT segundos são recuperadas automaticamente:

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

Recupere tarefas obsoletas manualmente:

wp queuety recover

Filas

As tarefas são despachadas para filas nomeadas. Cada worker processa uma fila ou uma lista ordenada de filas (veja Workers):

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

wp queuety work --queue=emails

A fila padrão é default. Você pode executar processos de worker separados para cada fila e isolar as cargas de trabalho.

Limitação de taxa

Limite a frequência com que um manipulador pode ser executado dentro de um intervalo de tempo:

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

Veja Limitação de taxa para obter detalhes.

Classes de tarefas despacháveis

Queuety v0.6.0 introduz uma API de tarefa baseada em classes. Em vez de separar o nome do manipulador, a carga útil e a classe do manipulador, você define uma única classe que encapsula tudo: a carga útil (como propriedades públicas tipadas), a lógica de execução e o middleware opcional.

A interface Contracts\Job

use Queuety\Contracts\Job;

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

Uma classe Job armazena sua carga útil como propriedades públicas e executa sua própria lógica em handle(). Não é necessário um método config() separado nem registrar um manipulador.

A trait Dispatchable

Adicione a trait Dispatchable para obter um método estático dispatch() na classe da tarefa:

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

Despache-a passando os argumentos do construtor:

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

A trait Dispatchable chama internamente Queuety::dispatch(), portanto, todas as opções de PendingJob (fila, prioridade, atraso, max_attempts, unique, after) funcionam exatamente da mesma forma.

Como funciona a serialização

Ao despachar uma instância de Job, o método JobSerializer de Queuety extrai o nome totalmente qualificado da classe como manipulador e todas as propriedades públicas como carga útil. No worker, o serializador reconstrói a instância associando as chaves da carga útil aos nomes dos parâmetros do construtor. Enums com valor de suporte são serializados como o valor subjacente e restaurados automaticamente.

Propriedades do trabalho

As classes de tarefas podem declarar propriedades públicas para controlar o comportamento de novas tentativas e do tempo limite. O worker as lê por reflexão e as aplica 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...
    }
}

O array $backoff fornece atrasos progressivamente maiores entre tentativas. Por exemplo, [10, 30, 60, 120] significa que a primeira nova tentativa aguarda 10 segundos, a segunda aguarda 30 segundos, a terceira aguarda 60 segundos e as tentativas seguintes aguardam 120 segundos.

As propriedades relacionadas a recursos são úteis quando as classes de tarefas representam diferentes perfis de custo. Uma atualização rápida de cache e uma chamada grande a um provedor podem ser, ambas, "uma tarefa", mas não devem necessariamente ter o mesmo peso quando um orçamento de fluxo de trabalho ou uma decisão de admissão de workers tenta manter o processo dentro de limites seguros.

Para ver o modelo completo, incluindo orçamentos de fluxo de trabalho e admissão no worker, consulte Gerenciamento de recursos.

Hook de falha

Quando uma tarefa é arquivada após esgotar todas as tentativas, Queuety chama um método failed() na instância da tarefa, se houver. Use-o para limpeza, alertas ou registros.

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

O método failed() recebe a exceção que causou a falha final. A instância da tarefa é desserializada com a carga útil original, portanto, todas as propriedades públicas ficam disponíveis.

Despacho condicional

A trait Dispatchable fornece dispatch_if() e dispatch_unless() para despacho condicional. Ambos aceitam uma condição booleana como primeiro argumento, seguida pelos argumentos do construtor da tarefa.

// 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() despacha quando a condição é true e retorna PendingJob. Retorna null quando a condição é false.

dispatch_unless() despacha quando a condição é false e retorna PendingJob. Retorna null quando a condição é true.

Ambos os métodos permitem encadear opções de PendingJob no momento do despacho:

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

Despacho síncrono

Use Queuety::dispatch_sync() para executar uma tarefa imediatamente no processo atual, sem adicioná-la à fila. Isso é útil para testes e para tarefas que precisam ser executadas em linha.

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

O método handle() da tarefa é chamado diretamente. Nenhuma linha é criada no banco de dados, nenhum worker é envolvido e o middleware não é aplicado.

Encadeamento de tarefas

O encadeamento de tarefas executa uma sequência, uma após a outra. Cada tarefa da cadeia depende da conclusão bem-sucedida da anterior. Se alguma tarefa falhar, as tarefas seguintes permanecem pendentes.

Usando 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() retorna um ChainBuilder com métodos encadeáveis para definir a fila e um manipulador de falhas. O método dispatch() cria todas as tarefas com relações depends_on sequenciais e retorna o ID da primeira tarefa.

Usando with_chain()

A trait Dispatchable fornece um método with_chain(). No momento, ele equivale a Queuety::chain( $jobs ): a classe em que é chamado não é adicionada à cadeia (#13), portanto, inclua também na lista a primeira tarefa do array:

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

Tratamento de falhas na cadeia

Registre uma classe de manipulador de falhas usando catch(). A classe do manipulador deve ter um método handle(). Ele é chamado quando qualquer tarefa da cadeia é arquivada permanentemente. O worker oferece suporte a handle(), handle( Job $job ) e handle( Job $job, \Throwable $exception ).

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

Adicionando middleware

As classes de tarefas podem definir um método middleware() para retornar middleware que envolve a execução:

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

Veja Middleware para obter detalhes sobre o middleware integrado e como criar middleware personalizado.

Comparação com tarefas baseadas em manipuladores

As duas abordagens têm suporte completo. Os jobs existentes baseados em handlers continuam funcionando sem alterações. Você pode misturar os dois estilos no mesmo projeto.

Ciclo de vida dos jobs

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