Symfony Messenger nel 2026: Code, Worker e Architettura Asincrona per Colloqui Tecnici

Symfony Messenger rappresenta il componente di riferimento dell'ecosistema Symfony per la gestione dei carichi di lavoro asincroni in PHP. La sua architettura poggia su tre pilastri: un message bus strutturato, transport dedicati e worker supervisionati. Con Symfony 7.3 e la roadmap verso la versione 8.0, il componente si arricchisce del middleware di deduplicazione, del transport AMQP streaming e del keepalive per Doctrine, raggiungendo una maturita funzionale paragonabile a sistemi specializzati come Laravel Horizon o Sidekiq.

Architettura di Symfony Messenger: Bus, Transport e Worker

Il componente Messenger opera separando tre responsabilita ben distinte: il dispatching tramite il bus, la consegna tramite il transport e l'elaborazione tramite il worker. Un messaggio viene definito come un semplice oggetto PHP privo di dipendenze esterne. Un handler corrisponde a una classe invocabile decorata con l'attributo #[AsMessageHandler]. Il bus collega messaggi e handler, instradando ogni dispatch attraverso una pipeline di middleware e, opzionalmente, serializzando il messaggio verso un transport per l'elaborazione differita.

namespace App\Message;

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

Il messaggio trasporta esclusivamente dati scalari, non entita Doctrine. Passare identificatori anziche oggetti evita problemi di serializzazione e mantiene i messaggi leggeri. L'handler recupera i dati aggiornati dal database al momento dell'elaborazione, garantendo coerenza anche in caso di ritardi nella coda.

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

L'invio del messaggio da un controller o un servizio si riduce a una singola istruzione:

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

Il bus determina se l'esecuzione avviene in modo sincrono o asincrono in base alla configurazione del routing dei transport. Qualora nessun transport risulti associato al messaggio, l'handler viene eseguito immediatamente nello stesso processo della richiesta HTTP.

Configurazione dei Transport e Backend delle Code

Messenger supporta diversi backend: Doctrine DBAL, Redis, Amazon SQS, Beanstalkd, AMQP (RabbitMQ) e il nuovo transport AMQP streaming introdotto nel 2025. Ogni transport viene definito tramite una stringa DSN nella configurazione YAML.

# 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

La suddivisione dei transport per priorita garantisce che le operazioni critiche per il business (generazione di fatture) vengano elaborate prima delle attivita di manutenzione ordinaria (pulizia dei file temporanei). Ogni transport dispone della propria strategia di retry, calibrata sulla criticita e sull'idempotenza dei messaggi che gestisce. Il transport failed basato su Doctrine funge da archivio per i messaggi non elaborabili, consentendo l'analisi e il retry manuale.

Gestione dei Worker con Supervisor

I worker consumano i messaggi dai transport e li inoltrano ai rispettivi handler. In ambiente di produzione, il comando messenger:consume viene eseguito sotto un process manager come Supervisor o systemd, che ne garantisce il riavvio automatico in caso di terminazione.

# 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

I tre flag di limitazione prevengono memory leak e assicurano il riavvio periodico dei worker. Il flag --memory-limit termina il processo al superamento della soglia di memoria allocata, --time-limit impone un limite temporale in secondi e --limit definisce il numero massimo di messaggi da elaborare prima dell'uscita. Supervisor si occupa di riavviare il processo dopo ogni terminazione controllata.

; /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

Impostando numprocs=2 si avviano due worker paralleli, raddoppiando il throughput di elaborazione. Il valore va regolato in base al numero di core CPU disponibili sul server e al tempo medio di elaborazione dei messaggi. Il parametro stopwaitsecs=30 concede ai worker un margine di 30 secondi per terminare l'elaborazione del messaggio corrente prima di un arresto forzato.

Pipeline dei Middleware e CQRS con Bus Multipli

I middleware avvolgono ogni operazione di dispatch, aggiungendo logiche trasversali all'intera applicazione. Lo stack di middleware integrato gestisce validazione, transazioni Doctrine e routing dei messaggi verso i transport configurati.

# 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

Questa configurazione implementa il pattern CQRS (Command Query Responsibility Segregation) attraverso la separazione in tre bus distinti. Il command bus gestisce le mutazioni di stato all'interno di una transazione Doctrine, garantendo atomicita delle operazioni di scrittura. Il query bus si occupa delle operazioni di sola lettura, senza il peso della transazione. L'event bus ammette zero handler tramite l'opzione allow_no_handlers, abilitando un pattern pub/sub in cui i listener possono essere aggiunti o rimossi in modo indipendente senza vincoli sulla presenza di sottoscrittori.

Middleware di Deduplicazione in Symfony 7.3

I messaggi duplicati rappresentano uno spreco di risorse computazionali e possono causare effetti collaterali gravi, come l'addebito doppio di un pagamento o l'invio multiplo della stessa notifica. Symfony 7.3 introduce il DeduplicateMiddleware per scartare automaticamente i messaggi identici gia presenti nella coda di elaborazione.

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

