Модуль 13 ⏱ ~33 мин чтения 📐 4 Mermaid диаграмм

2. Exchanges. Direct, Fanout, Topic, Headers

В предыдущей статье 1. RabbitMQ. Архитектура и концепции мы выяснили фундаментальный закон архитектуры AMQP: в RabbitMQ отправитель никогда не помещает сообщение напрямую в очередь. Поток данных всегда направляется в Exchange (Обменник).

Если провести аналогию с классической почтовой службой, то Exchange — это автоматизированный распределительный логистический хаб. Отправитель наклеивает на конверт почтовый индекс или ярлык категории (Routing Key), а хаб, сверяясь со своей маршрутной картой (Bindings), мгновенно решает: отправить посылку в один конкретный почтомат, размножить ее копии по десяткам филиалов или сразу отправить в шредер, если адресат не существует.

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

Механика маршрутизации

Процесс маршрутизации опирается на взаимодействие трех сущностей:

  1. Exchange Name: Имя обменника, принимающего входящие фреймы публикации.
  2. Routing Key (Ключ маршрутизации): Строка-метка, прикрепляемая отправителем к сообщению при вызове ch.Publish(...). Она служит логическим адресом или тегом полезной нагрузки (например, orders.eu.created или image.resize).
  3. Binding Key (Ключ привязки): Шаблон или критерий отбора, задаваемый при связывании конкретной очереди с данным обменником.

Когда брокер принимает фрейм Basic.Publish, обменник сопоставляет Routing Key с Binding Key всех очередей, подключенных к нему в данный момент, и направляет ссылки на сообщение во все совпавшие очереди.

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

Алгоритм работы: Обменник направляет сообщение в очередь только в том случае, если ее Binding Key байт-в-байт совпадает с Routing Key сообщения.

graph LR
    subgraph Publisher ["Отправитель"]
        P["Producer"]
    end

    subgraph DirectEx ["Direct Exchange: 'tasks.direct'"]
        direction TB
        RouteLookup["O(1) Hash Map Lookup<br>(Mnesia / ETS)"]
    end

    subgraph Queues ["Очереди"]
        Q1["Queue PDF Workers<br>(Binding: 'pdf')"]
        Q2["Queue Image Workers<br>(Binding: 'image')"]
        Q3["Queue Audit PDF<br>(Binding: 'pdf')"]
    end

    P -->|"Routing Key = 'pdf'"| RouteLookup
    RouteLookup -->|"Точное совпадение"| Q1
    RouteLookup -.->|"Не совпадает"| Q2
    RouteLookup -->|"Точное совпадение"| Q3

    classDef pub fill:#1e293b,stroke:#38bdf8,stroke-width:2px,color:#f8fafc;
    classDef ex fill:#312e81,stroke:#818cf8,stroke-width:2px,color:#f8fafc;
    classDef q fill:#064e3b,stroke:#34d399,stroke-width:2px,color:#f8fafc;
    classDef skip fill:#1e293b,stroke:#64748b,stroke-dasharray: 5 5,color:#94a3b8;

    class P pub;
    class RouteLookup ex;
    class Q1,Q3 q;
    class Q2 skip;

На схеме сообщение с Routing Key = pdf попадет и в Queue PDF Workers, и в Queue Audit PDF, так как обе очереди имеют точное совпадение ключа привязки, а Queue Image Workers будет проигнорирована.

Особенности производительности (Mechanical Sympathy)

Direct Exchange — один из самых быстрых инструментов маршрутизации в RabbitMQ. Поиск целевых очередей под капотом реализован через хэш-таблицу $O(1)$ во внутренней оперативной памяти Erlang (таблицы ETS — Erlang Term Storage). Вычисление адресата сводится к мгновенному поиску строки по хэшу без разбора префиксов и регулярных выражений.

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

Алгоритм работы: Полностью игнорирует Routing Key. Каждое поступившее сообщение безусловно дублируется (через ссылки в памяти) во все очереди, привязанные к данному обменнику.

