Какие знаешь типы Exchanges существуют в RabbitMQ?

Типы Exchange в RabbitMQ

В RabbitMQ Exchange — это сущность, принимающая сообщения от производителей (producers) и распределяющая их по очередям (queues) согласно определенным правилам, называемым типом Exchange. Тип Exchange определяет логику маршрутизации сообщений и является ключевым механизмом для реализации различных паттернов messaging.

RabbitMQ поддерживает четыре основных типа Exchange, каждый из которых предназначен для конкретных сценариев использования.

1. Direct Exchange (Прямой)

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

• Принцип работы: Сообщение отправляется только в те очереди, чей binding key полностью совпадает с routing key сообщения.
• Сценарий использования: Идеально для задач типа "точка-точка" (point-to-point), когда нужно направлять сообщения конкретному потребителю или группе потребителей с одинаковым ключом. Часто используется для реализации рабочих очередей (worker queues).

// Пример создания Direct Exchange в C# с использованием RabbitMQ.Client
using RabbitMQ.Client;

var factory = new ConnectionFactory() { HostName = "localhost" };
using var connection = factory.CreateConnection();
using var channel = connection.CreateModel();

// Объявление Direct Exchange с именем "direct_logs"
channel.ExchangeDeclare(exchange: "direct_logs", type: ExchangeType.Direct);

// Связывание очереди "queue_severe" с ключом "severe"
channel.QueueBind(queue: "queue_severe",
                  exchange: "direct_logs",
                  routingKey: "sever");

// Публикация сообщения с routingKey "severe" будет доставлено только в queue_severe
var body = Encoding.UTF8.GetBytes("Critical Error!");
channel.BasicPublish(exchange: "direct_logs",
                     routingKey: "sever",
                     basicProperties: null,
                     body: body);

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

Этот тип игнорирует routing key. Он рассылает каждое полученное сообщение во все очереди, связанные с этим Exchange.

• Принцип работы: "Размножает" сообщение на все связанные очереди. Все потребители получают одинаковые сообщения.
• Сценарий использования: Паттерн публикация/подписка (pub/sub) в чистом виде. Например, рассылка уведомлений всем подписчикам системы, обновлений состояния всем подключенным клиентам.

// Пример создания Fanout Exchange
channel.ExchangeDeclare(exchange: "fanout_news", type: ExchangeType.Fanout);

// Все три очереди получат сообщение, независимо от routingKey (он даже может быть пустым)
channel.QueueBind(queue: "queue_mobile", exchange: "fanout_news", routingKey: "");
channel.QueueBind(queue: "queue_web", exchange: "fanout_news", routingKey: "");
channel.QueueBind(queue: "queue_email", exchange: "fanout_news", routingKey: "");

channel.BasicPublish(exchange: "fanout_news",
                     routingKey: "", // Игнорируется
                     basicProperties: null,
                     body: Encoding.UTF8.GetBytes("News Flash!"));

3. Topic Exchange (Тематический)

Наиболее мощный и гибкий тип. Он позволяет осуществлять маршрутизацию по шаблону (pattern) binding key. routing key сообщения должен состоять из слов, разделенных точками, а binding key может содержать специальные символы:

• * (звездочка) — заменяет одно слово.

• # (хеш) — заменяет ноль или более слов.

• Принцип работы: Сообщение доставляется в очереди, чей шаблон binding key совпадает с routing key сообщения.

• Сценарий использования: Сложные системы подписки, где потребители интересуются определенными категориями или темами сообщений (например, логи по регионам и уровням: log.us.error, log.eu.info).

// Пример создания Topic Exchange
channel.ExchangeDeclare(exchange: "topic_logs", type: ExchangeType.Topic);

// Очередь 1: получает все логи ошибок из США
channel.QueueBind(queue: "queue_us_errors",
                  exchange: "topic_logs",
                  routingKey: "log.us.error");

// Очередь 2: получает все логи из Европы (любой уровень)
channel.QueueBind(queue: "queue_eu_all",
                  exchange: "topic_logs",
                  routingKey: "log.eu.*");

// Очередь 3: получает все логи (любой регион, любой уровень) - широкий подписчик
channel.QueueBind(queue: "queue_global",
                  exchange: "topic_logs",
                  routingKey: "log.#");

// Сообщение с routingKey "log.us.error" попадет в queue_us_errors и queue_global.
// Сообщение "log.eu.info" попадет в queue_eu_all и queue_global.

4. Headers Exchange (Заголовочный)

Маршрутизация в этом типе основана не на ключе, а на заголовках сообщения (headers), которые представляют собой пары ключ-значение. При связывании очереди указывается набор критериев (x-match) для заголовков.

• Принцип работы: Сравнивает заголовки сообщения с аргументами (x-match), указанными при связывании. Аргумент x-match может быть all (сообщение должно соответствовать всем критериям) или any (соответствовать любому из критериев).
• Сценарий использования: Когда логика маршрутизации зависит от сложных атрибутов сообщения, которые сложно выразить в строковом ключе (например, версия протокола, тип устройства, флаги).

// Пример создания Headers Exchange (в RabbitMQ.Client тип указывается как "headers")
channel.ExchangeDeclare(exchange: "headers_data", type: ExchangeType.Headers);

// Связывание очереди, требующей наличия заголовков "format=json" и "priority=high"
var headersArgs = new Dictionary<string, object>
{
    { "x-match", "all" }, // Соответствие всем условиям
    { "format", "json" },
    { "priority", "high" }
};
channel.QueueBind(queue: "queue_json_high",
                  exchange: "headers_data",
                  routingKey: "", // routingKey игнорируется
                  arguments: headersArgs);

// Публикация сообщения с соответствующими заголовками
var properties = channel.CreateBasicProperties();
properties.Headers = new Dictionary<string, object>
{
    { "format", "json" },
    { "priority", "high" }
};
channel.BasicPublish(exchange: "headers_data",
                     routingKey: "",
                     basicProperties: properties,
                     body: Encoding.UTF8.GetBytes("{...}"));

Default Exchange

Стоит также упомянуть Default Exchange — это предопределенный Direct Exchange с пустым именем (""). Каждая очередь автоматически связывается с ним с binding key, равным имени самой очереди. Это позволяет публиковать сообщения напрямую в очередь, указывая ее имя как routingKey.

// Публикация напрямую в очередь "my_queue" через Default Exchange
channel.BasicPublish(exchange: "", // Пустое имя - Default Exchange
                     routingKey: "my_queue", // Имя очереди
                     basicProperties: null,
                     body: body);

Выбор типа Exchange зависит от требований архитектуры:

• Direct — для простого назначения задач.
• Fanout — для широковещательных рассылок.
• Topic — для гибкой, основанной на темах, маршрутизации (наиболее популярен для сложных систем).
• Headers — для маршрутизации по сложным атрибутам, когда routing key недостаточно.

Понимание и правильное применение этих типов позволяет строить эффективные, масштабируемые и легко адаптируемые системы на основе RabbitMQ.