1. RabbitMQ. Архитектура и концепции

Введение в экосистему очередей мы сделали в статьях 1. Обзор раздела. Асинхронность как основа масштабирования и 3. Очереди сообщений. Зачем они нужны. Теперь мы переходим к практическому и архитектурному разбору одного из самых проверенных временем брокеров сообщений в индустрии — RabbitMQ.

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

Smart Broker, Dumb Consumer

В мире асинхронных систем существует два полярных архитектурных лагеря:

1. Smart Broker, Dumb Consumer (Умный брокер — простой потребитель): Типичный представитель — RabbitMQ. Брокер берет на себя всю тяжелую интеллектуальную работу: динамическую фильтрацию, маршрутизацию по сложным шаблонам, отслеживание подтверждений (Acks), расчет очередей повторных попыток (Retries), Dead Lettering и перемещение сообщений между буферами. Потребитель (Consumer) остается предельно легковесным: он просто подключается к своей персональной очереди и обрабатывает входящий поток.
2. Dumb Broker, Smart Consumer (Простой брокер — умный потребитель): Типичный представитель — Apache Kafka. Здесь брокер — это фактически немодифицируемый журнал записей на диске (Append-Only Commit Log). Брокеру безразлично, кто прочитал сообщение, а кто отстал: хранением и смещением указателей (Offsets), партиционированием и логикой повторов целиком управляет клиент.

RabbitMQ исторически строился вокруг спецификации промышленного стандарта AMQP 0-9-1 (Advanced Message Queuing Protocol). Глубокое понимание RabbitMQ невозможно без осознания того, почему для его реализации была выбрана платформа Erlang/OTP.

Базовые концепции AMQP

В модели AMQP 0-9-1 сообщения никогда не публикуются отправителем напрямую в очередь. Между отправителем и хранилищем всегда находится интеллектуальный посредник.

graph LR
    subgraph Producers ["Отправители (Go Apps)"]
        P1["Producer 1"]
        P2["Producer 2"]
    end

    subgraph AMQP_Broker ["RabbitMQ Broker (vhost: /)"]
        Ex["Exchange<br>(Обменник / Почтовый сортировщик)"]

        subgraph Queues ["Очереди (Erlang-процессы)"]
            Q1["Queue: orders.process<br>(RAM / Disk)"]
            Q2["Queue: audit.events<br>(RAM / Disk)"]
            Q3["Queue: notifications.email<br>(RAM / Disk)"]
        end

        Ex -->|"Binding: order.created"| Q1
        Ex -->|"Binding: order.*"| Q2
        Ex -->|"Binding: *.created"| Q3
    end

    subgraph Consumers ["Потребители (Воркеры)"]
        C1["Worker 1"]
        C2["Audit Service"]
        C3["Email Notifier"]
    end

    P1 -->|"Basic.Publish<br>Routing Key: order.created"| Ex
    P2 -->|"Basic.Publish<br>Routing Key: order.updated"| Ex

    Q1 -->|"Push (Basic.Deliver)"| C1
    Q2 -->|"Push (Basic.Deliver)"| C2
    Q3 -->|"Push (Basic.Deliver)"| 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 P1,P2 pub;
    class Ex ex;
    class Q1,Q2,Q3 q;
    class C1,C2,C3 sub;
  

1. Virtual Host (vhost)

Виртуальный хост (vhost) — это фундаментальная единица мультиарендности (Multi-Tenancy) внутри одного кластера RabbitMQ. Подобно тому как базы данных в PostgreSQL или пространства имен (Namespaces) в Kubernetes изолируют сущности разных сервисов или сред (staging vs production), vhost изолирует обменники, очереди, пользователей и права доступа:

• Имена очередей уникальны только в рамках одного vhost (можно иметь очередь orders в vhost /prod и такую же очередь orders в /stage).
• По умолчанию при установке создается виртуальный хост с именем /.

2. Exchange (Обменник)

Это маршрутизатор сообщений («сортировочный терминал»). Отправитель (Producer) публикует сообщение исключительно в Exchange, снабжая его метаданными: Routing Key (ключ маршрутизации) и заголовками.

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

3. Queue (Очередь)

Физический кольцевой буфер в оперативной памяти и на диске, где сообщения упорядоченно ожидают своей доставки потребителям либо истечения срока жизни (TTL). В отличие от эфемерного Exchange, очередь обладает физическим состоянием, счетчиками метрик и потребляет системные ресурсы.

4. Binding (Привязка)

Связующее звено между Exchange и Queue. Биндинг представляет собой таблицу маршрутизации для брокера: он говорит обменнику, по какому критерию входящие сообщения должны попадать в конкретную очередь. Биндинг может содержать строку шаблона (Binding Key) или набор пар ключ-значение (для Headers Exchange).

