Symfony Messenger: Filas, Workers e Arquitetura Assíncrona para Entrevistas em 2026

O Symfony Messenger gerencia cargas de trabalho assíncronas em PHP por meio de um message bus estruturado, transports dedicados e workers supervisionados. Com o Symfony 7.3 e o 8.0 trazendo middleware de deduplicação, streaming AMQP e keepalive para Doctrine, o componente compete diretamente com sistemas de filas dedicados como Laravel Horizon ou Sidekiq em completude de funcionalidades.

Arquitetura do Symfony Messenger: Bus, Transport e Worker

O componente Messenger separa três responsabilidades: o despacho (bus), a entrega (transport) e o processamento (worker). Uma mensagem é um objeto PHP simples. Um handler é uma classe invocável. O bus conecta ambos, roteando por meio de middlewares e, opcionalmente, serializando a mensagem para um transport para processamento assíncrono.

namespace App\Message;

final readonly class InvoiceGenerated
{
    public function __construct(
        public int $orderId,
        public string $customerEmail,
    ) {}
}

Essa mensagem carrega apenas dados escalares, sem entidades Doctrine. Passar IDs em vez de objetos evita problemas de serialização e mantém as mensagens leves. O handler busca dados atualizados diretamente no banco de dados no momento do processamento.

namespace App\MessageHandler;

use App\Message\InvoiceGenerated;
use App\Service\InvoiceService;
use App\Service\MailerService;
use Symfony\Component\Messenger\Attribute\AsMessageHandler;

#[AsMessageHandler]
final readonly class InvoiceGeneratedHandler
{
    public function __construct(
        private InvoiceService $invoiceService,
        private MailerService $mailerService,
    ) {}

    public function __invoke(InvoiceGenerated $message): void
    {
        $pdf = $this->invoiceService->generatePdf($message->orderId);
        $this->mailerService->sendInvoice(
            $message->customerEmail,
            $pdf,
        );
    }
}

O despacho da mensagem a partir de um controller ou service resume-se a uma linha:

$this->bus->dispatch(new InvoiceGenerated(
    orderId: $order->getId(),
    customerEmail: $order->getCustomer()->getEmail(),
));

O bus decide entre execução síncrona ou assíncrona com base na configuração de roteamento dos transports.

Configuração de Transports e Backends de Fila

O Messenger suporta Doctrine DBAL, Redis, Amazon SQS, Beanstalkd, AMQP (RabbitMQ) e o novo transport streaming AMQP introduzido em 2025. Cada transport corresponde a uma string DSN.

# config/packages/messenger.yaml
framework:
    messenger:
        failure_transport: failed

        transports:
            async_priority_high:
                dsn: '%env(MESSENGER_TRANSPORT_DSN)%'
                options:
                    queue_name: high
                retry_strategy:
                    max_retries: 3
                    delay: 1000
                    multiplier: 3
                    max_delay: 60000

            async_priority_low:
                dsn: '%env(MESSENGER_TRANSPORT_DSN)%'
                options:
                    queue_name: low
                retry_strategy:
                    max_retries: 5
                    delay: 5000
                    multiplier: 2

            failed:
                dsn: 'doctrine://default?queue_name=failed'

        routing:
            'App\Message\InvoiceGenerated': async_priority_high
            'App\Message\CleanupTempFiles': async_priority_low

A separação de transports por prioridade garante que tarefas sensíveis ao tempo (geração de fatura) sejam processadas antes de tarefas de manutenção (limpeza de arquivos temporários). Cada transport recebe sua própria estratégia de retry, ajustada conforme a criticidade e a idempotência das mensagens.

Gerenciamento de Workers com Supervisor

Os workers consomem mensagens dos transports. Em produção, o comando messenger:consume executa sob um gerenciador de processos como Supervisor ou systemd.

# Consume high-priority messages first, then low-priority
php bin/console messenger:consume async_priority_high async_priority_low \
    --memory-limit=128M \
    --time-limit=3600 \
    --limit=500

