Filas e Jobs Assíncronos Já leu

Algumas operações não precisam — e não devem — acontecer na requisição HTTP: enviar email, gerar PDF, processar imagem, notificar sistemas externos. Cada milissegundo que você gasta nessas tarefas é milissegundo que o usuário espera.

Filas resolvem isso: a requisição enfileira o trabalho e retorna imediatamente. Um processo separado (worker) consome a fila em background e executa os jobs. O resultado: respostas HTTP em menos de 50ms mesmo quando a operação real leva minutos.

Neste artigo implementamos uma fila com Redis (lista circular), um sistema de jobs com retry automático, dead-letter queue para falhas persistentes, e um worker robusto com tratamento de sinais do sistema operacional.

Estrutura de uma fila com Redis

Redis é a escolha mais comum para filas PHP por três motivos: operações atômicas com LPUSH/BRPOP, TTL nativo para expiração automática, e latência abaixo de 1ms para enfileirar e desenfileirar.

<?php
declare(strict_types=1);

namespace App\Queue;

// ── Contrato: qualquer job deve implementar esta interface ─────────────
interface JobInterface
{
    public function processar(): void;

    // Número máximo de tentativas antes de ir para a dead-letter queue
    public function tentativasMaximas(): int;

    // Delay em segundos antes da próxima tentativa (backoff exponencial)
    public function backoffEmSegundos(int $tentativa): int;
}

// ── Envelope: metadados ao redor do job ───────────────────────────────
final class JobEnvelope
{
    public function __construct(
        public readonly string $id,
        public readonly string $classeJob,
        public readonly array  $payload,
        public readonly int    $tentativa       = 1,
        public readonly int    $criadoEm        = 0,
        public readonly ?string $erro           = null,
    ) {}

    public static function criar(string $classeJob, array $payload): self
    {
        return new self(
            id:        bin2hex(random_bytes(16)),
            classeJob: $classeJob,
            payload:   $payload,
            tentativa: 1,
            criadoEm:  time(),
        );
    }

    public function comFalha(string $erro): self
    {
        return new self(
            id:        $this->id,
            classeJob: $this->classeJob,
            payload:   $this->payload,
            tentativa: $this->tentativa + 1,
            criadoEm:  $this->criadoEm,
            erro:      $erro,
        );
    }

    public function serializar(): string
    {
        return json_encode([
            'id'        => $this->id,
            'classe'    => $this->classeJob,
            'payload'   => $this->payload,
            'tentativa' => $this->tentativa,
            'criado_em' => $this->criadoEm,
            'erro'      => $this->erro,
        ], JSON_UNESCAPED_UNICODE);
    }

    public static function desserializar(string $json): self
    {
        $dados = json_decode($json, associative: true);
        return new self(
            id:        $dados['id'],
            classeJob: $dados['classe'],
            payload:   $dados['payload'],
            tentativa: $dados['tentativa'],
            criadoEm:  $dados['criado_em'],
            erro:      $dados['erro'] ?? null,
        );
    }
}

Dispatcher — enfileira jobs

<?php
declare(strict_types=1);

namespace App\Queue;

class QueueDispatcher
{
    public function __construct(
        private readonly \Redis $redis,
        private readonly string $filaDefault = 'jobs:default',
    ) {}

    // Enfileira e retorna o ID do job — rápido, não bloqueia
    public function despachar(string $classeJob, array $payload, string $fila = ''): string
    {
        $envelope = JobEnvelope::criar($classeJob, $payload);
        $nomeFila = $fila ?: $this->filaDefault;

        // LPUSH: insere no início da lista (BRPOP consome do fim = FIFO)
        $this->redis->lPush($nomeFila, $envelope->serializar());

        return $envelope->id;
    }

    // Agenda job para execução futura (usando sorted set por timestamp)
    public function agendarPara(\DateTimeImmutable $quando, string $classeJob, array $payload): string
    {
        $envelope = JobEnvelope::criar($classeJob, $payload);
        $score     = $quando->getTimestamp();

        // ZADD: sorted set com timestamp como score
        $this->redis->zAdd('jobs:scheduled', $score, $envelope->serializar());

        return $envelope->id;
    }

    // Move jobs agendados cujo tempo chegou para a fila principal
    public function promoverJobsAgendados(): int
    {
        $agora    = time();
        $prontos  = $this->redis->zRangeByScore('jobs:scheduled', 0, $agora);

        if (empty($prontos)) return 0;

        // Transação atômica: remove do scheduled e insere na fila
        $this->redis->multi();
        foreach ($prontos as $item) {
            $this->redis->zRem('jobs:scheduled', $item);
            $this->redis->lPush($this->filaDefault, $item);
        }
        $this->redis->exec();

        return count($prontos);
    }

    public function tamanho(string $fila = ''): int
    {
        return (int) $this->redis->lLen($fila ?: $this->filaDefault);
    }
}

Jobs concretos

<?php
declare(strict_types=1);

namespace App\Jobs;

use App\Queue\JobInterface;

