Symfony Messenger: черги, воркери та асинхронна архітектура у 2026 році

Компонент Symfony Messenger перетворює PHP-додатки на повноцінні асинхронні системи, забезпечуючи обробку задач через структуровану шину повідомлень, конфігуровані транспорти та керовані воркери. Починаючи з Symfony 7.3 та 8.0, де з'явилися middleware дедуплікації, streaming AMQP-транспорт і Doctrine keepalive, цей компонент досяг функціональної зрілості, порівнянної зі спеціалізованими системами черг на кшталт Laravel Horizon чи Sidekiq. Розуміння внутрішньої архітектури Messenger, механізмів гарантії доставки та стратегій масштабування воркерів залишається критичною компетенцією для PHP-розробників, оскільки ці теми регулярно піднімаються на технічних співбесідах у 2026 році.

Архітектура Symfony Messenger: шина, транспорт та воркер

Messenger побудований на чіткому розділенні трьох відповідальностей. Шина (bus) приймає повідомлення та проганяє його через ланцюжок middleware. Транспорт (transport) відповідає за фізичну доставку повідомлення до брокера черг. Воркер (worker) вичитує повідомлення з транспорту та передає його відповідному обробнику. Повідомлення --- це звичайний PHP-об'єкт без зовнішніх залежностей, а обробник --- invokable-клас, що містить бізнес-логіку.

namespace App\Message;

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

Повідомлення несе виключно скалярні дані --- ідентифікатори та примітивні типи. Передача Doctrine-сутностей всередині повідомлень призводить до проблем серіалізації та ризику працювати із застарілим станом об'єкта. Обробник завжди завантажує свіжі дані з бази на момент фактичного виконання.

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

Відправка повідомлення з контролера або сервісу зводиться до єдиного виклику:

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

Рішення про синхронне чи асинхронне виконання приймає шина на основі конфігурації маршрутизації транспортів. Якщо для класу повідомлення транспорт не призначено, обробка відбувається синхронно у контексті поточного HTTP-запиту. Саме цей принцип забезпечує прозорість архітектури: код обробника не містить жодних відомостей про спосіб доставки, що дозволяє перемикатися між режимами виключно через конфігурацію.

Конфігурація транспортів та бекенди черг

Messenger підтримує широкий набір бекендів: Doctrine DBAL, Redis, Amazon SQS, Beanstalkd, AMQP (RabbitMQ) та streaming AMQP-транспорт, представлений у 2025 році. Кожен транспорт визначається через 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

Поділ на кілька транспортів із різними пріоритетами гарантує, що термінові операції (наприклад, формування рахунку) обробляються раніше за фонові задачі (очищення тимчасових файлів). Кожен транспорт має власну стратегію повторних спроб: параметр multiplier задає коефіцієнт експоненційного наростання затримки між спробами. Для високопріоритетного транспорту із множником 3 затримки становлять 1, 3 та 9 секунд, тоді як для низькопріоритетного із множником 2 --- 5, 10, 20, 40 та 80 секунд.

Управління воркерами через Supervisor

Воркери --- це довготривалі процеси, що споживають повідомлення з транспортів та передають їх обробникам. У продакшн-середовищі команда messenger:consume працює під управлінням процес-менеджера Supervisor або 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

Три прапорці обмеження виконують захисну функцію: --memory-limit зупиняє воркер при досягненні порогу споживання пам'яті, --time-limit обмежує максимальний час роботи одного процесу, а --limit задає верхню межу кількості оброблених повідомлень. Після зупинки за будь-яким із цих критеріїв Supervisor негайно створює новий процес.

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

Директива numprocs=2 запускає два паралельні воркери, що подвоює пропускну здатність системи. Оптимальна кількість воркерів визначається кількістю доступних ядер CPU та середнім часом обробки одного повідомлення. Параметр stopwaitsecs=30 надає воркеру 30 секунд для коректного завершення поточної обробки перед примусовою зупинкою процесу.

Pipeline middleware та CQRS із кількома шинами

Middleware утворюють конвеєр, через який проходить кожне повідомлення під час диспетчеризації. Кожен middleware отримує конверт (envelope) із повідомленням та набором stamps, може модифікувати його, додати нові stamps або перервати ланцюжок виконання. Конфігурація кількох незалежних шин дозволяє реалізувати архітектурний патерн CQRS.

# 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) обгортає виконання у Doctrine-транзакцію та виконує валідацію вхідних даних, гарантуючи атомарність операцій запису. Шина запитів (query.bus) обслуговує виключно операції читання без транзакційного контексту. Шина подій (event.bus) допускає нуль або кілька обробників для одного повідомлення, реалізуючи патерн pub/sub: нові слухачі додаються без будь-яких змін у наявному коді. Така архітектурна організація є стандартним підходом для масштабних Symfony-проєктів.

Middleware дедуплікації в Symfony 7.3

Дублювання повідомлень у черзі здатне спричинити серйозні побічні ефекти: подвійне списання коштів з рахунку клієнта, повторне надсилання електронних листів або створення дублікатів записів у базі даних. Symfony 7.3 вирішує цю проблему через DeduplicateMiddleware, який автоматично відхиляє повідомлення, тотожне тому, що вже перебуває в черзі на обробку.

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

