Symfony Messenger: как делать правильно и не терять сообщения

Вы вынесли отправку писем в фоновую очередь через Symfony Messenger — и через неделю обнаружили дубли: клиент получил три одинаковых письма. Или наоборот: часть сообщений исчезла бесследно после падения воркера. Обе проблемы решаются не переписыванием бизнес-логики, а правильной настройкой транспорта, повторов и обработчиков. Разберём, какие решения выигрышны и чем за них платишь. Ориентируюсь на Symfony 6.4 LTS и 7.x — базовое поведение Messenger в этих ветках совпадает.

Синхронно или асинхронно: где проходит граница

По умолчанию, если вы не настроили маршрутизацию, сообщение обрабатывается синхронно — в том же процессе, что и диспетчинг. Это удобно для локальной разработки, но не даёт ничего, ради чего берут Messenger. Асинхронность включается привязкой класса сообщения к транспорту в config/packages/messenger.yaml.

Первый вопрос, на который надо ответить до кода: что произойдёт, если обработчик упадёт. Если пользователь ждёт результат прямо сейчас (валидация формы) — это не задача для очереди. Если результат может появиться через секунду-другую и его отсутствие не ломает ответ (письмо, вебхук, пересчёт агрегата) — асинхронный транспорт оправдан.

Messenger не делает вашу систему надёжнее сам по себе. Он переносит момент, когда вы столкнётесь с потерей или дублем сообщения, — с момента запроса на момент сбоя воркера. Вопрос лишь в том, готовы ли вы к этому моменту.

Выбор транспорта

Транспорт — это то, где физически лежит очередь между диспетчингом и обработкой. От него зависят гарантии доставки и цена эксплуатации.

Для старта на существующем проекте с базой чаще всего разумен Doctrine-транспорт: не нужен новый сервис, легко отлаживать, сообщения видны в таблице messenger_messages. Когда поллинг БД начнёт мешать — переезжаете на RabbitMQ, поменяв DSN и почти не трогая код.

Конфигурация базовой связки

# config/packages/messenger.yaml (Symfony 6.4 / 7.x)
framework:
    messenger:
        transports:
            async:
                dsn: '%env(MESSENGER_TRANSPORT_DSN)%'
                retry_strategy:
                    max_retries: 3
                    delay: 1000        # мс до первого повтора
                    multiplier: 2      # каждый следующий повтор ждёт вдвое дольше
                    max_delay: 0
            failed:
                dsn: 'doctrine://default?queue_name=failed'
        failure_transport: failed
        routing:
            'App\Message\SendWelcomeEmail': async

DSN в .env для Doctrine-транспорта: MESSENGER_TRANSPORT_DSN=doctrine://default?auto_setup=1. Точные варианты DSN для каждого транспорта смотрите в документации Messenger — они различаются параметрами.

Повторы и мёртвая очередь

Если обработчик бросил исключение, Messenger по умолчанию вернёт сообщение в очередь и повторит согласно retry_strategy. После исчерпания попыток сообщение уходит в failure_transport — это и есть dead-letter очередь. Без неё исчерпавшее попытки сообщение просто теряется.

Две ошибки здесь встречаются постоянно:

• Повторяются невосстановимые ошибки. Если письмо не отправилось из-за невалидного адреса, три повтора ничего не изменят — только съедят время воркера. Бросайте UnrecoverableMessageHandlingException, чтобы сообщение сразу ушло в failed без повторов.
• Нет разбора failed-очереди. Сообщения копятся, и никто не знает. Смотреть их — командой php bin/console messenger:failed:show, повторить — messenger:failed:retry, удалить — messenger:failed:remove.

Идемпотентность: главное, что забывают

Ключевой факт: Messenger даёт гарантию at-least-once, а не exactly-once. Сообщение может быть доставлено обработчику более одного раза — например, если воркер обработал сообщение, но упал до того, как подтвердил (ack) его брокеру. После перезапуска сообщение придёт снова.

