Symfony Messenger на практике: от настройки маршрутизации и retry до спасения сообщений из failed transport

Асинхронная обработка задач снижает время ответа веб-приложения и помогает масштабировать систему под нагрузкой. Однако при переходе на очереди неизбежно возникают сбои: сторонние API перестают отвечать, базы данных сталкиваются с deadlock, а невалидные входные данные приводят к фатальным исключениям.

Компонент Symfony Messenger предоставляет готовый инструментарий для построения отказоустойчивой архитектуры: от декларативной маршрутизации и экспоненциальных повторов (retry) до сохранения упавших задач в failure_transport и безопасного перезапуска воркеров.

---

1. Базовые сущности Messenger: сообщение и обработчик

Архитектура Symfony Messenger строится вокруг четкого разделения данных и логики их обработки.

Создание класса DTO (Message)

Сообщение — это простой PHP-объект (DTO), который сериализуется и передается в шину сообщений. Он должен содержать только те данные, которые необходимы для выполнения конкретной операции:

// src/Message/ProcessOrderPayment.php
namespace App\Message;

final readonly class ProcessOrderPayment
{
    public function __construct(
        public int $orderId,
        public int $amountInCents,
        public string $currency = 'RUB'
    ) {}
}

Реализация Handler через атрибут #[AsMessageHandler]

Обработчик содержит бизнес-логику. В современных версиях Symfony класс помечается атрибутом #[AsMessageHandler], а тип аргумента в методе __invoke() указывает шине, какое именно сообщение должен принимать данный обработчик:

// src/MessageHandler/ProcessOrderPaymentHandler.php
namespace App\MessageHandler;

use App\Message\ProcessOrderPayment;
use Psr\Log\LoggerInterface;
use Symfony\Component\Messenger\Attribute\AsMessageHandler;

#[AsMessageHandler]
final class ProcessOrderPaymentHandler
{
    public function __construct(
        private LoggerInterface $logger
    ) {}

    public function __invoke(ProcessOrderPayment $message): void
    {
        $this->logger->info('Обработка платежа', [
            'order_id' => $message->orderId,
            'amount' => $message->amountInCents,
        ]);

        // Бизнес-логика списания средств
    }
}

Проверка зарегистрированных обработчиков

После добавления классов проверить корректность их связывания можно через консоль:

php bin/console debug:messenger

Команда выведет список всех обнаруженных сообщений, шины и соответствующие классы обработчиков.

---

2. Маршрутизация и асинхронный транспорт

По умолчанию сообщение, отправленное через шину (MessageBusInterface::dispatch()), выполняется синхронно прямо в текущем HTTP-запросе или консольной команде. Чтобы перевести задачу в очередь, необходимо настроить транспорт и правила маршрутизации.

Настройка транспортов и правил в messenger.yaml

# config/packages/messenger.yaml
framework:
    messenger:
        transports:
            async: '%env(MESSENGER_TRANSPORT_DSN)%'

        routing:
            'App\Message\ProcessOrderPayment': async

Правила маршрутизации поддерживают:

• конкретные классы сообщений;
• интерфейсы (все сообщения, реализующие интерфейс, уйдут в указанный транспорт);
• родительские классы или подстановочные знаки (wildcards).

Если класс сообщения не подпадает ни под одно правило секции routing, оно выполняется синхронно.

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

Для извлечения и обработки сообщений из транспорта запускается фоновый процесс:

php bin/console messenger:consume async -vv --memory-limit=256M --time-limit=3600 --limit=1000

Ключевые флаги контроля стабильности:

• --memory-limit=256M — штатно завершает процесс при превышении лимита оперативной памяти (защита от утечек);
• --time-limit=3600 — останавливает воркер через указанное время (в секундах), позволяя менеджеру процессов перезапустить чистый инстанс;
• --limit=1000 — остановка после обработки заданного количества сообщений;
• --failure-limit=5 — аварийная остановка воркера, если получено указанное число ошибок подряд.

---

3. Стратегия повторов (Retry Strategy) при сбоях

Временные сбои (сетевые таймауты, кратковременные блокировки в базе данных) не должны приводить к мгновенной потере задачи. Messenger поддерживает настройку политики повторов на уровне каждого транспорта.

Конфигурация retry, multiplier и jitter

# config/packages/messenger.yaml
framework:
    messenger:
        transports:
            async:
                dsn: '%env(MESSENGER_TRANSPORT_DSN)%'
                retry_strategy:
                    max_retries: 3
                    delay: 1000        # Начальная задержка: 1000 мс (1 сек)
                    multiplier: 2       # Удвоение задержки при каждом следующем сбое
                    max_delay: 10000    # Максимальный интервал задержки: 10 сек
                    jitter: 0.1         # Случайное отклонение +/- 10%