As três flags de limite previnem vazamentos de memória e garantem que os workers reiniciem periodicamente. O Supervisor reinicia o processo após cada encerramento.

; /etc/supervisor/conf.d/messenger-worker.conf
[program:messenger-consume]
command=php /var/www/app/bin/console messenger:consume async_priority_high async_priority_low --memory-limit=128M --time-limit=3600
user=www-data
numprocs=2
process_name=%(program_name)s_%(process_num)02d
autostart=true
autorestart=true
startsecs=0
stopwaitsecs=30
stdout_logfile=/var/log/messenger-worker.log
stderr_logfile=/var/log/messenger-worker-error.log

A configuração numprocs=2 cria dois workers paralelos, dobrando a capacidade de processamento. O ajuste deve considerar o número de cores da CPU do servidor e o tempo médio de processamento de cada mensagem.

Pipeline de Middleware e CQRS com Múltiplos Buses

O middleware envolve cada despacho de mensagem, adicionando aspectos transversais. A pilha de middleware nativa trata validação, transações Doctrine e roteamento.

# config/packages/messenger.yaml
framework:
    messenger:
        default_bus: command.bus
        buses:
            command.bus:
                middleware:
                    - validation
                    - doctrine_transaction
            query.bus:
                middleware:
                    - validation
            event.bus:
                default_middleware:
                    allow_no_handlers: true
                middleware:
                    - validation

Essa configuração implementa CQRS (Command Query Responsibility Segregation). Os commands alteram estado dentro de uma transação Doctrine. As queries são somente leitura. Os events permitem zero handlers, habilitando um padrão pub/sub onde listeners podem ser adicionados de forma independente.

Middleware de Deduplicação no Symfony 7.3

Mensagens duplicadas desperdiçam recursos e podem causar efeitos colaterais como a cobrança dupla de um cliente. O Symfony 7.3 introduz o DeduplicateMiddleware para ignorar automaticamente mensagens idênticas já presentes na fila.

namespace App\Message;

use Symfony\Component\Messenger\Stamp\DeduplicateStamp;

final readonly class SendWelcomeEmail
{
    public function __construct(
        public int $userId,
    ) {}
}

// Dispatching with deduplication
use Symfony\Component\Messenger\Stamp\DeduplicateStamp;

$this->bus->dispatch(
    new SendWelcomeEmail(userId: 42),
    [new DeduplicateStamp(id: 'welcome-email-42')],
);

O DeduplicateStamp recebe um identificador de recurso de lock. Se uma mensagem com o mesmo ID já estiver pendente, o novo despacho é descartado silenciosamente. Essa funcionalidade depende do componente Lock com um store serializável (Redis, Memcached ou banco de dados).

Estratégias de Retry e Failure Transport

Quando um handler lança uma exceção, o Messenger tenta reprocessar a mensagem conforme a estratégia de retry do transport. Após esgotar as tentativas, a mensagem é movida para o failure transport.

namespace App\MessageHandler;

use App\Message\ProcessPayment;
use Symfony\Component\Messenger\Attribute\AsMessageHandler;
use Symfony\Component\Messenger\Exception\RecoverableMessageHandlingException;

#[AsMessageHandler]
final class PaymentHandler
{
    public function __invoke(ProcessPayment $message): void
    {
        try {
            $this->gateway->charge($message->amount, $message->token);
        } catch (GatewayTimeoutException $e) {
            // Recoverable: retry with backoff
            throw new RecoverableMessageHandlingException(
                'Payment gateway timeout, retrying',
                previous: $e,
            );
        } catch (InvalidCardException $e) {
            // Unrecoverable: send to failure transport immediately
            throw new UnrecoverableMessageHandlingException(
                'Invalid card, no retry',
                previous: $e,
            );
        }
    }
}

