Symfony Messenger: Kolejki, Workery i Architektura Asynchroniczna — Pytania Rekrutacyjne 2026

Przetwarzanie asynchroniczne stanowi fundament nowoczesnych aplikacji webowych, a Symfony Messenger od kilku lat pelni role centralnego komponentu tego ekosystemu odpowiedzialnego za kolejkowanie i obsluge zadan w tle. Komponent ten implementuje wzorzec message bus, separujac wysylanie wiadomosci od ich przetwarzania — co pozwala na budowanie skalowalnych, odpornych na awarie systemow bez blokowania glownego watku HTTP. Symfony 7.3 i nadchodzace Symfony 8.0 wprowadzaja szereg usprawnien, dzieki ktorym Messenger staje sie pelnoprawna alternatywa dla dedykowanych systemow kolejkowych pokroju Laravel Horizon czy Sidekiq. Ponizszy artykul omawia architekture, konfiguracje produkcyjna oraz zaawansowane wzorce, ktore pojawiaja sie na rozmowach rekrutacyjnych w 2026 roku.

Architektura Symfony Messenger: Bus, Transport i Worker

Messenger rozdziela trzy fundamentalne odpowiedzialnosci w architekturze asynchronicznej. Bus pelni role dyspozytora — przyjmuje wiadomosc i przepuszcza ja przez lancuch middleware. Transport odpowiada za fizyczne dostarczenie wiadomosci do backendu kolejki (Doctrine, Redis, RabbitMQ). Worker to dlugotrwaly proces PHP, ktory pobiera wiadomosci z transportu i przekazuje je do odpowiednich handlerow.

Wiadomosc (message) to zwykly obiekt PHP, ktory powinien zawierac wylacznie dane skalarne — identyfikatory numeryczne i ciagi znakow zamiast encji Doctrine. Przekazywanie identyfikatorow eliminuje problemy z serializacja i gwarantuje, ze handler operuje na aktualnych danych pobieranych z bazy w momencie przetwarzania.

namespace App\Message;

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

Klasa wiadomosci jest oznaczona jako final readonly, co gwarantuje niemutowalnosc — raz utworzony obiekt nie moze zostac zmodyfikowany. Handler rejestruje sie automatycznie w kontenerze DI dzieki atrybutowi #[AsMessageHandler] i implementuje wzorzec callable z metoda __invoke:

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

Wyslanie wiadomosci z kontrolera lub serwisu ogranicza sie do jednej instrukcji. O tym, czy przetwarzanie nastapi synchronicznie czy asynchronicznie, decyduje konfiguracja routingu transportow — kod wysylajacy nie musi o tym wiedziec:

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

Ta separacja jest kluczowa z perspektywy testowalnosci. Testy jednostkowe moga weryfikowac wyslanie wiadomosci bez uruchamiania calej infrastruktury kolejkowej, a zmiana backendu kolejki nie wymaga modyfikacji kodu biznesowego.

Konfiguracja transportow i backendow kolejek

Messenger obsluguje szesc backendow kolejek: Doctrine DBAL, Redis, Amazon SQS, Beanstalkd, AMQP (RabbitMQ) oraz streaming AMQP transport wprowadzony w 2025 roku. Kazdy transport definiowany jest jako DSN string w zmiennej srodowiskowej, a konfiguracja YAML umozliwia precyzyjne dostrojenie strategii ponawiania i priorytetow.

# 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

Podzial na transporty o roznym priorytecie zapewnia, ze zadania krytyczne czasowo (generowanie faktur, potwierdzenia platnosci) sa przetwarzane przed operacjami porzadkowymi (czyszczenie plikow tymczasowych, generowanie raportow). Parametr multiplier odpowiada za wykladniczy wzrost opoznienia miedzy kolejnymi probami ponawiania — przy wartosci 3 i delay: 1000 opoznienia wynosza kolejno 1s, 3s i 9s. Parametr max_delay ogranicza maksymalne opoznienie do 60 sekund, zapobiegajac zbyt dlugim przerwom miedzy probami.

Zarzadzanie workerami z Supervisorem

Workery to procesy PHP dzialajace w tle, ktore w petli pobieraja wiadomosci z transportow i przekazuja je do handlerow. W srodowisku produkcyjnym komenda messenger:consume powinna dzialac pod nadzorem process managera — Supervisor lub systemd — ktory automatycznie restartuje proces po jego zakonczeniu.

# 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

Trzy flagi limitujace pelnia kluczowa role w stabilnosci produkcyjnej. Flaga --memory-limit konczy proces po przekroczeniu 128 MB pamieci, zapobiegajac wyciekom typowym dla dlugotrwalych procesow PHP. Opcja --time-limit wymusza restart co godzine, co umozliwia wdrazanie nowych wersji kodu bez recznej interwencji. Parametr --limit okresla maksymalna liczbe przetworzonych wiadomosci, po ktorej worker gracefully sie konczy. Supervisor natychmiast uruchamia nowa instancje.

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