// ── Job de email ──────────────────────────────────────────────────────
class EnviarEmailJob implements JobInterface
{
    public function __construct(
        private readonly string $destinatario,
        private readonly string $assunto,
        private readonly string $corpo,
        private readonly string $template = 'padrao',
    ) {}

    public static function criar(string $destinatario, string $assunto, string $corpo): array
    {
        // Retorna o payload para o Dispatcher
        return [
            'destinatario' => $destinatario,
            'assunto'      => $assunto,
            'corpo'        => $corpo,
        ];
    }

    public function processar(): void
    {
        // Em produção: PHPMailer, Symfony Mailer, AWS SES, etc.
        $enviado = mail(
            $this->destinatario,
            $this->assunto,
            $this->corpo,
            ['Content-Type' => 'text/html; charset=UTF-8'],
        );

        if (!$enviado) {
            throw new \RuntimeException("Falha ao enviar email para {$this->destinatario}");
        }

        error_log("[EmailJob] Enviado para {$this->destinatario}: {$this->assunto}");
    }

    public function tentativasMaximas(): int { return 3; }

    public function backoffEmSegundos(int $tentativa): int
    {
        // Backoff exponencial: 60s, 300s, 900s
        return 60 * (3 ** ($tentativa - 1));
    }
}

// ── Job de geração de relatório ───────────────────────────────────────
class GerarRelatorioJob implements JobInterface
{
    public function __construct(
        private readonly int    $pedidoId,
        private readonly string $formato,   // 'pdf' | 'csv'
        private readonly string $emailDestino,
    ) {}

    public function processar(): void
    {
        error_log("[RelatorioJob] Gerando {$this->formato} para pedido #{$this->pedidoId}");

        // Simula geração pesada
        $conteudo = $this->gerarConteudo();
        $caminho  = $this->salvarArquivo($conteudo);

        // Enfileira job de email com o relatório — composição de jobs
        // $dispatcher->despachar(EnviarEmailJob::class, [...]);

        error_log("[RelatorioJob] Relatório salvo em {$caminho}");
    }

    private function gerarConteudo(): string
    {
        // Lógica de geração real aqui
        return "Relatório do pedido #{$this->pedidoId} em {$this->formato}";
    }

    private function salvarArquivo(string $conteudo): string
    {
        $caminho = sys_get_temp_dir() . "/relatorio-{$this->pedidoId}.{$this->formato}";
        file_put_contents($caminho, $conteudo);
        return $caminho;
    }

    public function tentativasMaximas(): int  { return 1; } // Sem retry — idempotência complexa
    public function backoffEmSegundos(int $n): int { return 0; }
}

Worker — consome a fila

<?php
declare(strict_types=1);

namespace App\Queue;

// ── Worker robusto com tratamento de sinais ───────────────────────────
class Worker
{
    private bool  $rodando       = true;
    private int   $processados   = 0;
    private int   $falhas        = 0;
    private float $iniciadoEm;

    public function __construct(
        private readonly \Redis $redis,
        private readonly string $fila        = 'jobs:default',
        private readonly string $filaDLQ     = 'jobs:dead',
        private readonly int    $timeout     = 5,   // segundos para BRPOP esperar
        private readonly int    $limiteMemoria = 128, // MB
    ) {
        $this->iniciadoEm = microtime(true);
    }

    public function iniciar(): void
    {
        $this->registrarSinais();
        error_log("[Worker] Iniciado. PID: " . getmypid() . ", Fila: {$this->fila}");

        while ($this->rodando) {
            $this->verificarMemoria();

            // BRPOP bloqueia até que um item chegue ou o timeout expire
            $item = $this->redis->brPop([$this->fila], $this->timeout);

            if ($item === null || $item === false) {
                continue; // timeout — verifica sinais e continua
            }

            [, $json] = $item; // [nomeDaFila, valor]
            $this->processarEnvelope($json);
        }

        error_log("[Worker] Encerrado. Processados: {$this->processados}, Falhas: {$this->falhas}");
    }

    private function processarEnvelope(string $json): void
    {
        $envelope = JobEnvelope::desserializar($json);

        try {
            $job = $this->instanciarJob($envelope);
            $job->processar();
            $this->processados++;

            error_log("[Worker] ✓ Job {$envelope->id} ({$envelope->classeJob}) processado.");

        } catch (\Throwable $e) {
            $this->falhas++;
            $envelopeComFalha = $envelope->comFalha($e->getMessage());

            error_log("[Worker] ✗ Job {$envelope->id} falhou (tentativa {$envelope->tentativa}): {$e->getMessage()}");

            $job = $this->instanciarJob($envelope);

            if ($envelope->tentativa < $job->tentativasMaximas()) {
                // Recoloca na fila com delay (backoff)
                $delay = $job->backoffEmSegundos($envelope->tentativa);
                $this->recolocarComDelay($envelopeComFalha, $delay);
            } else {
                // Esgotou as tentativas — vai para a Dead Letter Queue
                $this->moverParaDLQ($envelopeComFalha);
            }
        }
    }

    private function instanciarJob(JobEnvelope $envelope): JobInterface
    {
        $classe = $envelope->classeJob;

        if (!class_exists($classe)) {
            throw new \RuntimeException("Classe de job não encontrada: {$classe}");
        }

        // Instancia o job a partir do payload usando named arguments
        return new $classe(...$envelope->payload);
    }