• multiplier реализует экспоненциальную задержку (exponential backoff): 1 с -> 2 с -> 4 с.
• jitter вносит случайное смещение в расчет задержки. Это защищает инфраструктуру от эффекта Thundering Herd (лавинообразного наплыва повторных запросов на восстанавливающийся сервис от десятков параллельных воркеров одновременно).

Управление поведением через исключения

Поведением очереди можно управлять непосредственно из кода обработчика с помощью встроенных классов исключений:

1. UnrecoverableMessageHandlingException — сообщает Messenger, что повторять операцию бессмысленно (например, данные повреждены или сущность удалена). Сообщение немедленно прекращает попытки retry и перемещается в failure_transport.
2. RecoverableMessageHandlingException — заставляет воркер отправить сообщение на повтор даже в тех случаях, когда исчерпан лимит max_retries. В конструктор можно передать индивидуальное время задержки (например, из HTTP-заголовка Retry-After).

use Symfony\Component\Messenger\Exception\UnrecoverableMessageHandlingException;

public function __invoke(ProcessOrderPayment $message): void
{
    if ($message->amountInCents <= 0) {
        throw new UnrecoverableMessageHandlingException('Некорректная сумма платежа.');
    }
    
    // Выполнение списания
}

---

4. Failed Transport: защита от потери данных

Важно: Если лимит повторов max_retries исчерпан, а failure_transport не настроен ни глобально, ни для конкретного транспорта, сообщение удаляется навсегда.

Выделение очереди сбоев

Для изоляции проблемных задач настраивается транспорт сбоев:

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

        transports:
            async:
                dsn: '%env(MESSENGER_TRANSPORT_DSN)%'
            
            failed: 'doctrine://default?queue_name=failed'

При исчерпании лимита попыток сообщение вместе со стеком вызовов и метаданными ошибки перемещается в очередь failed.

Инспекция и работа с упавшими сообщениями

Для анализа и восстановления данных используется семейство консольных команд messenger:failed:*:

# Просмотр списка всех упавших сообщений
php bin/console messenger:failed:show

# Детальная инспекция сообщения (с трейсом ошибки и дампом DTO)
php bin/console messenger:failed:show 42 -vv

# Интерактивный повтор (с запросом подтверждения для каждого элемента)
php bin/console messenger:failed:retry -vv

# Принудительный повтор конкретных ID
php bin/console messenger:failed:retry 42 43 --force

# Удаление сообщений без повторного выполнения
php bin/console messenger:failed:remove 42
php bin/console messenger:failed:remove --all

---

5. Практический сценарий: тест падения обработчика и восстановление

Разберем полный жизненный цикл: генерация сбоя, исчерпание попыток, переход в failed-очередь и успешное восстановление.

Шаг 1. Создание искусственного сбоя

Добавим в обработчик исключение:

// src/MessageHandler/ProcessOrderPaymentHandler.php
namespace App\MessageHandler;

use App\Message\ProcessOrderPayment;
use RuntimeException;
use Symfony\Component\Messenger\Attribute\AsMessageHandler;

#[AsMessageHandler]
final class ProcessOrderPaymentHandler
{
    public function __invoke(ProcessOrderPayment $message): void
    {
        // Имитируем падение внешнего платежного шлюза
        throw new RuntimeException('Платежный шлюз временно недоступен: 503 Service Unavailable');
    }
}

Шаг 2. Запуск воркера и наблюдение за retry

Отправляем сообщение в шину и запускаем воркер с подробным выводом:

php bin/console messenger:consume async -vv

В консоли зафиксируется цепочка событий:

1. Попытка 1: исключение RuntimeException -> повтор отложен на 1000 мс.
2. Попытка 2: исключение RuntimeException -> повтор отложен на 2000 мс.
3. Попытка 3: исключение RuntimeException -> повтор отложен на 4000 мс.
4. Исчерпание лимита -> ProcessOrderPayment has been sent to the failure transport (failed).

Шаг 3. Инспекция ошибки

Проверяем наличие записи в failed-транспорте:

php bin/console messenger:failed:show

Вывод отобразит ID сообщения, класс App\Message\ProcessOrderPayment и время возникновения сбоя.

Шаг 4. Устранение бага и повторная отправка

Убираем генерацию исключения из кода обработчика и отправляем упавшую задачу на повторное исполнение:

php bin/console messenger:failed:retry 1 --force -vv