Parametr numprocs=2 uruchamia dwie rownolegle instancje workera, podwajajac przepustowosc przetwarzania. Wartosc te nalezy dostosowac do liczby rdzeni CPU i sredniej dlugosci przetwarzania wiadomosci. Opcja stopwaitsecs=30 daje workerowi 30 sekund na zakonczenie biezacej wiadomosci przed wymuszonym zakonczeniem procesu — unika sie w ten sposob utraty przetwarzanych danych.

Pipeline middleware i CQRS z wieloma busami

Middleware tworzy lancuch przetwarzania otaczajacy kazda wysylana wiadomosc. Kazdy element lancucha moze wykonywac operacje przed i po handlerze, modyfikowac koperte (envelope) wiadomosci, dodawac stampe lub calkowicie przerwac przetwarzanie. Wbudowane middleware obsluguja walidacje, transakcje Doctrine oraz routing do transportow.

Wzorzec CQRS (Command Query Responsibility Segregation) wymaga separacji operacji modyfikujacych stan od operacji odczytu. Symfony Messenger natywnie wspiera ten wzorzec poprzez definiowanie wielu busow, z ktorych kazdy posiada dedykowany zestaw middleware:

# 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

Command bus obsluguje komendy modyfikujace stan aplikacji. Middleware doctrine_transaction gwarantuje, ze caly handler jest wykonywany wewnatrz transakcji bazodanowej — w przypadku wyjatku wszystkie zmiany sa automatycznie wycofywane. Query bus sluzy wylacznie do odczytu danych i nie wymaga transakcji. Event bus realizuje wzorzec pub/sub — opcja allow_no_handlers: true oznacza, ze zdarzenie moze nie miec zadnego subskrybenta, co jest naturalne w systemach opartych na zdarzeniach domenowych. Taka architektura pozwala na niezalezne skalowanie operacji odczytu i zapisu oraz ulatwia testowanie warstw w izolacji.

Middleware deduplikacji w Symfony 7.3

Duplikaty wiadomosci stanowia jeden z najczestszych problemow w systemach asynchronicznych. Uzytkownik klikajacy przycisk wielokrotnie, zdarzenia emitowane z roznych czesci systemu, retransmisje sieciowe, race conditions przy wysokim obciazeniu — kazdy z tych scenariuszy moze prowadzic do podwojnego przetworzenia z powaznymi konsekwencjami biznesowymi: podwojne obciazenie karty platniczej, wyslanie dwoch identycznych e-maili powitalnych lub utworzenie duplikatu zamowienia.

Symfony 7.3 wprowadza DeduplicateMiddleware, ktory rozwiazuje ten problem na poziomie infrastruktury kolejkowej. Mechanizm opiera sie na komponencie Lock i wymaga serializowalnego backendu blokad (Redis, Memcached lub bazy danych).

namespace App\Message;

use Symfony\Component\Messenger\Stamp\DeduplicateStamp;

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

Deduplikacja aktywowana jest poprzez dolaczenie DeduplicateStamp podczas wysylania wiadomosci. Identyfikator deduplikacji musi jednoznacznie identyfikowac dana operacje biznesowa:

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

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

Jesli wiadomosc z identycznym ID jest juz w kolejce i oczekuje na przetworzenie, nowa wiadomosc zostanie cicho odrzucona. System przechowuje identyfikatory przez konfigurowalny czas, po ktorym ten sam ID moze byc uzyty ponownie. W praktyce identyfikator powinien zawierac typ operacji i identyfikator encji, np. welcome-email-42 lub payment-confirmation-1337, co zapewnia unikalnosc w kontekscie biznesowym.

Strategie ponawiania i transport bledow

Obsluga bledow w Messenger opiera sie na kluczowym rozroznieniu miedzy bledami odzyskiwalnymi (recoverable) a nieodzyskiwalnymi (unrecoverable). Precyzyjne zarzadzanie typem bledu pozwala uniknac marnowania prob ponawiania na wiadomosci, ktore nigdy nie zostana przetworzone pomyslnie.

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 sygnalizuje problem tymczasowy — timeout bramki platniczej, chwilowa niedostepnosc zewnetrznego API — i uruchamia strategie ponawiania z wykladniczym cofaniem (exponential backoff). UnrecoverableMessageHandlingException omija mechanizm retry calkowicie i natychmiast przenosi wiadomosc do transportu bledow. Stosowanie tego wyjatku jest wlasciwe dla sytuacji trwalych: nieprawidlowy numer karty, brakujace dane wejsciowe, naruszenie regul biznesowych.

Zarzadzanie nieudanymi wiadomosciami odbywa sie przez zestaw komend CLI. Symfony 7.3 wprowadza filtrowanie po klasie wiadomosci, co znaczaco ulatwia prace z duzymi zbiorami bledow w produkcji:

# 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 i dlugotrwale wiadomosci

Dlugotrwale przetwarzanie wiadomosci generuje dwa powazne problemy produkcyjne. Po pierwsze, wygasniecie visibility timeout transportu powoduje, ze ta sama wiadomosc zostaje ponownie dostarczona do innego workera, co prowadzi do duplikatu przetwarzania. Po drugie, dlugotrwaly proces moze utracic polaczenie z baza danych Doctrine, co objawia sie bledem "MySQL server has gone away" lub "SQLSTATE[HY000]: General error: 2006".