O RecoverableMessageHandlingException aciona a estratégia de retry. O UnrecoverableMessageHandlingException ignora as retentativas por completo e envia a mensagem diretamente para o failure transport. Essa distinção evita desperdício de tentativas em mensagens permanentemente inválidas.

# Inspect and manage failed messages
php bin/console messenger:failed:show
php bin/console messenger:failed:show 20 --transport=failed

# Retry specific messages
php bin/console messenger:failed:retry 20 30

# Filter and remove by class (Symfony 7.3+)
php bin/console messenger:failed:remove --class-filter="App\Message\CleanupTempFiles"

Doctrine Keepalive e Mensagens de Longa Duração

Handlers de mensagens de longa duração correm o risco de ter suas mensagens reentregues quando o timeout de visibilidade do transport expira. O Symfony 7.2 introduziu keepalive para Redis, SQS e Beanstalkd. O Symfony 7.3 estende esse suporte ao transport Doctrine.

# Enable keepalive to prevent redelivery during long processing
php bin/console messenger:consume async --keepalive

A flag --keepalive atualiza periodicamente o timestamp delivered_at na tabela do transport Doctrine, sinalizando que o worker ainda está processando ativamente a mensagem. Sem essa flag, uma mensagem em processamento por mais tempo que o timeout do transport (padrão de 5 minutos) seria capturada por outro worker, causando processamento duplicado.

Atributo #[AsMessage] para Roteamento Declarativo

O Symfony 7.2 introduziu o atributo #[AsMessage], transferindo o roteamento de transport do YAML para a própria classe da mensagem.

namespace App\Message;

use Symfony\Component\Messenger\Attribute\AsMessage;

#[AsMessage(transport: 'async_priority_low')]
final readonly class GenerateReport
{
    public function __construct(
        public int $reportId,
        public string $format = 'pdf',
    ) {}
}

Isso elimina a seção routing no messenger.yaml para essa mensagem. O transport é declarado diretamente na origem, tornando o código autodocumentado. Ambas as abordagens (roteamento YAML e atributo) coexistem, com o atributo tendo precedência.

Transport Streaming AMQP para Alta Vazão

O transport AMQP tradicional utiliza polling (get()) para buscar mensagens, gerando carga desnecessária no RabbitMQ. O transport streaming AMQP lançado em 2025 muda para um modelo push (consume()), reduzindo latência e consumo de recursos.

# config/packages/messenger.yaml
framework:
    messenger:
        transports:
            streaming:
                dsn: 'amqp-lib://guest:guest@localhost:5672/%2f/messages'
                options:
                    exchange:
                        name: app_events
                        type: topic
                    queues:
                        order_events:
                            binding_keys: ['order.*']

As diferenças-chave em relação ao transport AMQP padrão: nenhuma extensão C necessária (utiliza php-amqplib), entrega streaming por conexões TCP de longa duração e suporte nativo a topic exchanges com roteamento por binding keys. Esse transport processa milhares de mensagens por segundo com sobrecarga mínima de CPU.

Conclusão

• Passar IDs escalares nas mensagens, sem entidades Doctrine; buscar dados atualizados no handler para evitar problemas de serialização e estado obsoleto
• Separar transports por prioridade e criticidade; cada transport com sua própria estratégia de retry e pool de workers dedicado
• Usar RecoverableMessageHandlingException e UnrecoverableMessageHandlingException para controlar explicitamente o comportamento de retentativas
• Ativar --keepalive nos workers com transport Doctrine para prevenir reentrega de mensagens de longa duração
• Aplicar DeduplicateStamp em mensagens onde processamento duplicado causa efeitos colaterais (pagamentos, e-mails, notificações)
• Implementar CQRS com múltiplos buses: command bus para mutações com transações Doctrine, query bus para leituras, event bus para pub/sub
• Usar Supervisor ou systemd em produção com as flags --memory-limit, --time-limit e --limit para o gerenciamento do ciclo de vida dos workers
• Considerar o transport streaming AMQP para cenários de alta vazão que exigem latência inferior a um milissegundo