Il DeduplicateStamp accetta un identificatore univoco che funge da risorsa per il meccanismo di lock. Se un messaggio con lo stesso ID risulta gia in attesa di elaborazione, il nuovo dispatch viene silenziosamente scartato senza generare eccezioni. Questa funzionalita richiede il componente Lock configurato con uno store serializzabile come Redis, Memcached o database.

Strategie di Retry e Failure Transport

Quando un handler solleva un'eccezione durante l'elaborazione, Messenger ritenta automaticamente il messaggio secondo la strategia di retry configurata sul transport di appartenenza. Dopo aver esaurito tutti i tentativi disponibili, il messaggio viene spostato nel failure transport per l'analisi e il retry manuale da parte del team operativo.

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

RecoverableMessageHandlingException attiva la strategia di retry con backoff esponenziale, concedendo al sistema esterno il tempo di riprendersi da un errore transitorio. UnrecoverableMessageHandlingException bypassa completamente i retry e invia il messaggio direttamente al failure transport. Questa distinzione risulta fondamentale per evitare di sprecare tentativi su messaggi permanentemente non elaborabili, come nel caso di una carta di credito invalidata.

# 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 Messaggi a Lunga Esecuzione

Gli handler con tempi di elaborazione prolungati rischiano la riconsegna del messaggio quando il timeout di visibilita del transport scade prima del completamento dell'operazione. Symfony 7.2 ha introdotto il keepalive per Redis, SQS e Beanstalkd. Symfony 7.3 estende questa funzionalita al transport Doctrine, colmando una lacuna significativa per le applicazioni che utilizzano il database come broker di messaggi.

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

Il flag --keepalive aggiorna periodicamente il timestamp delivered_at nella tabella del transport Doctrine, segnalando che il worker sta ancora elaborando attivamente il messaggio. Senza questa opzione, un messaggio la cui elaborazione supera il timeout del transport (5 minuti per impostazione predefinita) viene prelevato da un altro worker, causando una doppia elaborazione. Questa funzionalita risulta particolarmente rilevante per operazioni come la generazione di report complessi, l'elaborazione batch di immagini o le migrazioni massive di dati.

L'Attributo #[AsMessage] per il Routing Dichiarativo

Symfony 7.2 ha introdotto l'attributo #[AsMessage], spostando la configurazione del routing dei transport dal file YAML centralizzato direttamente nella classe del messaggio.

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',
    ) {}
}

Questa sintassi elimina la necessita della sezione routing in messenger.yaml per il messaggio in questione. Il transport viene dichiarato direttamente nel codice sorgente, rendendo la codebase auto-documentante e riducendo il rischio di disallineamento tra configurazione e codice. Entrambi gli approcci (routing YAML e attributo PHP) possono coesistere nello stesso progetto; in caso di conflitto, l'attributo ha la precedenza.

Transport AMQP Streaming per Code ad Alto Throughput

Il transport AMQP tradizionale utilizza il polling con il metodo get() per recuperare i messaggi dalla coda, generando carico superfluo su RabbitMQ anche in assenza di messaggi da elaborare. Il transport AMQP streaming rilasciato nel 2025 adotta un modello push basato su consume(), riducendo significativamente latenza e consumo di risorse.

# 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.*']

Le differenze sostanziali rispetto al transport AMQP predefinito riguardano tre aspetti fondamentali. In primo luogo, non viene richiesta l'estensione C per AMQP: il transport si basa interamente su php-amqplib. In secondo luogo, la consegna avviene in streaming su connessioni TCP persistenti, eliminando l'overhead del polling periodico. Infine, il supporto nativo per topic exchange con routing tramite binding key consente pattern di distribuzione avanzati. Questo transport gestisce migliaia di messaggi al secondo con un overhead CPU minimale, rendendolo la scelta ideale per architetture event-driven ad alto volume.

Conclusione

• Inserire identificatori scalari nei messaggi, non entita Doctrine; recuperare i dati aggiornati nell'handler per evitare problemi di serializzazione e stato obsoleto
• Suddividere i transport per priorita e criticita; ogni transport necessita della propria strategia di retry e di un pool di worker dedicato
• Utilizzare RecoverableMessageHandlingException e UnrecoverableMessageHandlingException per controllare in modo esplicito il comportamento dei retry
• Abilitare --keepalive sui worker con transport Doctrine per prevenire la riconsegna di messaggi a lunga elaborazione
• Applicare il DeduplicateStamp sui messaggi in cui la doppia elaborazione genera effetti collaterali (pagamenti, email, notifiche)
• Implementare CQRS con bus multipli: command bus per le mutazioni con transazioni Doctrine, query bus per le letture, event bus per il pattern pub/sub
• Configurare Supervisor o systemd in produzione con i flag --memory-limit, --time-limit e --limit per una gestione robusta del ciclo di vita dei worker
• Valutare il transport AMQP streaming per scenari ad alto throughput che richiedono latenza nell'ordine dei sub-millisecondi