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;
// O MULTI aqui não daria a garantia que parece dar: a leitura acima
// aconteceu fora dele. Com quatro workers lendo o mesmo sorted set,
// todos enxergam os mesmos jobs e todos os enfileirariam.
// Quem desempata é o zRem, que devolve quantos elementos removeu de
// fato: só o processo que conseguir remover o item o enfileira.
$movidos = 0;
foreach ($prontos as $item) {
if ($this->redis->zRem('jobs:scheduled', $item) === 1) {
$this->redis->lPush($this->filaDefault, $item);
$movidos++;
}
}
return $movidos;
}
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 de razão 3: 60s, 180s, 540s.
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();
// Sem esta chamada o retry não acontece. O recolocarComDelay() manda
// o job para o sorted set 'jobs:scheduled', e o BRPOP abaixo só olha
// a fila principal: sem alguém movendo um para o outro, todo job
// reagendado fica parado para sempre — sem sumir e sem falhar,
// que é a forma mais difícil de perceber.
$this->promoverAgendados();
// 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()}");
// Instanciar de novo dentro do catch pede proteção: se a falha
// original foi exatamente essa — classe de job que sumiu num deploy
// — a exceção escaparia daqui e derrubaria o worker a cada
// mensagem malformada que passasse pela fila.
try {
$job = $this->instanciarJob($envelope);
} catch (\Throwable) {
$this->moverParaDLQ($envelopeComFalha);
return;
}
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);
}
/**
* Move para a fila principal os jobs agendados cujo horário já chegou.
* O limite por rodada existe para que o worker não gaste um ciclo inteiro
* promovendo quando um pico de reagendamentos vence de uma vez só.
*/
private function promoverAgendados(int $limite = 50): int
{
$prontos = $this->redis->zRangeByScore(
'jobs:scheduled', 0, time(), ['limit' => [0, $limite]]
);
$movidos = 0;
foreach ($prontos as $item) {
// Quem consegue remover é quem enfileira: com vários workers, o
// zRem é o que impede o mesmo job de entrar na fila duas vezes.
if ($this->redis->zRem('jobs:scheduled', $item) === 1) {
$this->redis->lPush($this->fila, $item);
$movidos++;
}
}
return $movidos;
}
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
Uma fila troca latência por complexidade, e a conta só fecha se a segunda parte for levada a sério. A resposta HTTP cai para poucos milissegundos, mas em troca aparecem perguntas que a requisição síncrona nunca fez: o que acontece se o worker morrer no meio, quantas vezes vale a pena tentar de novo, e o que fazer com o job que falhou todas. A dead-letter queue não é refinamento — é o que separa "falhou e ficou registrado" de "sumiu".
Repare que jobs precisam ser idempotentes, e essa exigência não vem da fila: vem do retry. Um job de e-mail que roda duas vezes manda duas mensagens; um que debita saldo duas vezes cobra em dobro. Por isso o job de relatório deste artigo declara uma única tentativa — assumir que a operação não pode ser repetida é uma resposta legítima, e mais honesta que fingir que pode.
Fontes e leituras recomendadas
- 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
- Symfony Messenger — barramento de mensagens com transports, retry e failure queue: https://symfony.com/doc/current/messenger.html
- RabbitMQ — tutorial de work queues em PHP, com ack e distribuicao entre consumidores: https://www.rabbitmq.com/tutorials/tutorial-two-php
Exercícios
Exercício 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.
Ver resposta
✓ Resposta: A distinção entre erro transitório e definitivo é o que separa uma fila saudável de uma que se autodestrói: retentar um cartão recusado cinco vezes não o aprova, gasta cota do gateway e pode duplicar cobrança. E respeitar o Retry-After do 429 é o mínimo de educação com o serviço do outro lado.
<?php
declare(strict_types=1);
final class RateLimitException extends \RuntimeException
{
public function __construct(public readonly int $retryAfter = 300)
{
parent::__construct("Gateway limitou a taxa; tentar em {$retryAfter}s.");
}
}
/** Erro definitivo: retentar não muda o resultado. */
final class PagamentoRecusadoException extends \RuntimeException
{
public function __construct(public readonly string $codigo, public readonly int $pedidoId)
{
parent::__construct("Pagamento do pedido {$pedidoId} recusado (código {$codigo}).");
}
}
interface JobInterface
{
public function handle(): void;
public function tentativasMaximas(): int;
/** Segundos de espera antes da tentativa $tentativa (1-based). */
public function backoff(int $tentativa): int;
/** Erros que NÃO devem ser retentados — vão direto para a DLQ. */
public function naoRetentar(): array;
}
final class ProcessarPagamentoJob implements JobInterface
{
public function __construct(
public readonly int $pedidoId,
public readonly int $valorCentavos,
public readonly string $cartaoToken,
) {}
public function handle(): void
{
$resposta = $this->chamarGateway();
// 429: o gateway está sobrecarregado. Não é erro nosso — vale
// esperar bastante e tentar de novo.
if ($resposta['status'] === 429) {
throw new RateLimitException(
retryAfter: (int) ($resposta['headers']['Retry-After'] ?? 300)
);
}
// 402: o banco recusou. Retentar 5 vezes só irrita o cliente e
// pode gerar cobrança duplicada.
if ($resposta['status'] === 402) {
throw new PagamentoRecusadoException(
codigo: (string) ($resposta['corpo']['codigo'] ?? 'desconhecido'),
pedidoId: $this->pedidoId,
);
}
if ($resposta['status'] >= 500) {
// Erro transitório do servidor: backoff padrão.
throw new \RuntimeException("Gateway indisponível ({$resposta['status']}).");
}
if ($resposta['status'] !== 200) {
throw new \RuntimeException("Resposta inesperada: {$resposta['status']}.");
}
error_log("Pedido {$this->pedidoId} pago — {$resposta['corpo']['autorizacao']}");
}
public function tentativasMaximas(): int { return 5; }
/**
* Backoff exponencial para erro comum; escalada agressiva quando o
* gateway limitou a taxa — martelar um gateway limitado só piora.
*/
public function backoff(int $tentativa): int
{
return match ($tentativa) {
1 => 60, // 1 min
2 => 300, // 5 min
3 => 900, // 15 min
default => 3600, // 1 h
};
}
public function naoRetentar(): array
{
return [PagamentoRecusadoException::class];
}
/** Mock do gateway. */
private function chamarGateway(): array
{
return match (true) {
$this->valorCentavos > 1_000_000 => ['status' => 402,
'corpo' => ['codigo' => '51'], 'headers' => []],
$this->cartaoToken === 'tok_limitado' => ['status' => 429,
'corpo' => [], 'headers' => ['Retry-After' => '600']],
default => ['status' => 200,
'corpo' => ['autorizacao' => 'AUT-' . bin2hex(random_bytes(3))], 'headers' => []],
};
}
}
// ---------- O worker que aplica as políticas -----------------------------
final class Worker
{
public function __construct(
private readonly \Redis $redis,
private readonly string $fila = 'jobs:default',
) {}
public function processar(JobInterface $job, array $envelope): void
{
$tentativa = ($envelope['tentativas'] ?? 0) + 1;
try {
$job->handle();
$this->redis->incr('stats:jobs:sucesso');
} catch (\Throwable $e) {
// Erro definitivo: nem tenta de novo.
foreach ($job->naoRetentar() as $classe) {
if ($e instanceof $classe) {
$this->paraDlq($envelope, $e, 'erro_definitivo');
return;
}
}
if ($tentativa >= $job->tentativasMaximas()) {
$this->paraDlq($envelope, $e, 'tentativas_esgotadas');
return;
}
// O próprio erro pode ditar a espera (Retry-After do 429).
$espera = $e instanceof RateLimitException
? max($e->retryAfter, $job->backoff($tentativa))
: $job->backoff($tentativa);
$envelope['tentativas'] = $tentativa;
$envelope['ultimo_erro'] = $e->getMessage();
// ZADD com score = quando pode rodar: a fila de atrasados é um
// sorted set, e o worker só puxa o que já venceu.
$this->redis->zAdd('jobs:delayed', time() + $espera, json_encode($envelope));
$this->redis->incr('stats:jobs:retentados');
}
}
private function paraDlq(array $envelope, \Throwable $e, string $motivo): void
{
$envelope['falhou_em'] = date('c');
$envelope['motivo'] = $motivo;
$envelope['erro'] = $e->getMessage();
$envelope['excecao'] = $e::class;
$this->redis->lPush('jobs:dlq', json_encode($envelope));
$this->redis->incr('stats:jobs:dlq');
error_log("DLQ ({$motivo}): {$e->getMessage()}");
}
}Exercício 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.
Ver resposta
✓ Resposta: O número mais útil do painel é atrasados.vencidos: job que já passou da hora e continua parado significa worker morto. Fila crescendo você percebe; worker parado com fila pequena passa despercebido por horas.
<?php
// public/admin/queues.php — GET /admin/queues
declare(strict_types=1);
require __DIR__ . '/../../vendor/autoload.php';
final class PainelDeFilas
{
private const FILAS = ['jobs:high', 'jobs:default', 'jobs:low'];
public function __construct(private readonly \Redis $redis) {}
public function relatorio(): array
{
return [
'gerado_em' => date('c'),
'filas' => $this->tamanhos(),
'atrasados' => $this->atrasados(),
'dlq' => $this->dlq(10),
'estatisticas'=> $this->estatisticas(),
'redis' => $this->infoRedis(),
];
}
private function tamanhos(): array
{
// pipeline: uma ida ao Redis em vez de N. Com 3 filas não muda nada,
// com 30 muda muito.
$pipe = $this->redis->multi(\Redis::PIPELINE);
foreach (self::FILAS as $fila) {
$pipe->lLen($fila);
}
$resultados = $pipe->exec();
$saida = [];
foreach (self::FILAS as $i => $fila) {
$saida[$fila] = (int) $resultados[$i];
}
$saida['total'] = array_sum($saida);
return $saida;
}
private function atrasados(): array
{
$agora = time();
return [
'total' => (int) $this->redis->zCard('jobs:delayed'),
// Já venceram mas ninguém puxou: sinal de worker parado.
'vencidos' => (int) $this->redis->zCount('jobs:delayed', '-inf', $agora),
'proximo_em' => (function () use ($agora): ?int {
$proximos = $this->redis->zRange('jobs:delayed', 0, 0, true);
return $proximos ? max(0, (int) reset($proximos) - $agora) : null;
})(),
];
}
private function dlq(int $quantos): array
{
$itens = $this->redis->lRange('jobs:dlq', 0, $quantos - 1) ?: [];
return [
'total' => (int) $this->redis->lLen('jobs:dlq'),
'ultimos' => array_map(static function (string $json): array {
$envelope = json_decode($json, true) ?: [];
return [
'job' => $envelope['job'] ?? '?',
'motivo' => $envelope['motivo'] ?? '?',
'erro' => mb_substr((string) ($envelope['erro'] ?? ''), 0, 160),
'tentativas' => $envelope['tentativas'] ?? 0,
'falhou_em' => $envelope['falhou_em'] ?? null,
];
}, $itens),
];
}
private function estatisticas(): array
{
$chaves = ['stats:jobs:sucesso', 'stats:jobs:retentados', 'stats:jobs:dlq'];
// mGet: um round-trip para todos os contadores.
$valores = $this->redis->mGet($chaves);
$stats = [];
foreach ($chaves as $i => $chave) {
$stats[str_replace('stats:jobs:', '', $chave)] = (int) ($valores[$i] ?: 0);
}
$total = $stats['sucesso'] + $stats['dlq'];
$stats['taxa_sucesso'] = $total > 0
? round($stats['sucesso'] / $total * 100, 2)
: 100.0;
// Contadores por hora, para desenhar o gráfico das últimas 24h.
$porHora = [];
for ($h = 23; $h >= 0; $h--) {
$chave = 'stats:jobs:hora:' . date('YmdH', strtotime("-{$h} hours"));
$porHora[date('H:00', strtotime("-{$h} hours"))] = (int) ($this->redis->get($chave) ?: 0);
}
$stats['ultimas_24h'] = $porHora;
return $stats;
}
private function infoRedis(): array
{
$info = $this->redis->info('memory');
return [
'memoria_usada' => $info['used_memory_human'] ?? '?',
'memoria_pico' => $info['used_memory_peak_human'] ?? '?',
];
}
}
// ---------- Endpoint ------------------------------------------------------
$redis = new Redis();
$redis->connect($_ENV['REDIS_HOST'] ?? '127.0.0.1', 6379);
// Painel operacional é dado sensível: nunca exponha sem autenticação.
if (($_SERVER['HTTP_X_ADMIN_TOKEN'] ?? '') !== ($_ENV['ADMIN_TOKEN'] ?? null)
|| ($_ENV['ADMIN_TOKEN'] ?? '') === '') {
http_response_code(401);
echo json_encode(['erro' => 'nao_autorizado']);
exit;
}
header('Content-Type: application/json; charset=utf-8');
header('Cache-Control: no-store');
echo json_encode(
(new PainelDeFilas($redis))->relatorio(),
JSON_PRETTY_PRINT | JSON_UNESCAPED_UNICODE | JSON_UNESCAPED_SLASHES,
);
// {
// "filas": { "jobs:high": 0, "jobs:default": 42, "jobs:low": 310, "total": 352 },
// "atrasados": { "total": 18, "vencidos": 0, "proximo_em": 240 },
// "dlq": { "total": 3, "ultimos": [ ... ] },
// "estatisticas": { "sucesso": 18422, "dlq": 3, "taxa_sucesso": 99.98 }
// }Exercício 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).
Ver resposta
✓ Resposta: Vale entender o limite do MULTI/EXEC: ele garante isolamento (nenhum outro cliente se intromete) e "tudo ou nada" quanto à execução, mas não tem rollback — se um comando falhar em runtime, os anteriores já valeram. Não é transação de banco relacional.
<?php
declare(strict_types=1);
final class QueueDispatcher
{
public function __construct(
private readonly \Redis $redis,
private readonly string $filaPadrao = 'jobs:default',
) {}
public function despachar(JobInterface $job, string $prioridade = 'default'): string
{
$envelope = $this->envelopar($job);
$this->redis->lPush($this->filaDe($prioridade), json_encode($envelope));
return $envelope['id'];
}
/**
* Ou TODOS entram, ou nenhum entra.
*
* MULTI/EXEC no Redis enfileira os comandos e executa todos de uma vez,
* sem que outro cliente se intrometa no meio. Não é transação de banco:
* não há rollback se um comando falhar por erro de runtime — mas nenhum
* comando é executado se o EXEC não acontecer, que é a garantia que o
* exercício pede.
*
* @param JobInterface[] $jobs
* @return string[] ids gerados
*/
public function despacharLote(array $jobs, string $prioridade = 'default'): array
{
if ($jobs === []) {
return [];
}
$envelopes = array_map($this->envelopar(...), $jobs);
$fila = $this->filaDe($prioridade);
$pipe = $this->redis->multi(); // MULTI
foreach ($envelopes as $envelope) {
$pipe->lPush($fila, json_encode($envelope, JSON_THROW_ON_ERROR));
}
$pipe->incrBy('stats:jobs:enfileirados', count($envelopes));
$resultados = $pipe->exec(); // EXEC
// exec() devolve false se a transação foi abortada (ex.: WATCH
// detectou alteração). Nesse caso NADA foi enfileirado.
if ($resultados === false) {
throw new \RuntimeException('Transação Redis abortada; nenhum job foi enfileirado.');
}
return array_column($envelopes, 'id');
}
/** Versão com WATCH: aborta se a fila mudar durante a montagem. */
public function despacharLoteSeFilaEstavel(array $jobs, string $prioridade = 'default'): array
{
$fila = $this->filaDe($prioridade);
$this->redis->watch($fila);
$tamanhoAntes = $this->redis->lLen($fila);
if ($tamanhoAntes > 10_000) {
$this->redis->unwatch();
throw new \RuntimeException('Fila congestionada; lote recusado.');
}
return $this->despacharLote($jobs, $prioridade);
}
private function envelopar(JobInterface $job): array
{
return [
'id' => bin2hex(random_bytes(12)),
'job' => $job::class,
'payload' => get_object_vars($job),
'tentativas' => 0,
'criado_em' => date('c'),
];
}
private function filaDe(string $prioridade): string
{
return match ($prioridade) {
'high' => 'jobs:high',
'low' => 'jobs:low',
default => $this->filaPadrao,
};
}
}
// ---------------------------------------- Tests/Integration/QueueDispatcherTest.php
declare(strict_types=1);
namespace Tests\Integration;
use PHPUnit\Framework\Attributes\Test;
use PHPUnit\Framework\TestCase;
final class QueueDispatcherTest extends TestCase
{
private \Redis $redis;
private QueueDispatcher $dispatcher;
protected function setUp(): void
{
$this->redis = new \Redis();
$this->redis->connect('127.0.0.1', 6379);
$this->redis->select(15); // banco 15 só para teste
$this->redis->flushDB();
$this->dispatcher = new QueueDispatcher($this->redis);
}
protected function tearDown(): void
{
$this->redis->flushDB();
}
#[Test]
public function loteEnfileiraTodosOsJobs(): void
{
$jobs = [
new ProcessarPagamentoJob(1, 10000, 'tok_a'),
new ProcessarPagamentoJob(2, 20000, 'tok_b'),
new ProcessarPagamentoJob(3, 30000, 'tok_c'),
];
$ids = $this->dispatcher->despacharLote($jobs);
self::assertCount(3, $ids);
self::assertSame(3, $this->redis->lLen('jobs:default'));
self::assertSame(3, (int) $this->redis->get('stats:jobs:enfileirados'));
// Ids únicos: dois jobs nunca compartilham identificador.
self::assertCount(3, array_unique($ids));
}
#[Test]
public function loteVazioNaoTocaNaFila(): void
{
self::assertSame([], $this->dispatcher->despacharLote([]));
self::assertSame(0, $this->redis->lLen('jobs:default'));
self::assertFalse($this->redis->exists('stats:jobs:enfileirados') > 0);
}
#[Test]
public function atomicidadeNenhumJobEntraSeATransacaoAborta(): void
{
$this->redis->set('jobs:default', 'valor-que-nao-eh-lista');
try {
$this->dispatcher->despacharLote([new ProcessarPagamentoJob(1, 100, 'tok')]);
} catch (\Throwable) {
// esperado
}
// A chave continua sendo a string: nenhum lPush teve efeito.
self::assertSame('string', $this->redis->type('jobs:default') === \Redis::REDIS_STRING
? 'string' : 'outro');
}
#[Test]
public function prioridadeDirecionaParaAFilaCerta(): void
{
$this->dispatcher->despacharLote([new ProcessarPagamentoJob(1, 100, 'a')], 'high');
$this->dispatcher->despacharLote([new ProcessarPagamentoJob(2, 200, 'b')], 'low');
self::assertSame(1, $this->redis->lLen('jobs:high'));
self::assertSame(1, $this->redis->lLen('jobs:low'));
self::assertSame(0, $this->redis->lLen('jobs:default'));
}
}Exercício 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).
Ver resposta
✓ Resposta: O ponto que o exercício ensina: a prioridade não é um campo do job — é a ordem das chaves no BRPOP. E cuidado com o outro lado: se jobs:high nunca esvaziar, jobs:low jamais é atendida. Fila de baixa prioridade sob pressão constante precisa de worker dedicado.
<?php
declare(strict_types=1);
final class WorkerPrioritario
{
/** A ORDEM DO ARRAY é a prioridade: BRPOP percorre as chaves na ordem
* dada e devolve o primeiro elemento que encontrar. */
private const FILAS = ['jobs:high', 'jobs:default', 'jobs:low'];
private bool $rodando = true;
public function __construct(
private readonly \Redis $redis,
private readonly int $timeout = 5,
) {}
public function rodar(): void
{
// Sinais precisam ser despachados: sem isto, o handler nunca roda.
pcntl_async_signals(true);
pcntl_signal(SIGTERM, fn () => $this->rodando = false);
pcntl_signal(SIGINT, fn () => $this->rodando = false);
echo "Worker iniciado (PID " . getmypid() . ")", PHP_EOL;
while ($this->rodando) {
$this->promoverAtrasados();
// BRPOP bloqueia até chegar job ou estourar o timeout — sem
// polling, sem queimar CPU. O timeout existe para que o laço
// volte de tempos em tempos e possa checar $this->rodando.
$resultado = $this->redis->brPop(self::FILAS, $this->timeout);
if (empty($resultado)) {
continue; // timeout: nenhuma fila tinha job
}
[$fila, $payload] = $resultado;
$this->executar($fila, $payload);
}
echo 'Worker encerrado com elegância.', PHP_EOL;
}
/** Move para a fila normal os jobs atrasados cujo horário já chegou. */
private function promoverAtrasados(): void
{
$vencidos = $this->redis->zRangeByScore('jobs:delayed', '-inf', (string) time(),
['limit' => [0, 50]]);
foreach ($vencidos as $payload) {
// zRem antes do lPush: se outro worker já removeu, não duplica.
if ($this->redis->zRem('jobs:delayed', $payload) > 0) {
$envelope = json_decode($payload, true) ?: [];
$this->redis->lPush($envelope['fila'] ?? 'jobs:default', $payload);
}
}
}
private function executar(string $fila, string $payload): void
{
$envelope = json_decode($payload, true);
if (!is_array($envelope) || !isset($envelope['job'])) {
$this->redis->lPush('jobs:dlq', $payload);
return;
}
$inicio = hrtime(true);
try {
$classe = $envelope['job'];
if (!class_exists($classe) || !is_subclass_of($classe, JobInterface::class)) {
throw new \RuntimeException("Job desconhecido: {$classe}");
}
$job = new $classe(...array_values($envelope['payload'] ?? []));
$job->handle();
$this->redis->incr('stats:jobs:sucesso');
$this->redis->incr('stats:jobs:hora:' . date('YmdH'));
printf("[%s] %s ok (%.1f ms)%s", $fila, $classe,
(hrtime(true) - $inicio) / 1_000_000, PHP_EOL);
} catch (\Throwable $e) {
$envelope['fila'] = $fila;
(new Worker($this->redis, $fila))->processar($job ?? null, $envelope);
}
}
}
// ---------- Despachando com prioridade -----------------------------------
$dispatcher = new QueueDispatcher($redis);
// Pagamento não espera atrás de 300 e-mails de newsletter.
$dispatcher->despachar(new ProcessarPagamentoJob(42, 34990, 'tok_x'), prioridade: 'high');
$dispatcher->despachar(new EnviarNewsletterJob(assinanteId: 9001), prioridade: 'low');
(new WorkerPrioritario($redis))->rodar();
// [jobs:high] ProcessarPagamentoJob ok (182.4 ms)
// [jobs:low] EnviarNewsletterJob ok (44.1 ms)Exercício 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.
Ver resposta
✓ Resposta: A armadilha nº 1 do fork é herdar conexão: pai e filho no mesmo socket de Redis ou MySQL produzem respostas trocadas e erros impossíveis de reproduzir — sempre reconecte no filho. A nº 2 é relançar sem pausa: um filho que morre no boot vira milhares de forks por segundo.
<?php
// worker-pool.php — uso: php worker-pool.php [numero_de_filhos]
declare(strict_types=1);
require __DIR__ . '/vendor/autoload.php';
final class PoolDeWorkers
{
/** @var array<int, bool> pid => vivo */
private array $filhos = [];
private bool $encerrando = false;
public function __construct(
private readonly int $quantidade = 4,
private readonly string $script = 'worker.php',
) {}
public function iniciar(): void
{
if (!function_exists('pcntl_fork')) {
throw new \RuntimeException('pcntl não disponível (só CLI, só Unix).');
}
pcntl_async_signals(true);
// O pai não morre no SIGTERM: ele repassa aos filhos e espera.
pcntl_signal(SIGTERM, $this->encerrar(...));
pcntl_signal(SIGINT, $this->encerrar(...));
echo "Pai {$this->pid()} iniciando {$this->quantidade} filho(s)…", PHP_EOL;
for ($i = 0; $i < $this->quantidade; $i++) {
$this->criarFilho();
}
$this->supervisionar();
}
private function criarFilho(): void
{
$pid = pcntl_fork();
if ($pid === -1) {
throw new \RuntimeException('Falha no fork.');
}
if ($pid === 0) {
// ---- FILHO ----
// Tudo aqui é uma cópia da memória do pai. A conexão com Redis
// NÃO pode ser herdada: dois processos no mesmo socket embaralham
// as respostas. Reconecte sempre depois do fork.
$this->rodarComoFilho();
exit(0); // filho nunca volta ao laço do pai
}
// ---- PAI ----
$this->filhos[$pid] = true;
echo " filho {$pid} criado", PHP_EOL;
}
private function rodarComoFilho(): void
{
// O filho herda os handlers do pai: reinstale os dele.
pcntl_signal(SIGTERM, static function (): void {
echo ' filho ' . getmypid() . ' encerrando…', PHP_EOL;
exit(0);
});
$redis = new \Redis();
$redis->connect($_ENV['REDIS_HOST'] ?? '127.0.0.1', 6379);
(new WorkerPrioritario($redis))->rodar();
}
private function supervisionar(): void
{
while (!$this->encerrando || $this->filhos !== []) {
// WNOHANG: não bloqueia; o pai continua respondendo a sinais.
$pid = pcntl_waitpid(-1, $status, WNOHANG);
if ($pid > 0) {
unset($this->filhos[$pid]);
$codigo = pcntl_wifexited($status) ? pcntl_wexitstatus($status) : -1;
if ($this->encerrando) {
echo " filho {$pid} encerrou ({$codigo})", PHP_EOL;
} elseif ($codigo === 0) {
echo " filho {$pid} terminou normalmente; recriando", PHP_EOL;
$this->criarFilho();
} else {
// Saída != 0 = morreu com erro. Relançar é o trabalho do
// supervisor — mas com pausa, senão um bug vira loop de
// fork a milhares por segundo.
echo " ⚠️ filho {$pid} morreu (código {$codigo}); relançando em 1s", PHP_EOL;
sleep(1);
$this->criarFilho();
}
continue;
}
usleep(200_000); // 200 ms: o supervisor quase não consome CPU
}
echo "Pai {$this->pid()} encerrado.", PHP_EOL;
}
private function encerrar(): void
{
if ($this->encerrando) {
return;
}
$this->encerrando = true;
echo 'SIGTERM recebido: avisando ', count($this->filhos), ' filho(s)…', PHP_EOL;
foreach (array_keys($this->filhos) as $pid) {
posix_kill($pid, SIGTERM);
}
// Prazo para terminar o job em andamento; depois, SIGKILL.
$limite = time() + 30;
while ($this->filhos !== [] && time() < $limite) {
$pid = pcntl_waitpid(-1, $status, WNOHANG);
if ($pid > 0) {
unset($this->filhos[$pid]);
}
usleep(100_000);
}
foreach (array_keys($this->filhos) as $pid) {
echo " filho {$pid} não respondeu: SIGKILL", PHP_EOL;
posix_kill($pid, SIGKILL);
}
}
private function pid(): int { return getmypid(); }
}
$quantidade = (int) ($argv[1] ?? 4);
(new PoolDeWorkers(max(1, $quantidade)))->iniciar();
// $ php worker-pool.php 3
// Pai 8412 iniciando 3 filho(s)…
// filho 8413 criado
// filho 8414 criado
// filho 8415 criado
// ^C
// SIGTERM recebido: avisando 3 filho(s)…
// filho 8413 encerrando…