graph LR
    subgraph Publisher ["Издатель события"]
        P["User Service<br>(Event: ProfileUpdated)"]
    end

    subgraph FanoutEx ["Fanout Exchange: 'user.events'"]
        Distributor["Широковещательный дистрибьютор<br>(Игнорирует Routing Key)"]
    end

    subgraph Subscribers ["Подписчики (Pub/Sub)"]
        Q1["Queue: cache_invalidator<br>(Сброс Redis)"]
        Q2["Queue: ws_push<br>(Уведомление браузера)"]
        Q3["Queue: analytics_dwh<br>(Запись в ClickHouse)"]
    end

    P -->|"Routing Key = '' (любой)"| Distributor
    Distributor ==> Q1
    Distributor ==> Q2
    Distributor ==> Q3

    classDef pub fill:#1e293b,stroke:#38bdf8,stroke-width:2px,color:#f8fafc;
    classDef ex fill:#312e81,stroke:#818cf8,stroke-width:2px,color:#f8fafc;
    classDef q fill:#064e3b,stroke:#34d399,stroke-width:2px,color:#f8fafc;

    class P pub;
    class Distributor ex;
    class Q1,Q2,Q3 q;

Когда применять

Идеальный выбор для архитектурного паттерна Publish/Subscribe (Издатель / Подписчики). Классический пример: сервис профилей обновляет данные пользователя. Он публикует одно событие в Fanout Exchange, а далее оно параллельно разлетается по независимым потребителям:

  1. Сервис кэширования инвалидирует ключи в Redis.
  2. Сервис веб-сокетов отправляет пуш на мобильное устройство клиента.
  3. Сервис аналитики перекладывает событие в ClickHouse.

Производительность

Это абсолютный лидер по скорости среди всех обменников RabbitMQ. Процессу Erlang вообще не требуется выполнять строковых сравнений или вычислять хэш-функции. Сложность операции — $O(N)$, где $N$ — число очередей, подключенных к обменнику: брокер просто проходит по связанному списку подписчиков и инкрементирует счетчики ссылок на бинарный буфер.

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

Алгоритм работы: Маршрутизирует сообщения на основе сопоставления Routing Key со специальными строковыми шаблонами очередей (Binding Key), поддерживающими групповые подстановочные знаки (wildcards).

Ключи маршрутизации обязаны состоять из отдельных слов, разделенных точками (длина ключа ограничена спецификацией AMQP до 255 байт). Например: user.created.eu или telemetry.eu.sensor_12.temperature.

В шаблонах привязки (Binding Key) допускается использование двух специальных метасимволов:

  • * (звездочка) — заменяет ровно одно слово между точками.
  • # (решетка) — заменяет ноль, одно или произвольное количество слов.
flowchart TD
    subgraph TopicEx ["Topic Exchange: 'system.logs'"]
        TrieLookup["Префиксное дерево (Trie)<br>Маршрутизация по шаблонам"]
    end

    P1["Producer: log.error.db"] --> TopicEx
    P2["Producer: log.info.web"] --> TopicEx
    P3["Producer: log.warning.cache.redis"] --> TopicEx

    TopicEx -->|"Binding: 'log.error.*'"| Q_DB["Queue: Critical DB Errors<br>(Поймает только log.error.db)"]
    TopicEx -->|"Binding: 'log.*.web'"| Q_Web["Queue: Web Logs<br>(Поймает только log.info.web)"]
    TopicEx -->|"Binding: 'log.#'"| Q_All["Queue: Central Elastic Log<br>(Поймает абсолютно все логи)"]

    classDef pub fill:#1e293b,stroke:#38bdf8,stroke-width:2px,color:#f8fafc;
    classDef ex fill:#312e81,stroke:#818cf8,stroke-width:2px,color:#f8fafc;
    classDef q fill:#064e3b,stroke:#34d399,stroke-width:2px,color:#f8fafc;

    class P1,P2,P3 pub;
    class TrieLookup ex;
    class Q_DB,Q_Web,Q_All q;