Сетевой уровень: Connections и Channels

Для разработчика на Go критически важно понимать физику сетевого взаимодействия с RabbitMQ. Неосторожная работа с сокетами может легко исчерпать файловые дескрипторы ОС или привести к деградации производительности из-за конкуренции за мьютексы.

Спецификация AMQP четко разделяет два уровня: Connection и Channel.

sequenceDiagram
    autonumber
    participant App as "Go Application"
    participant Kernel as "Linux Kernel (TCP/Socket)"
    participant Broker as "RabbitMQ Broker (Erlang)"

    Note over App,Broker: 1. Дорогостоящее открытие TCP Connection
    App->>Kernel: sys_socket() + connect() [3-way handshake SYN/ACK]
    Kernel->>Broker: TCP Established (fd allocated, sk_buff buffers)
    App->>Broker: AMQP Protocol Header (AMQP 0-9-1)
    Broker->>App: Connection.Start / Connection.Tune
    App->>Broker: Connection.Open (vhost: /)
    Broker->>App: Connection.Open-Ok

    Note over App,Broker: 2. Легковесное открытие виртуальных Channels (мультиплексирование)
    App->>Broker: Channel.Open (Channel ID: 1)
    Broker->>App: Channel.Open-Ok
    App->>Broker: Channel.Open (Channel ID: 2)
    Broker->>App: Channel.Open-Ok

    Note over App,Broker: 3. Передача данных во фреймах с ID канала
    App->>Broker: [Frame: Ch 1 | Basic.Publish | RoutingKey: orders]
    App->>Broker: [Frame: Ch 2 | Basic.Deliver | Msg Payload]
  

Connection

Это реальное, физическое сетевое соединение (TCP-сокет) между процессом Go и демоном RabbitMQ.

• Mechanical Sympathy: Установка TCP-соединения — ресурсоемкая операция. Она требует выполнения сетевого трехстороннего рукопожатия (3-way handshake), выделения дескриптора в таблице файлов процесса, аллокации буферов приема и передачи ядра (rmem/wmem, структуры sock и sk_buff), а также инициализации структур в рантайме Go (netpoll на базе epoll/kqueue). Если используется TLS/SSL, к этому добавляется криптографический обмен ключами.
• Поддержание десятков тысяч открытых соединений создает паразитное давление на память операционной системы и бэкенда.

Channel

Чтобы устранить оверхед от постоянного открытия TCP-сокетов под каждую задачу или горутину, AMQP использует протокольное мультиплексирование.

Channel — это легковесный двунаправленный виртуальный канал, существующий внутри единственного физического TCP-соединения. Каждому каналу присваивается уникальный числовой идентификатор (Channel ID).

Все операции (объявление очередей QueueDeclare, публикация Publish, подписка Consume, подтверждение Ack) выполняются исключительно через Channel.

Как это выглядит в Go

Для языка Go официальным стандартом де-факто является библиотека github.com/rabbitmq/amqp091-go (форк классического streadway/amqp, поддерживаемый командой RabbitMQ).

package main

