Типы exchange: direct, fanout, topic

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

Тип exchange — правило сравнения routing key сообщения с binding key связок. direct — точное совпадение, fanout — совпадение не проверяется вообще, topic — совпадение по шаблону из слов.

Все три обменника принимают одни и те же сообщения и складывают их в одни и те же очереди. Разница только в вопросе, который обменник задаёт про каждый binding: «этот ключ равен моему?», «а мне всё равно», «этот ключ подходит под мой образец?». Разберём по одному, с рабочим кодом.

direct: точное совпадение

Самый предсказуемый тип. Сообщение попадёт в очередь, только если её binding key побайтово равен routing key сообщения. Идеален, когда категорий немного и они фиксированные: уровни логов, приоритеты задач, страны.

import pika

connection = pika.BlockingConnection(pika.ConnectionParameters("localhost"))
channel = connection.channel()

channel.exchange_declare(exchange="logs.direct", exchange_type="direct", durable=True)

# Очередь дежурного: только серьёзное
channel.queue_declare(queue="alerts", durable=True)
channel.queue_bind(exchange="logs.direct", queue="alerts", routing_key="error")
channel.queue_bind(exchange="logs.direct", queue="alerts", routing_key="critical")

# Очередь архива: вообще всё
channel.queue_declare(queue="log.archive", durable=True)
for level in ("debug", "info", "warning", "error", "critical"):
    channel.queue_bind(exchange="logs.direct", queue="log.archive", routing_key=level)

channel.basic_publish(exchange="logs.direct", routing_key="error", body=b"disk is full")
# -> попадёт в alerts И в log.archive

channel.basic_publish(exchange="logs.direct", routing_key="debug", body=b"cache hit")
# -> попадёт только в log.archive

connection.close()

Обратите внимание на две вещи. Первое: у очереди alerts два binding'а — и это штатный способ сказать «хочу два вида ключей». Второе: сообщение с ключом error подошло двум очередям, значит, каждая получит свою независимую копию. Дежурный отработает алерт, архив запишет строку на диск — они друг другу не мешают.

fanout: копия всем

Fanout не смотрит на routing key вообще. Совсем. Что бы вы туда ни положили, сообщение уйдёт во все привязанные очереди. Это чистая широковещательная рассылка события.

import pika

connection = pika.BlockingConnection(pika.ConnectionParameters("localhost"))
channel = connection.channel()

channel.exchange_declare(exchange="order.events", exchange_type="fanout", durable=True)

# Три независимых подписчика на одно и то же событие
for queue in ("email.orders", "analytics.orders", "warehouse.orders"):
    channel.queue_declare(queue=queue, durable=True)
    channel.queue_bind(exchange="order.events", queue=queue)  # routing_key не нужен

channel.basic_publish(
    exchange="order.events",
    routing_key="",  # будет проигнорирован, можно оставить пустым
    body=b'{"event": "order.paid", "order_id": 4821}',
)

connection.close()

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

Обратная сторона: fanout не умеет фильтровать. Если аналитике нужны все события, а складу — только оплаченные заказы, склад всё равно получит всё и будет фильтровать в коде. Это работает, но означает лишний трафик и лишний CPU у потребителя. Как только появляется слово «только» — вам нужен topic.

topic: маршрутизация по шаблону

Topic — компромисс между гибкостью и простотой, и на практике именно он используется в 80% реальных систем. Routing key делится точками на слова: logs.auth.error — это три слова. В binding key разрешены два спецсимвола:

• * — заменяет ровно одно слово;
• # — заменяет ноль или больше слов.

import pika

connection = pika.BlockingConnection(pika.ConnectionParameters("localhost"))
channel = connection.channel()

channel.exchange_declare(exchange="app.topic", exchange_type="topic", durable=True)

# Все ошибки любого сервиса
channel.queue_declare(queue="errors.all", durable=True)
channel.queue_bind(exchange="app.topic", queue="errors.all", routing_key="logs.*.error")

# Всё, что связано с авторизацией, любого уровня и глубины
channel.queue_declare(queue="auth.everything", durable=True)
channel.queue_bind(exchange="app.topic", queue="auth.everything", routing_key="logs.auth.#")