Сообщение извлекается из хранилища failed, успешно выполняется обработчиком и очищается из очереди ошибок.

---

6. Эксплуатация в Production: Graceful Restart воркеров

Воркеры Messenger работают как непрерывные демоны (long-running processes). Код приложения кэшируется в оперативной памяти при старте. При деплое нового релиза процессы необходимо перезапускать, иначе они продолжат исполнять старую версию кодовой базы.

Почему нельзя использовать kill -9

Принудительное завершение процесса сигналом SIGKILL обрывает выполнение посреди транзакции. Это может привести к рассинхронизации данных (например, запись в сторонней системе создана, а статус в БД не обновился).

Механизм messenger:stop-workers

Symfony решает задачу мягкой остановки (graceful shutdown) через метку времени в системном кэше:

php bin/console messenger:stop-workers

1. Команда записывает timestamp остановки в пул кэша (cache.app).
2. Воркер перед получением очередного сообщения сравнивает время своего старта с меткой в кэше.
3. Если метка новее, воркер завершает текущую задачу, закрывает сетевые соединения и корректно выходит с кодом 0.

Связка с Supervisor

В продакшене воркеры запускаются под управлением менеджера процессов (Supervisor или systemd). Супервизор фиксирует остановку процесса и запускает свежий инстанс.

Пример конфигурации Supervisor (/etc/supervisor/conf.d/messenger.conf):

[program:messenger-worker]
command=php /var/www/app/bin/console messenger:consume async --memory-limit=256M --time-limit=3600
process_name=%(program_name)s_%(process_num)02d
numprocs=4
autostart=true
autorestart=true
startsecs=5
stopwaitsecs=30
user=www-data
redirect_stderr=true
stdout_logfile=/var/log/supervisor/messenger.log

Рекомендуемый пайплайн деплоя:

1. Развернуть новую версию кода.
2. Применить миграции базы данных: php bin/console doctrine:migrations:migrate.
3. Очистить и прогреть кэш: php bin/console cache:clear.
4. Отправить сигнал остановки: php bin/console messenger:stop-workers.
5. Supervisor автоматически перезапустит воркеры с обновленным кодом.

---

7. Чек-лист надежности перед релизом

• [ ] Все асинхронные DTO явно зарегистрированы в секции routing.
• [ ] Для всех рабочих транспортов настроен failure_transport.
• [ ] Заданы параметры retry_strategy (включая multiplier и jitter).
• [ ] Команда запуска воркера содержит флаги --memory-limit и --time-limit.
• [ ] Воркеры запущены под управлением Supervisor / systemd с директивой autorestart=true.
• [ ] В пайплайн деплоя включена команда php bin/console messenger:stop-workers.
• [ ] Настроен мониторинг и оповещения при появлении записей в транспорте failed.

---

Часто задаваемые вопросы (FAQ)

1. Что происходит с сообщением, если лимит retry исчерпан, а failure_transport не настроен?

Если транспорт сбоев не определен ни на уровне конкретного транспорта, ни глобально через framework.messenger.failure_transport, сообщение после исчерпания попыток удаляется из очереди без возможности восстановления.

2. Зачем нужен параметр jitter в настройках retry_strategy?

Джиттер вносит случайное отклонение в интервал задержки перед повторной попыткой. Если упало множество параллельных задач, джиттер распределяет их повторные запросы во времени и защищает восстанавливающиеся сервисы от перегрузки.

3. Как воркер узнает о необходимости остановиться при вызове messenger:stop-workers?

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

4. Можно ли отправить сообщение в failure_transport сразу, пропустив все повторы?

Да. Для этого в обработчике следует выбросить исключение Symfony\Component\Messenger\Exception\UnrecoverableMessageHandlingException. Воркер пропустит оставшиеся попытки retry и сразу переместит задачу в транспорт сбоев.

5. Как защититься от утечек памяти при длительной работе messenger:consume?

Используйте флаг --memory-limit (например, --memory-limit=256M) вместе с --time-limit (например, --time-limit=3600). При достижении лимитов воркер завершится штатно, а внешний менеджер процессов (Supervisor) перезапустит чистый экземпляр.

---

Заключение

Symfony Messenger предоставляет готовую и гибкую платформу для построения асинхронных очередей. Грамотная настройка retry_strategy в сочетании с failure_transport защищает данные от безвозвратной потери при авариях, а интеграция команды messenger:stop-workers с менеджерами процессов обеспечивает непрерывный и безопасный деплой без простоя системы.

Источники

• Messenger: Sync & Queued Message Handling