Direct vs. Topic vs. Fanout: выбор правильного типа обмена

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

Практический выбор прост: используйте direct для точной маршрутизации, fanout для широковещательной рассылки и topic для маршрутизации с подстановочными знаками по схеме именования.

Как работает маршрутизация RabbitMQ

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

Три важных термина:

• Обмен: принимает опубликованные сообщения.
• Очередь: хранит сообщения до тех пор, пока потребители не получат их.
• Привязка: соединяет очередь с обменом, иногда с ключом привязки.

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

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

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

Используйте его, когда варианты маршрутизации известны и точны. Хорошие примеры включают уровень логирования, тип задачи или очереди, специфичные для арендатора, где достаточно точных имен.

Например, привяжите error_queue с error и info_queue с info. Сообщение, опубликованное с ключом маршрутизации error, попадает в error_queue; сообщение с info — в info_queue. Если две очереди привязаны с error, обе очереди получают копию. Потребители в одной очереди по-прежнему конкурируют за сообщения из этой очереди.

import pika

connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()
channel.exchange_declare(exchange='logs_direct', exchange_type='direct')

channel.basic_publish(
    exchange='logs_direct',
    routing_key='error',
    body=b'Database connection failed'
)

connection.close()

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

Широковещательный обмен (Fanout Exchange)

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

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

Например, широковещательный обмен user_updates может питать три очереди: email_queue, search_index_queue и analytics_queue. Одно опубликованное обновление профиля достигает всех трех сервисов.

import pika

connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()
channel.exchange_declare(exchange='user_updates', exchange_type='fanout')

channel.basic_publish(
    exchange='user_updates',
    routing_key='',
    body=b'User 123 changed their display name'
)

connection.close()

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

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

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

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

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

Например, с ключами маршрутизации, такими как order.created.us и order.cancelled.eu:

• order.*.us соответствует одному действию для заказов в США.
• order.created.# соответствует созданным заказам во всех регионах и необязательным дополнительным словам.
• #.eu соответствует событиям, заканчивающимся на eu.

import pika

connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()
channel.exchange_declare(exchange='events_topic', exchange_type='topic')

channel.basic_publish(
    exchange='events_topic',
    routing_key='order.created.us',
    body=b'Order 9001 created'
)

connection.close()

Выбирайте тематический обмен, когда потребителям нужны отфильтрованные подмножества более крупного потока событий, например service.environment.severity, entity.action.region или tenant.event.priority.

Краткое руководство по выбору

Потребность	Тип обмена	Пример
Точная маршрутизация	Direct	error, invoice.created
Широковещательная рассылка всем подписчикам	Fanout	событие очистки кэша
Маршрутизация с подстановочными знаками	Topic	orders.*.us, logs.#

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

Вывод

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