channel.basic_publish(exchange="app.topic", routing_key="logs.auth.error", body=b"invalid token")
# -> подошло обоим binding'ам: сообщение получат errors.all и auth.everything

channel.basic_publish(exchange="app.topic", routing_key="logs.billing.error", body=b"gateway timeout")
# -> только errors.all

channel.basic_publish(exchange="app.topic", routing_key="logs.auth.db.slow", body=b"query 3s")
# -> только auth.everything (у * не хватило слов, у # — хватило)

connection.close()

Последний пример — самый показательный. Ключ logs.auth.db.slow состоит из четырёх слов. Шаблон logs.*.error требует ровно трёх и последнее — error: не подошёл. Шаблон logs.auth.# требует, чтобы первые два слова были logs и auth, а дальше хоть пусто, хоть десять слов: подошёл.

Таблица сравнения

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

Под капотом три типа — это три разные структуры данных, и стоимость маршрутизации у них разная.

• fanout — просто список очередей. Дешевле некуда: взял список, разослал.
• direct — хеш-таблица «binding key → список очередей». Один поиск по ключу, тоже практически бесплатно.
• topic — дерево по словам (trie). Брокер идёт по словам routing key и собирает подходящие ветки, учитывая * и #. Дороже, но на реальных объёмах разница заметна, только когда binding'ов десятки тысяч.

Отсюда бытовой совет: не оптимизируйте это преждевременно. Разница между direct и topic на сотне binding'ов теряется в шуме сети. Берите тот тип, который честно описывает вашу предметную область.

Ещё несколько важных фактов о механике:

• Дедупликация на уровне очереди. Если очередь подошла сразу под несколько binding'ов одного обменника — она получит одну копию, а не по копии на binding.
• Встроенные обменники. В каждом vhost уже есть amq.direct, amq.fanout, amq.topic, amq.headers и default exchange. Использовать их можно, но для своей системы лучше завести именованные — понятнее в UI и не мешают чужому коду.
• Exchange можно привязать к exchange. Метод exchange_bind позволяет строить каскады: например, fanout-обменник «все события» пересылает копию в topic-обменник, где уже идёт фильтрация. Приём мощный, но злоупотребление им превращает топологию в лабиринт.
• Тип нельзя изменить. Существующий обменник только пересоздать: удалить и объявить заново с новым типом.

Частые ошибки

• Указывают routing key у fanout и ждут фильтрации. Он проигнорирован. Молча. Потом долго удивляются, почему аналитика получает служебные события.
• Пишут шаблон в basic_publish. Публикация с ключом logs.*.error отправит сообщение с буквальным ключом, где звёздочка — обычный символ. Шаблоны живут только в binding key. Это ошибка номер один у новичков в topic.
• Берут direct там, где нужен topic. Симптом: у одной очереди двадцать binding'ов, и при добавлении нового типа события приходится править конфиг всех потребителей. Знак, что пора перейти на иерархические ключи и #.
• Очередь-помойка с binding #. Такая очередь получает вообще всё, включая события, которых на момент её создания ещё не существовало. Растёт, забивает диск, никто её не читает. Если делаете такую очередь для отладки — ставьте ей TTL или max-length.
• Пытаются поменять тип существующего обменника. Получают 406 PRECONDITION_FAILED и закрытый канал. Нужен новый тип — новое имя обменника или удаление старого.
• Забывают про unroutable в topic. Опечатались в шаблоне (log.*.error вместо logs.*.error) — очередь просто ничего не получает, и никто не падает. Первый диагностический шаг при «сообщения не доходят» — вкладка Bindings в management-плагине.

Итоги

• direct — точное совпадение ключа. Просто, быстро, для фиксированных категорий.
• fanout — копия всем привязанным очередям, routing key игнорируется полностью.
• topic — шаблоны из слов: * = ровно одно слово, # = ноль или больше слов.
• Одна очередь может иметь несколько binding'ов — но копию сообщения получит только одну.
• Шаблоны допустимы только в binding key; при публикации ключ всегда конкретный.
• Тип обменника после создания не меняется — только пересоздание.