import (
    "context"
    "log"
    "time"

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

func main() {
    // 1. Создаем ОДНО долгоживущее физическое TCP-соединение на весь сервис
    conn, err := amqp.Dial("amqp://guest:guest@localhost:5672/")
    if err != nil {
        log.Fatalf("Ошибка подключения к RabbitMQ: %v", err)
    }
    defer conn.Close()

    // 2. Открываем легковесный Channel для текущего рабочего контекста
    ch, err := conn.Channel()
    if err != nil {
        log.Fatalf("Не удалось открыть AMQP-канал: %v", err)
    }
    defer ch.Close()

    // 3. Объявляем очередь (идемпотентная операция)
    q, err := ch.QueueDeclare(
        "orders.processing", // name
        true,                // durable (сохранять очередь при перезапуске брокера)
        false,               // auto-delete (удалять, когда отключился последний consumer)
        false,               // exclusive (доступна только текущему connection)
        false,               // no-wait
        nil,                 // arguments
    )
    if err != nil {
        log.Fatalf("Ошибка QueueDeclare: %v", err)
    }

    ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
    defer cancel()

    // 4. Публикуем сообщение
    body := []byte(`{"order_id": "ord-7749", "amount": 12500}`)
    err = ch.PublishWithContext(ctx,
        "",     // exchange (дефолтный direct exchange)
        q.Name, // routing key (имя очереди)
        false,  // mandatory
        false,  // immediate
        amqp.Publishing{
            DeliveryMode: amqp.Persistent, // надежная запись на диск (fsync)
            ContentType:  "application/json",
            Body:         body,
            Timestamp:    time.Now().UTC(),
        },
    )
    if err != nil {
        log.Fatalf("Ошибка публикации: %v", err)
    }

    log.Println("Сообщение успешно отправлено в брокер!")
}

Сообщения и фреймы под капотом

Когда в Go вызывается метод публикации (ch.Publish(...) или современный ch.PublishWithContext(...)), полезная нагрузка не передается единым монолитным куском байт. На транспортном уровне AMQP 0-9-1 сообщение разбивается на серию стандартизированных фреймов (Frames):

flowchart TD
    subgraph Stream ["Поток байт по TCP (Channel ID: 1)"]
        direction LR
        F1["Method Frame<br>(Basic.Publish, Exchange, RoutingKey)"]
        F2["Header Frame<br>(ContentType, DeliveryMode, Headers, BodySize)"]
        F3["Body Frame 1<br>(Payload Chunk 1)"]
        F4["Body Frame N<br>(Payload Chunk N)"]

        F1 --> F2 --> F3 --> F4
    end

    subgraph Internal ["Zero-Copy маршрутизация внутри RabbitMQ"]
        Parser["Парсер заголовков (BEAM)"]
        Router["Маршрутизатор (Exchange)"]
        QueueBuffer["Буфер очереди (RAM)"]

        Parser -->|"Анализирует только фреймы 1 и 2"| Router
        Router -->|"Перемещает указатель на Body"| QueueBuffer
    end

    F1 -.-> Parser
    F2 -.-> Parser
    F3 -.-> QueueBuffer
    F4 -.-> QueueBuffer

    classDef frame fill:#1e293b,stroke:#38bdf8,stroke-width:2px,color:#f8fafc;
    classDef internal fill:#312e81,stroke:#a5b4fc,stroke-width:1px,color:#f8fafc;

    class F1,F2,F3,F4 frame;
    class Parser,Router,QueueBuffer internal;
  

Каждое передаваемое сообщение состоит как минимум из 3 типов фреймов:

1. Method Frame: Инструкция для брокера (Basic.Publish). Содержит имя обменника, Routing Key и флаги (mandatory, immediate).
2. Header Frame: Метаданные (свойства Basic.Properties). Содержит размер полезной нагрузки в байтах (BodySize), режим доставки (DeliveryMode: transient или persistent), ContentType, CorrelationId и пользовательские заголовки.
3. Body Frame(s): Непосредственный срез байт (payload). Если полезная нагрузка превышает максимальный размер фрейма (frame_max, по умолчанию 128 КБ), рантайм разбивает ее на цепочку последовательных Body Frame.

Zero-Copy и оптимизация маршрутизации

Такое строгое разделение протокола дает RabbitMQ огромное преимущество в производительности:
Для маршрутизации сообщения брокеру вообще не нужно парсить и десериализовать тело сообщения (Body). Брокер считывает лишь легковесные Method и Header фреймы, вычисляет целевые очереди и просто связывает внутренние указатели на сырые байты Body в памяти Erlang. Сами данные перемещаются без лишних аллокаций.

Итог

1. Модель «Умный брокер — глупый клиент»: RabbitMQ берет на себя маршрутизацию, дедупликацию в буферах, управление состоянием очередей и гарантии доставки, разгружая код сервисов.
2. Связка AMQP 0-9-1: Сообщения никогда не направляются прямо в очередь. Маршрут всегда выглядит так:
   $$\text{Producer} \xrightarrow{\text{Publish}} \text{Exchange} \xrightarrow{\text{Binding + Routing Key}} \text{Queue} \xrightarrow{\text{Deliver}} \text{Consumer}$$
3. Процессная модель Erlang: Очередь — это один легковесный процесс BEAM. Предел производительности одной очереди упирается в производительность одного ядра CPU. Высокие нагрузки требуют шардирования очередей или перехода на Quorum Queues.
4. Сетевой уровень в Go: Одно тяжелое *amqp.Connection на процесс и независимые легковесные *amqp.Channel на каждую рабочую горутину. Совместное использование одного канала между параллельными горутинами порождает lock contention и риски падения воркеров при сбросе канала.
5. Фрейминг: Разделение на Method, Header и Body фреймы позволяет RabbitMQ выполнять маршрутизацию со скоростью перекладывания указателей в памяти без десериализации полезной нагрузки.

Теперь, когда мы изучили каркас брокера и основы протокола AMQP, пришло время разобраться в главном механизме сортировки сообщений — обменниках. В следующей лекции 2. Exchanges. Direct, Fanout, Topic, Headers мы детально разберем все типы Exchange, их битовые маски, деревья поиска и практические сценарии применения в Go.