Symfony 7.2 wprowadzil mechanizm keepalive dla transportow Redis, SQS i Beanstalkd. Symfony 7.3 rozszerza te funkcjonalnosc na transport Doctrine, co jest istotne dla zespolow korzystajacych z bazy danych jako backendu kolejki.

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

Flaga --keepalive okresowo aktualizuje znacznik czasowy delivered_at w tabeli transportu Doctrine, sygnalizujac innym workerom, ze wiadomosc jest aktywnie przetwarzana. Bez tej flagi wiadomosc przetwarzana dluzej niz domyslny timeout (5 minut) zostanie pobrana przez innego workera. Mechanizm jednoczesnie utrzymuje aktywne polaczenie z baza danych Doctrine, eliminujac bledy zwiazane z zamknietymi polaczeniami podczas dlugiego przetwarzania.

Atrybut #[AsMessage] — deklaratywny routing transportu

Tradycyjnie routing wiadomosci do transportow konfigurowany jest w sekcji routing pliku messenger.yaml. Symfony 7.2 wprowadzil alternatywne podejscie — atrybut #[AsMessage] przypisywany bezposrednio klasie wiadomosci. Takie rozwiazanie czyni kod samodokumentujacym sie: programista odczytuje transport danej wiadomosci bez koniecznosci przeszukiwania plikow konfiguracyjnych.

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

Atrybut #[AsMessage] eliminuje wpis w sekcji routing pliku messenger.yaml dla danego typu wiadomosci. Oba podejscia — routing YAML i atrybut PHP — moga wspolistniec w ramach jednego projektu. W przypadku konfliktu atrybut ma pierwszenstwo. W duzych projektach z setkami typow wiadomosci deklaratywny routing znaczaco upraszcza zarzadzanie konfiguracja i zmniejsza ryzyko bledow wynikajacych z niespojnosci miedzy kodem a plikami YAML.

Streaming AMQP Transport dla kolejek o wysokiej przepustowosci

Domyslny transport AMQP w Symfony korzysta z metody pollingu (basic.get) do pobierania wiadomosci z RabbitMQ, co generuje niepotrzebne obciazenie sieciowe i latencje. Streaming AMQP transport, wydany w 2025 roku, przelacza sie na model push (basic.consume), w ktorym RabbitMQ aktywnie dostarcza wiadomosci do workera przez dlugotrwale polaczenie TCP.

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

Kluczowe roznice wzgledem tradycyjnego transportu AMQP: brak wymagania rozszerzenia C (transport wykorzystuje biblioteke php-amqplib), dostarczanie strumieniowe przez dlugotrwale polaczenia TCP zamiast cyklicznego odpytywania oraz natywna obsluga topic exchange z routingiem opartym na binding keys. Powyzszy przyklad konfiguruje topic exchange z kolejka subskrybujaca wszystkie zdarzenia pasujace do wzorca order.* — np. order.created, order.shipped, order.cancelled. Transport ten obsluguje tysiace wiadomosci na sekunde przy minimalnym narzucie CPU, co czyni go optymalnym rozwiazaniem dla architektur event-driven i systemow o wysokiej przepustowosci.

Podsumowanie

Symfony Messenger w wersji 7.3 oferuje dojrzaly i kompletny ekosystem do przetwarzania asynchronicznego w PHP. Ponizej znajduje sie zestawienie kluczowych praktyk i wnioskow, ktore czesto pojawiaja sie na rozmowach rekrutacyjnych:

• Wiadomosci powinny zawierac wylacznie skalarne identyfikatory, nie encje Doctrine — handler pobiera aktualne dane z bazy w momencie przetwarzania, co eliminuje problemy z serializacja i stalym stanem
• Transporty nalezy podzielic wedlug priorytetu i krytycznosci; kazdy transport powinien posiadac wlasna strategie ponawiania i dedykowana pule workerow
• RecoverableMessageHandlingException i UnrecoverableMessageHandlingException zapewniaja pelna kontrole nad zachowaniem retry, eliminujac marnowanie prob na trwale uszkodzone wiadomosci
• Flaga --keepalive na transportach Doctrine zapobiega ponownemu dostarczeniu dlugo przetwarzanych wiadomosci i utrzymuje aktywne polaczenie z baza
• DeduplicateStamp nalezy stosowac wszedzie tam, gdzie podwojne przetworzenie powoduje efekty uboczne — platnosci, e-maile, powiadomienia push
• CQRS z wieloma busami umozliwia separacje komend (z transakcjami Doctrine), zapytan (bez efektow ubocznych) i zdarzen (pub/sub bez wymagania handlerow)
• W srodowisku produkcyjnym workery musza byc zarzadzane przez Supervisor lub systemd z flagami --memory-limit, --time-limit i --limit kontrolujacymi cykl zycia procesu
• Streaming AMQP transport stanowi optymalne rozwiazanie dla scenariuszy wymagajacych wysokiej przepustowosci i niskiej latencji, z pelna obsluga topic exchange i routingu opartego na binding keys