Какие существуют распространенные шаблоны сообщений RabbitMQ и когда их использовать?

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

RabbitMQ предоставляет вам обменники, очереди, привязки, подтверждения и свойства сообщений. Полезная проектная работа заключается в выборе того, как эти части подходят к вашему приложению. Распространенными шаблонами являются рабочие очереди, публикация/подписка, маршрутизация Direct или Topic и запрос/ответ.

---

Рабочие очереди: Распределение задач между работниками

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

Как это работает

1. Производитель отправляет задачи (сообщения) в одну очередь.
2. Несколько потребителей (работников) прослушивают одну и ту же очередь.
3. RabbitMQ доставляет каждое сообщение одному потребителю, поэтому работники делят между собой невыполненные задачи.

По умолчанию RabbitMQ распределяет сообщения по кругу между активными потребителями. Это распределение не всегда справедливо, если один работник выполняет медленные задачи, а другой — быстрые, поэтому вы обычно сочетаете этот шаблон с подтверждениями и лимитом предварительной выборки.

Используйте подтверждения

С помощью ручных подтверждений работник сообщает RabbitMQ, когда задача завершена. Если работник умирает до отправки basic_ack, RabbitMQ может перепоставить и повторно доставить это сообщение другому потребителю. Это делает рабочую очередь полезной для генерации отчетов, обработки изображений, задач биллинга или любой другой задачи, которую вы не хотите молча терять.

Установите количество предварительной выборки

basic.qos контролирует, сколько неподтвержденных сообщений потребитель может одновременно удерживать. prefetch_count равный 1 является безопасной отправной точкой для медленных, неравномерных задач, поскольку RabbitMQ не будет отправлять вторую задачу этому потребителю, пока он не подтвердит первую. Для более быстрых задач вы можете увеличить значение после измерения пропускной способности и использования памяти.

Пример реализации (концептуальный)

# Настройка потребителя для справедливого распределения
channel.basic_qos(prefetch_count=1)
channel.basic_consume(queue='task_queue', on_message_callback=worker_function)

# Логика работника должна отправить подтверждение после успешной обработки
worker_function(ch, method, properties, body):
    # Обработка задачи...
    ch.basic_ack(delivery_tag=method.delivery_tag)

Публикация/Подписка: Трансляция событий

Шаблон Pub/Sub предназначен для одновременной трансляции сообщений нескольким заинтересованным потребителям. В отличие от рабочих очередей, где каждое сообщение потребляется только одним работником, Pub/Sub гарантирует, что каждый подключенный подписчик получит копию сообщения.

Используйте обменник fanout

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

Варианты использования

• Уведомления в реальном времени: Трансляция события режима обслуживания на все экземпляры приложения.
• Распределение логирования: Отправка одного и того же события журнала в сервис архивации и сервис оповещения.
• Инвалидация кэша: Указание всем экземплярам сервиса очистить локальный кэш после изменения в базе данных.

Совет по реализации

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

Прямая и тематическая маршрутизация: Выборочная отправка событий

В то время как обменник Fanout обеспечивает слепую трансляцию, AMQP предлагает обменники для выборочной публикации, расширяя модель Pub/Sub.

Прямой обменник (Direct)

Сообщения маршрутизируются в очереди на основе точного совпадения между ключом маршрутизации сообщения и ключом привязки очереди. Это полезно, когда вам нужно целенаправленно обращаться к разным типам потребителей.

Например, ваш издатель логов может отправлять сообщения с ключами маршрутизации error, warning и info. Очередь оповещений может привязываться только к error, в то время как очередь архивации может привязываться ко всем трем уровням серьезности.

Тематический обменник (Topic)

Это самый гибкий тип обменника, позволяющий использовать подстановочные знаки в ключах привязки и маршрутизации. Ключ маршрутизации рассматривается как список, разделенный разделителями (например, точками .).

• * соответствует ровно одному слову.
• # соответствует нулю или более слов.

Ключ маршрутизации, такой как orders.us.created, может попасть в очередь мошенничества, привязанную к orders.*.created, и в очередь операций США, привязанную к #.us.#. Используйте тематические обменники, когда ваши правила маршрутизации являются реальными бизнес-категориями, а не просто одним фиксированным полем.

Запрос/Ответ: Запрос конкретного ответа

Шаблон запроса/ответа позволяет клиентскому приложению отправить запрос и синхронно ожидать ответа от работника (сервера). Хотя обмен сообщениями по своей сути асинхронен, этот шаблон имитирует традиционные удаленные вызовы процедур (RPC) через шину сообщений.

Используйте reply_to и correlation_id

1. Клиент отправляет запрос в очередь, например rpc_queue.
2. Клиент устанавливает reply_to в очередь обратного вызова, которую он потребляет.
3. Клиент устанавливает уникальный correlation_id.
4. Работник обрабатывает запрос и публикует ответ в очередь reply_to.
5. Клиент сопоставляет ответ с исходным запросом, проверяя correlation_id.

Варианты использования

• Поиск сервисов: Получение профиля пользователя или функционального флага из другого сервиса.
• Короткие решения: Проверка наличия на складе перед принятием заказа.

Используйте осторожно

Запрос/Ответ полезен, но он возвращает синхронное ожидание в вашу систему обмена сообщениями. Установите тайм-ауты клиента, обрабатывайте дублирующиеся ответы и избегайте использования RPC для длительных задач. Для медленной работы опубликуйте команду, верните идентификатор задачи и отправляйте события прогресса или завершения отдельно.

Концептуальный поток RPC

graph TD
    A[Клиент (Запросчик)] -->|1. Сообщение-запрос (включая reply_to, correlation_id)| B(Очередь запросов RPC);
    B --> C[Сервер (Работник)];
    C -->|2. Обработка запроса|
D[Результат];
    D -->|3. Сообщение-ответ (через reply_to, сохраняя correlation_id)| A;

---

Распространенные шаблоны RabbitMQ с первого взгляда

Шаблон	Тип обменника	Механизм маршрутизации	Ключевая особенность	Основной вариант использования
Рабочие очереди	По умолчанию или Direct	Одна очередь, разделяемая работниками	Одно сообщение, один потребитель	Балансировка нагрузки длительных задач
Публикация/Подписка	Fanout	Игнорирует ключ маршрутизации	Одно сообщение, все привязанные очереди	Системные трансляции, логирование
Прямая маршрутизация	Direct	Точное совпадение ключа маршрутизации	Целевое обращение к потребителям	Маршрутизация на основе серьезности или типа
Тематическая маршрутизация	Topic	Совпадение с подстановочными знаками (*, #)	Гибкая, сложная маршрутизация	Взаимодействие микросервисов, потоки событий
Запрос/Ответ (RPC)	Direct (для ответа)	Использует reply_to и correlation_id	Имитирует синхронные вызовы API	Немедленный поиск сервисов, небольшие транзакции

Вывод

Начните с формы общения. Используйте рабочие очереди, когда ровно один работник должен выполнить задачу, Pub/Sub, когда каждый подписчик должен увидеть событие, прямую или тематическую маршрутизацию, когда только некоторые подписчики должны его увидеть, и запрос/ответ только тогда, когда вызывающему действительно нужен немедленный ответ.