Значит, обработчик должен быть идемпотентным: повторная обработка того же сообщения не должна давать побочного эффекта дважды. Простой рабочий приём — таблица уже обработанных идентификаторов.

// src/MessageHandler/SendWelcomeEmailHandler.php

namespace App\MessageHandler;

use App\Message\SendWelcomeEmail;
use App\Repository\ProcessedMessageRepository;
use Symfony\Component\Messenger\Attribute\AsMessageHandler;
use Symfony\Component\Mailer\MailerInterface;

#[AsMessageHandler]
final class SendWelcomeEmailHandler
{
    public function __construct(
        private readonly MailerInterface $mailer,
        private readonly ProcessedMessageRepository $processed,
    ) {
    }

    public function __invoke(SendWelcomeEmail $message): void
    {
        // Если это сообщение уже обрабатывали — выходим без побочного эффекта
        if ($this->processed->exists($message->getIdempotencyKey())) {
            return;
        }

        // ... формирование и отправка письма через $this->mailer ...

        // Фиксируем факт обработки, чтобы повтор не отправил письмо снова
        $this->processed->markDone($message->getIdempotencyKey());
    }
}

Ключ идемпотентности кладите в само сообщение при его создании — например, UUID заказа или хеш от полезной нагрузки. Атрибут #[AsMessageHandler] доступен начиная с Symfony 6.2; на более старых ветках использовался интерфейс MessageHandlerInterface.

Транзакции и порядок операций

Частая ловушка: вы диспетчите сообщение внутри Doctrine-транзакции, а потом транзакция откатывается. Если транспорт синхронный или сообщение уже ушло в брокер — обработчик получит данные, которых в БД уже нет.

Решение — DispatchAfterCurrentBusMiddleware: оберните сообщение в DispatchAfterCurrentBusStamp, и оно отправится только после успешного завершения текущего сообщения на шине. Для связки с БД-транзакцией применяют DoctrineTransactionMiddleware, который оборачивает обработку в транзакцию. Точный состав middleware и порядок их регистрации сверяйте с документацией — он влияет на гарантии.

Запуск воркера в проде

Команда потребления — php bin/console messenger:consume async. В проде её нельзя запускать голой в цикле. Ключевые параметры:

• --limit=N — воркер завершится после N сообщений. Нужно, чтобы процесс не жил вечно и не накапливал утечки памяти.
• --memory-limit=128M — завершение при достижении лимита памяти.
• --time-limit=3600 — завершение по времени.

Перезапуск после завершения возлагается на супервизор — systemd или Supervisor. При деплое вызывайте messenger:stop-workers: воркеры дообработают текущее сообщение и корректно завершатся, после чего супервизор поднимет их уже с новым кодом. Иначе старый воркер продолжит крутить устаревшую логику.

Что ломается в реальности

• Долгие обработчики и long-running процесс. Воркер живёт долго, а Doctrine кеширует сущности. Устаревший EntityManager отдаёт старые данные. Помогает подписка на события Messenger и очистка/сброс менеджера между сообщениями либо перезапуск по --limit.
• Одно медленное сообщение блокирует очередь. Если обработка одного сообщения занимает минуты, остальные ждут. Разносите тяжёлые и лёгкие задачи по разным транспортам и запускайте отдельные воркеры.
• Сериализация «толстых» объектов. В сообщение кладут сущность целиком, при десериализации она уже не совпадает с БД. Кладите идентификатор и грузите свежие данные в обработчике.

Что настроить дальше

После базовой связки посмотрите на наблюдаемость: включите логирование через monolog для канала messenger, снимите метрику глубины очереди (для Doctrine — количество строк в messenger_messages, для RabbitMQ — через его метрики). Растущая глубина — сигнал, что воркеров не хватает. Инвалидация и приоритеты очередей — отдельная тема, здесь мы её не разворачивали. Следующий шаг для боевого проекта — алерт на размер failed-очереди: если туда что-то падает, вы должны узнать раньше клиента.