Разбор примеров шаблонов:

  • Шаблон log.error.*:
    • Совпадет с log.error.db и log.error.auth.
    • НЕ совпадет с log.error (не хватает одного слова в конце).
    • НЕ совпадет с log.error.db.primary (слишком много слов).
  • Шаблон log.#:
    • Совпадет с log.error.db, log.info, log.warning.cache.redis и даже с одиночным словом log.

4. Headers Exchange (Обменник по заголовкам)

Алгоритм работы: Полностью игнорирует значение Routing Key. Вместо этого логика маршрутизации анализирует словарь метаданных (Headers в свойствах фрейма Header Frame), прикрепленный к сообщению.

При связывании очереди с Headers Exchange указывается набор пар «ключ-значение», а также специальный системный заголовок x-match:

  • x-match: all (по умолчанию): Сообщение попадает в очередь только тогда, когда в его заголовках присутствуют все указанные в привязке ключи с совпадающими значениями (логическое AND).
  • x-match: any: Сообщению достаточно иметь совпадение хотя бы по одному заголовку из таблицы привязки (логическое OR).
graph TD
    subgraph Publisher ["Публикация с Headers"]
        P["Producer<br>Headers: {format: 'pdf', type: 'report'}"]
    end

    subgraph HeadersEx ["Headers Exchange: 'docs.headers'"]
        HeaderMatcher["Анализатор таблиц аргументов<br>(Проверка x-match)"]
    end

    subgraph Queues ["Очереди"]
        Q1["Queue: PDF Reports<br>Binding: {format: 'pdf', type: 'report', x-match: 'all'}"]
        Q2["Queue: Any PDF<br>Binding: {format: 'pdf', x-match: 'any'}"]
        Q3["Queue: Invoices<br>Binding: {type: 'invoice', x-match: 'all'}"]
    end

    P --> HeaderMatcher
    HeaderMatcher -->|"Совпали все (all)"| Q1
    HeaderMatcher -->|"Совпал format (any)"| Q2
    HeaderMatcher -.->|"Не совпал type"| Q3

    classDef pub fill:#1e293b,stroke:#38bdf8,stroke-width:2px,color:#f8fafc;
    classDef ex fill:#312e81,stroke:#818cf8,stroke-width:2px,color:#f8fafc;
    classDef q fill:#064e3b,stroke:#34d399,stroke-width:2px,color:#f8fafc;
    classDef skip fill:#1e293b,stroke:#64748b,stroke-dasharray: 5 5,color:#94a3b8;

    class P pub;
    class HeaderMatcher ex;
    class Q1,Q2 q;
    class Q3 skip;

Особенности производительности

Headers Exchange — наименее производительный тип обменника в экосистеме RabbitMQ.
Парсинг динамических бинарных структур AMQP table (словари произвольных типов, строк, чисел и массивов) требует постоянных аллокаций памяти и глубокого сканирования структур данных в Erlang.

Реализация на Go (Idiomatic Go)

Ниже представлен канонический пример настройки топологии RabbitMQ на Go с использованием драйвера amqp091-go.

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

Go 
package main

import (
    "context"
    "fmt"
    "log"
    "time"

    amqp "github.com/rabbitmq/amqp091-go"
)

// setupInfrastructure объявляет топологию: Topic Exchange, очередь и привязку по маске
func setupInfrastructure(ch *amqp.Channel) error {
    const (
        exchangeName = "events.topic"
        queueName    = "audit_logs"
        bindingKey   = "user.#" // Отлавливаем любые события жизненного цикла пользователя
    )

    // 1. Объявляем Topic Exchange
    // Durable = true гарантирует, что метаданные обменника сохранятся при перезапуске брокера
    err := ch.ExchangeDeclare(
        exchangeName, // name
        "topic",      // kind (direct, fanout, topic, headers)
        true,         // durable
        false,        // auto-delete (удалять, когда отключены все очереди)
        false,        // internal (true запрещает прямую публикацию клиентам)
        false,        // no-wait
        nil,          // args
    )
    if err != nil {
        return fmt.Errorf("ошибка ExchangeDeclare: %w", err)
    }

    // 2. Объявляем устойчивую (Durable) очередь
    q, err := ch.QueueDeclare(
        queueName,
        true,  // durable
        false, // auto-delete (удалять, если отключился последний consumer)
        false, // exclusive (очередь доступна только текущему TCP-соединению)
        false, // no-wait
        nil,   // args (например, x-message-ttl, x-max-length)
    )
    if err != nil {
        return fmt.Errorf("ошибка QueueDeclare: %w", err)
    }

    // 3. Создаем Binding между Exchange и Queue
    err = ch.QueueBind(
        q.Name,       // имя очереди
        bindingKey,   // routing pattern
        exchangeName, // имя обменника
        false,        // no-wait
        nil,          // args
    )
    if err != nil {
        return fmt.Errorf("ошибка QueueBind: %w", err)
    }

    log.Printf("Топология успешно инициализирована: Exchange [%s] -> Binding [%s] -> Queue [%s]",
        exchangeName, bindingKey, q.Name)
    return nil
}

