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

6. Routing patterns

В статье 2. Exchanges. Direct, Fanout, Topic, Headers мы подробно исследовали базовые механизмы маршрутизации AMQP 0-9-1 и увидели, как обменники направляют трафик на основе Routing Key. В лекции 5. Prefetch и QoS мы настроили безопасное потребление сообщений, защитив сервисы на Go от исчерпания оперативной памяти и аварийного завершения по OOM.

Теперь настало время объединить эти фундаментальные блоки в целостную систему. В реальной промышленной разработке изолированные очереди и простые обменники используются редко. Архитекторы объединяют их в специализированные Топологии (Topologies) или Паттерны маршрутизации (Routing Patterns), каждый из которых решает конкретную инженерную задачу распределенного взаимодействия.

1. Work Queues (Competing Consumers / Конкурирующие воркеры)

Это базовый и наиболее распространенный шаблон асинхронной обработки задач.

Архитектурная задача: У вас есть единый интенсивный поток ресурсоемких фоновых задач (например, конвертация видео, генерация PDF-отчетов или отправка email-рассылок), и вы хотите горизонтально масштабировать их выполнение между $N$ независимыми экземплярами сервиса на Go.

graph LR
    subgraph Publisher ["Продюсер"]
        P["API Gateway"]
    end

    subgraph Broker ["RabbitMQ"]
        Ex["Direct Exchange<br>(или Default '')"]
        Q["Queue: 'tasks.email'<br>(Один Erlang-процесс)"]
        Ex -->|"Routing Key: 'email'"| Q
    end

    subgraph Workers ["Пул воркеров (Competing Consumers)"]
        C1["Go Worker 1<br>(Prefetch: 5)"]
        C2["Go Worker 2<br>(Prefetch: 5)"]
        C3["Go Worker 3<br>(Prefetch: 5)"]
    end

    P -->|"Basic.Publish"| Ex
    Q -->|"Push (Round-Robin)"| C1
    Q -->|"Push (Round-Robin)"| C2
    Q -->|"Push (Round-Robin)"| C3

    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 sub fill:#1e293b,stroke:#f59e0b,stroke-width:2px,color:#f8fafc;

    class P pub;
    class Ex ex;
    class Q q;
    class C1,C2,C3 sub;

Механика работы и Mechanical Sympathy

Брокер распределяет сообщения между всеми активными подписчиками по алгоритму Round-Robin (циклический опрос). Однако этот опрос интеллектуален: он жестко координируется параметром prefetch_count. Если у Worker 1 локальный буфер неподтвержденных сообщений заполнен, планировщик RabbitMQ пропускает его и передает очередную задачу свободному Worker 2.

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

Архитектурная задача: Одно доменное событие должно быть гарантированно доставлено сразу нескольким несвязанным между собой подсистемам (Broadcast).

Классический пример: Пользователь загрузил новое видео в сервис.

  • Подсистема №1 должна извлечь аудиодорожку для транскрибации.
  • Подсистема №2 должна нарезать серию превью-кадров (Thumbnails).
  • Подсистема №3 должна разослать push-уведомления подписчикам канала.
graph TD
    subgraph Publisher ["Издатель"]
        P["Video Upload Service"]
    end

    subgraph FanoutEx ["Fanout Exchange: 'video.uploaded'"]
        Distributor["Широковещательный узел"]
    end

    subgraph Queues ["Изолированные очереди подсистем"]
        Q1["Queue: video.audio_extractor"]
        Q2["Queue: video.thumbnails"]
        Q3["Queue: video.push_notifications"]
    end

    subgraph Consumers ["Воркеры подсистем"]
        C1["Audio Transcriber"]
        C2["FFmpeg Resizer"]
        C3["Push Service"]
    end

    P -->|"Basic.Publish"| Distributor
    Distributor ==> Q1
    Distributor ==> Q2
    Distributor ==> Q3

    Q1 --> C1
    Q2 --> C2
    Q3 --> C3

    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 sub fill:#1e293b,stroke:#f59e0b,stroke-width:2px,color:#f8fafc;

    class P pub;
    class Distributor ex;
    class Q1,Q2,Q3 q;
    class C1,C2,C3 sub;

3. Content-Based Routing (Иерархическое дерево топиков)

Архитектурная задача: Создание гибкой шины доменных событий (Event-Driven Architecture), где микросервисы динамически подписываются только на строго интересующие их срезы данных без изменения кода продюсеров.

Конвенция именования ключей маршрутизации

Промышленным стандартом для Topic Exchange является сегментированная структура ключа через точку:
<domain>.<entity>.<action>.<version>
Например: billing.invoice.paid.v1 или auth.user.login_failed.v2.