DeduplicateStamp приймає рядковий ідентифікатор ресурсу блокування. Якщо повідомлення з таким ідентифікатором вже очікує обробки, нова диспетчеризація тихо ігнорується без генерації помилки. Механізм спирається на компонент Lock із серіалізованим сховищем (Redis, Memcached або реляційна база). Блокування автоматично знімається після обробки або видалення повідомлення з черги, відкриваючи можливість повторної відправки з тим самим ідентифікатором у майбутньому.

Стратегії повторних спроб та failure-транспорт

Коли обробник генерує виняток, Messenger автоматично повторює доставку повідомлення відповідно до стратегії retry, сконфігурованої для конкретного транспорту. Після вичерпання усіх спроб повідомлення переміщується до failure-транспорту, де зберігається разом із повною історією невдалих спроб, текстом виключення та всіма stamps конверта, що суттєво спрощує подальшу діагностику.

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 сигналізує системі, що збій має тимчасовий характер і повідомлення потрібно повторити з експоненційною затримкою. UnrecoverableMessageHandlingException натомість повністю минає механізм повторних спроб та негайно направляє повідомлення до failure-транспорту. Таке явне розмежування запобігає марному витрачанню ресурсів на повторення безповоротно пошкоджених повідомлень --- наприклад, платіж з недійсною карткою ніколи не буде успішним незалежно від кількості спроб.

# 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 для довготривалих повідомлень

Коли обробка повідомлення займає значний час --- генерація складного звіту, обробка великого масиву даних або взаємодія з повільним зовнішнім API --- виникає ризик повторної доставки через закінчення тайм-ауту видимості транспорту. Symfony 7.2 запровадив keepalive для Redis, SQS та Beanstalkd, а Symfony 7.3 розширив цей механізм на Doctrine-транспорт.

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

Прапорець --keepalive змушує воркер періодично оновлювати мітку часу delivered_at у таблиці Doctrine-транспорту. Це інформує інші воркери, що повідомлення все ще перебуває в активній обробці. Без keepalive повідомлення, яке обробляється довше за тайм-аут видимості (типово 5 хвилин), буде захоплене іншим воркером, що призведе до дублювання обробки. Для неідемпотентних операцій --- фінансових транзакцій, відправки повідомлень, модифікації зовнішніх систем --- наслідки можуть бути критичними.

Атрибут #[AsMessage] для декларативної маршрутизації

Атрибут #[AsMessage], представлений у Symfony 7.2, виносить прив'язку повідомлення до транспорту безпосередньо у PHP-клас, усуваючи потребу в YAML-конфігурації маршрутизації.

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

При використанні атрибута секція routing у messenger.yaml для цього типу повідомлення стає непотрібною. Транспорт оголошується безпосередньо біля визначення повідомлення, що робить кодову базу самодокументованою. Обидва механізми маршрутизації --- YAML та атрибут --- працюють одночасно, проте атрибут має вищий пріоритет. У великих проєктах із десятками або сотнями типів повідомлень такий підхід суттєво спрощує навігацію та зменшує когнітивне навантаження на розробників.

Streaming AMQP-транспорт для високопропускних сценаріїв

Класичний AMQP-транспорт у Symfony Messenger працює через механізм polling: воркер періодично надсилає запит get() до RabbitMQ, що створює додаткове мережеве навантаження навіть за відсутності нових повідомлень. Streaming AMQP-транспорт, випущений у 2025 році, переходить на push-модель через виклик consume(), де RabbitMQ сам проштовхує повідомлення до воркера через довготривале 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.*']

Основні переваги streaming-транспорту: відсутність необхідності у C-розширенні (замість нього використовується бібліотека php-amqplib), потокова доставка через постійні TCP-з'єднання та нативна підтримка topic exchanges із маршрутизацією за binding keys. Такий транспорт здатний обробляти тисячі повідомлень на секунду з мінімальним навантаженням на процесор. Для систем з високими вимогами до пропускної здатності --- обробка подій e-commerce, агрегація логів, real-time аналітика --- streaming AMQP залишається найбільш ефективним рішенням серед доступних транспортів.

Висновок

Ключові принципи роботи з Symfony Messenger для продакшн-розробки та підготовки до технічних співбесід:

• Передавати у повідомленнях лише скалярні ідентифікатори, а не Doctrine-сутності; завантажувати актуальні дані безпосередньо в обробнику для уникнення проблем серіалізації та застарілого стану
• Розділяти транспорти за рівнями пріоритету та критичності; кожен транспорт конфігурується з індивідуальною стратегією повторних спроб та окремим пулом воркерів
• Явно контролювати поведінку retry через RecoverableMessageHandlingException та UnrecoverableMessageHandlingException
• Вмикати --keepalive для воркерів на Doctrine-транспорті, щоб запобігти повторній доставці під час тривалої обробки
• Використовувати DeduplicateStamp для повідомлень, де дублювання створює побічні ефекти: платежі, сповіщення, генерація документів
• Організовувати CQRS через кілька шин: командна шина з Doctrine-транзакціями для мутацій, шина запитів для читання, шина подій для pub/sub із необов'язковими обробниками
• Запускати воркери під управлінням Supervisor або systemd із прапорцями --memory-limit, --time-limit та --limit для контролю життєвого циклу процесів
• Обирати streaming AMQP-транспорт для високопропускних сценаріїв, що вимагають мінімальної затримки та ефективного використання ресурсів