func main() {
    conn, err := amqp.Dial("amqp://guest:guest@localhost:5672/")
    if err != nil {
        log.Fatalf("Подключение не удалось: %v", err)
    }
    defer conn.Close()

    ch, err := conn.Channel()
    if err != nil {
        log.Fatalf("Ошибка открытия канала: %v", err)
    }
    defer ch.Close()

    if err := setupInfrastructure(ch); err != nil {
        log.Fatalf("Сбой настройки топологии: %v", err)
    }

    // Пример публикации в Topic Exchange
    ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
    defer cancel()

    routingKey := "user.profile.updated"
    payload := []byte(`{"user_id": 42, "status": "active"}`)

    err = ch.PublishWithContext(ctx,
        "events.topic", // exchange
        routingKey,     // routing key
        false,          // mandatory
        false,          // immediate
        amqp.Publishing{
            DeliveryMode: amqp.Persistent,
            ContentType:  "application/json",
            Body:         payload,
            Timestamp:    time.Now().UTC(),
        },
    )
    if err != nil {
        log.Fatalf("Ошибка публикации: %v", err)
    }

    log.Printf("Сообщение опубликовано с ключом [%s]", routingKey)
}

Сравнительная таблица обменников

Тип Exchange Алгоритм сопоставления Вычислительная сложность Типичный сценарий использования
Direct Точное равенство (Routing Key == Binding Key) $O(1)$ хэш-таблица Точечная отправка воркерам конкретных задач, RPC
Fanout Игнорирует ключи, транслирует всем $O(N)$ проход по списку Event-Driven архитектуры, Pub/Sub, инвалидация кэшей
Topic Шаблоны с масками * (1 слово) и # (0+ слов) $O(\log M)$ обход префиксного дерева (Trie) Доменные события с иерархией (geo.service.action)
Headers Сопоставление словаря заголовков (x-match: all/any) $O(K \cdot M)$ глубокое сканирование структур Редкие сценарии с десятками ортогональных метаданных

Итоги

  1. Exchange — диспетчер, а не хранилище: Он никогда не хранит сообщения на диске. Неприкаянные сообщения немедленно уничтожаются, если у них нет подходящих привязок (Bindings).
  2. BEAM оптимизирует дублирование: При доставке сообщения в десятки очередей полезная нагрузка не копируется многократно в RAM — очереди получают ссылки на общий буфер в куче Erlang.
  3. Direct быстр и прост благодаря поиску по хэш-таблице ETS. Помните о неявном безымянном Default Exchange.
  4. Fanout — максимальная пропускная способность для широковещательных рассылок.
  5. Topic — стандарт де-факто для микросервисной коммуникации; следите за глубиной вложенности топиков, чтобы не перегружать процессор поиском по Trie.
  6. Headers — используйте с осторожностью из-за высокой вычислительной стоимости разбора динамических заголовков.

Разобравшись с логикой маршрутизации, пора исследовать место, где сообщения физически оседают в ожидании воркеров. В следующей лекции 3. Queues и bindings мы подробно изучим анатомию очередей, флаги долговечности, эксклюзивности, автоудаления и тонкости работы с аргументами x-arguments.