flowchart TD
    subgraph TopicHub ["Topic Exchange: 'events.topic'"]
        Trie["Префиксное дерево маршрутизации (Trie)"]
    end

    P1["Billing Service: billing.invoice.paid.v1"] --> TopicHub
    P2["Billing Service: billing.invoice.cancelled.v1"] --> TopicHub
    P3["Auth Service: auth.user.created.v1"] --> TopicHub

    TopicHub -->|"Binding: 'billing.invoice.paid.*'"| Q1["Queue: Financial Analytics<br>(Только оплаченные счета)"]
    TopicHub -->|"Binding: 'billing.invoice.*.v1'"| Q2["Queue: Invoice History<br>(Любые действия с инвойсами)"]
    TopicHub -->|"Binding: '#'"| Q3["Queue: Global Audit 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 Trie ex;
    class Q1,Q2,Q3 q;

Накладные расходы на вычисление маршрута

Таблица правил привязок для Topic Exchange компилируется в памяти брокера в префиксное дерево (Trie).

  • Обход дерева выполняется для каждого входящего фрейма.
  • Чрезмерно длинные ключи (более 5–6 уровней вложенности) и обилие спецсимволов # кратно увеличивают расход тактов CPU.
  • Рекомендация: Держите глубину топиков в пределах 3–4 уровней.

4. Wire Tap (Неинвазивный аудит) и Alternate Exchange

В распределенных системах регулярно возникает потребность в «прослушке» или аудиторском логировании сообщений, которые по ошибке не смогли найти своего адресата (Unroutable Messages).

По умолчанию RabbitMQ молча удаляет сообщения, для которых в Exchange не нашлось ни одного совпадающего биндинга. Для предотвращения потерь данных протокол AMQP 0-9-1 предлагает изящный паттерн — Alternate Exchange (AE).

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

    subgraph Topo ["RabbitMQ Топология"]
        MainEx["Main Exchange: 'orders.topic'<br>(С аргументом alternate-exchange)"]
        NormalQ["Queue: 'orders.process'<br>(Binding: 'orders.created')"]

        AltEx["Alternate Exchange: 'unroutable.fanout'<br>(AE)"]
        AuditQ["Queue: 'unroutable.audit'<br>(Хранилище заблудившихся сообщений)"]

        MainEx -->|"Совпал ключ: 'orders.created'"| NormalQ
        MainEx -.->|"Ключ: 'order.baddata'<br>(Unroutable / Нет совпадений)"| AltEx
        AltEx --> AuditQ
    end

    P --> MainEx

    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 ae fill:#831843,stroke:#f43f5e,stroke-width:2px,color:#fdf2f8;

    class P pub;
    class MainEx ex;
    class NormalQ,AuditQ q;
    class AltEx ae;

Конфигурация Alternate Exchange на Go:

Go 
// Декларация основного обменника со ссылкой на Alternate Exchange
args := amqp.Table{
    "alternate-exchange": "audit.fanout", // Куда перенаправлять нераспределяемые сообщения
}
err := ch.ExchangeDeclare(
    "main.topic", // name
    "topic",      // kind
    true,         // durable
    false,        // auto-delete
    false,        // internal
    false,        // no-wait
    args,         // параметры
)

5. Request-Reply (Синхронный RPC через RabbitMQ)

Иногда асинхронный брокер используется для синхронного по своей семантике вызова (Remote Procedure Call). Например: фронтенд-шлюз на Go отправляет запрос на тяжелый математический расчет в Python-воркер и блокируется до получения результата.

sequenceDiagram
    autonumber
    participant Client as "Go RPC Client"
    participant Broker as "RabbitMQ"
    participant Server as "Worker Service (RPC Server)"

    Note over Client: Инициализация: Создание ОДНОЙ очереди ответов (replyQ)
    Client->>Broker: Basic.Publish (RPC Task)<br>ReplyTo: "client-replies-node1"<br>CorrelationId: "uuid-9901"
    Broker->>Server: Basic.Deliver (Task Payload)

    Server->>Server: Расчет результата (CPU-bound)

    Server->>Broker: Basic.Publish (Результат)<br>Exchange: "", RoutingKey: "client-replies-node1"<br>CorrelationId: "uuid-9901"
    Broker->>Client: Basic.Deliver (Ответ)
    Note over Client: Поиск канала в sync.Map по CorrelationId -> Разблокировка горутины

Архитектурная ловушка новичков (The Ephemeral Queue Trap)

Создавать новую эксклюзивную очередь (Exclusive Queue) на каждый входящий HTTP-запрос — фатальная архитектурная ошибка.
Создание очереди в кластере RabbitMQ — это тяжелая распределенная транзакция в метабазе Mnesia с синхронизацией по сети. При нагрузке в 200–300 RPS брокер захлебнется от постоянного создания и удаления очередей в ядре Erlang.

Правильный паттерн: Пул каналов и мультиплексирование через sync.Map

Инстанс клиента создает ровно одну очередь ответов на весь сервис. Внутренние горутины регистрируют свои каналы ожидания через sync.Map и channels по ключу CorrelationId.

Идиоматичный Go-код промышленного RPC-клиента:

Go 
package rpc

import (
    "context"
    "fmt"
    "sync"
    "time"

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

// RPCClient реализует потокобезопасный мультиплексированный клиент RPC
type RPCClient struct {
    ch       *amqp.Channel
    replyQ   string
    requests sync.Map // map[string]chan []byte (CorrelationId -> Go-канал)
}

// NewRPCClient инициализирует долгоживущую очередь для ответов и фоновый слушатель
func NewRPCClient(ch *amqp.Channel) (*RPCClient, error) {
    // 1. Создаем ОДНУ очередь для ВСЕХ ответов данного инстанса приложения
    q, err := ch.QueueDeclare(
        "",    // Имя сгенерирует брокер (например, amq.gen-Xa81...)
        false, // durable: очередь временная
        true,  // auto-delete: удалить при отключении сервиса
        true,  // exclusive: доступна только этому соединению
        false, // no-wait
        nil,
    )
    if err != nil {
        return nil, fmt.Errorf("queue declare error: %w", err)
    }

    client := &RPCClient{
        ch:     ch,
        replyQ: q.Name,
    }

    // 2. Запускаем фоновый consumer для прослушивания ответов
    msgs, err := ch.Consume(
        q.Name,
        "",    // consumer-tag
        true,  // autoAck: для временных RPC ответов допустим autoAck
        false, // exclusive
        false, // noLocal
        false, // noWait
        nil,
    )
    if err != nil {
        return nil, fmt.Errorf("consume error: %w", err)
    }

    go client.listenForReplies(msgs)

    return client, nil
}

// listenForReplies непрерывно читает поток ответов и пробуждает горутины
func (c *RPCClient) listenForReplies(msgs <-chan amqp.Delivery) {
    for d := range msgs {
        if chWaitAny, ok := c.requests.LoadAndDelete(d.CorrelationId); ok {
            chWait := chWaitAny.(chan []byte)
            chWait <- d.Body // Будим горутину, заблокированную в Call()
            close(chWait)
        }
    }
}

// Call отправляет RPC-запрос и синхронно ожидает ответ с учетом контекста
func (c *RPCClient) Call(ctx context.Context, targetQueue string, payload []byte) ([]byte, error) {
    corrID := uuid.NewString()
    waitCh := make(chan []byte, 1) // Буферизованный канал во избежание блокировки слушателя

    // Регистрируем ожидание по CorrelationId
    c.requests.Store(corrID, waitCh)
    defer c.requests.Delete(corrID) // Защита от утечек памяти при таймауте

    // Публикуем задачу
    err := c.ch.PublishWithContext(ctx,
        "",          // exchange (прямая отправка через default exchange)
        targetQueue, // routing key (имя очереди сервера)
        false,
        false,
        amqp.Publishing{
            ContentType:   "application/json",
            CorrelationId: corrID,   // Уникальный идентификатор запроса
            ReplyTo:       c.replyQ, // Куда сервер обязан прислать ответ
            Body:          payload,
            Timestamp:     time.Now().UTC(),
        },
    )
    if err != nil {
        return nil, fmt.Errorf("rpc publish failed: %w", err)
    }

    // Ожидаем ответ либо срабатывание таймаута контекста
    select {
    case <-ctx.Done():
        return nil, ctx.Err()
    case res := <-waitCh:
        return res, nil
    }
}

Итог

  1. Work Queues (Competing Consumers): Оптимальны для параллельного распределения задач между воркерами. Помните: хронологический порядок обработки сообщений при этом не сохраняется.
  2. Publish / Subscribe (Fanout): Идеален для рассылки широковещательных событий нескольким сервисам. Работает с околонулевым оверхедом по памяти благодаря Zero-Copy в BEAM.
  3. Content-Based Routing (Topic): Базовый каркас для событийно-ориентированных микросервисов. Проектируйте структуру топиков не глубже 3–4 уровней.
  4. Wire Tap & Alternate Exchange: Защищают систему от невидимой потери сообщений на этапе маршрутизации при опечатках в ключах или сбоях конфигурации.
  5. AMQP RPC: Требует строгой дисциплины. Никогда не создавайте очереди на каждый HTTP-запрос — используйте одну общую очередь ответов на процесс и мультиплексирование через CorrelationId.

Мы научились собирать гибкие топологии и отлавливать потерянные сообщения на этапе публикации. Но что делать с теми сообщениями, которые успешно дошли до очереди, но были отвергнуты воркером из-за некорректных данных или исчерпания лимита ретраев?

В следующей статье 7. Dead letter exchanges мы глубоко исследуем спасительный механизм Dead Letter Exchange (DLX) и разберем архитектуру очередей брака.