    private function recolocarComDelay(JobEnvelope $envelope, int $delaySegundos): void
    {
        if ($delaySegundos <= 0) {
            $this->redis->lPush($this->fila, $envelope->serializar());
            return;
        }

        $quando = new \DateTimeImmutable("+{$delaySegundos} seconds");
        $this->redis->zAdd('jobs:scheduled', $quando->getTimestamp(), $envelope->serializar());

        error_log("[Worker] Job {$envelope->id} reagendado para {$quando->format('H:i:s')} (+{$delaySegundos}s)");
    }

    private function moverParaDLQ(JobEnvelope $envelope): void
    {
        $this->redis->lPush($this->filaDLQ, $envelope->serializar());
        error_log("[Worker] Job {$envelope->id} movido para DLQ após {$envelope->tentativa} tentativas.");
    }

    private function verificarMemoria(): void
    {
        $usoMB = memory_get_usage(true) / 1024 / 1024;
        if ($usoMB > $this->limiteMemoria) {
            error_log("[Worker] Memória excedida ({$usoMB}MB). Encerrando graciosamente.");
            $this->rodando = false;
        }
    }

    private function registrarSinais(): void
    {
        if (!extension_loaded('pcntl')) return;

        // SIGTERM: encerramento gracioso (Kubernetes, systemd)
        pcntl_signal(SIGTERM, function (): void {
            error_log("[Worker] SIGTERM recebido — encerrando após job atual.");
            $this->rodando = false;
        });

        // SIGINT: Ctrl+C durante desenvolvimento
        pcntl_signal(SIGINT, function (): void {
            error_log("[Worker] SIGINT recebido — encerrando.");
            $this->rodando = false;
        });

        pcntl_async_signals(true); // processa sinais sem ticks
    }

    public function estatisticas(): array
    {
        return [
            'processados'    => $this->processados,
            'falhas'         => $this->falhas,
            'uptime_segundos'=> (int) (microtime(true) - $this->iniciadoEm),
            'memoria_mb'     => round(memory_get_usage(true) / 1024 / 1024, 2),
            'fila_tamanho'   => (int) $this->redis->lLen($this->fila),
            'dlq_tamanho'    => (int) $this->redis->lLen($this->filaDLQ),
        ];
    }
}

Ponto de entrada do worker

<?php
// bin/worker.php — executado como processo separado

declare(strict_types=1);

require_once __DIR__ . '/../vendor/autoload.php';

$redis = new \Redis();
$redis->connect(
    host:    $_ENV['REDIS_HOST'] ?? '127.0.0.1',
    port:    (int) ($_ENV['REDIS_PORT'] ?? 6379),
);

$fila = $argv[1] ?? 'jobs:default'; // php bin/worker.php jobs:emails

$worker = new \App\Queue\Worker(
    redis:          $redis,
    fila:           $fila,
    filaDLQ:        'jobs:dead',
    timeout:        5,
    limiteMemoria:  128,
);

$worker->iniciar();

# supervisord.conf — mantém o worker rodando em produção
[program:worker-default]
command=php /var/www/bin/worker.php jobs:default
numprocs=4
autostart=true
autorestart=true
stdout_logfile=/var/log/worker-default.log
stderr_logfile=/var/log/worker-default-error.log

Resumo do artigo

Exercício da semana

1. Implemente ProcessarPagamentoJob que chama uma API de pagamento (mock). Se a API retornar 429 Too Many Requests, relance RateLimitException — o worker deve fazer backoff mais agressivo (5min, 15min, 1h). Se retornar 402, lance PagamentoRecusadoException — sem retry, vai direto para DLQ.

2. Crie um painel de monitoramento simples em PHP puro: endpoint /admin/queues que exibe via JSON o tamanho de todas as filas, jobs na DLQ (últimos 10), e estatísticas de processamento salvas no Redis com INCR.

3. Implemente QueueDispatcher::despacharLote(array $jobs): array que enfileira múltiplos jobs numa única transação Redis (MULTI/EXEC). Escreva testes que verificam que ou todos são enfileirados ou nenhum (atomicidade).

4. Adicione prioridades à fila: despachar(..., prioridade: 'high') deve ir para jobs:high e despachar(..., prioridade: 'low') para jobs:low. O worker deve consumir primeiro a fila alta — use BRPOP ['jobs:high', 'jobs:low'] (Redis processa na ordem).

5. Desafio: implemente um worker com múltiplos processos filhos usando pcntl_fork(). O processo pai mantém N filhos rodando, relança filhos que morrem com código de saída diferente de 0, e encerra todos graciosamente ao receber SIGTERM.

Referências

• Redis — Lists (BRPOP, LPUSH): https://redis.io/docs/data-types/lists/
• Redis — Sorted Sets (ZADD, ZRANGEBYSCORE): https://redis.io/docs/data-types/sorted-sets/
• PHP Manual — pcntl_signal: https://www.php.net/manual/pt_BR/function.pcntl-signal.php
• Supervisor — Process Control: http://supervisord.org/configuration.html