Модуль RabbitMQ: обменники и очереди

Брокеры сообщений — второй после баз ходовой класс зависимостей, и у RabbitMQ своя специфика: тест проверяет не запросы, а доставку. Пока издатель и читатель живут в разных частях системы, мок брокера подтверждает только сам факт вызова — маршрутизацию, биндинги и подтверждения он подделать не может по определению. Модуль RabbitMQ закрывает разрыв: внутри он собирает GenericContainer с образом брокера, выставляет переменные окружения для учётки, публикует порты 5672 и 15672, монтирует собственный конфигурационный файл с пользователем и паролем — и дожидается полной готовности, прежде чем вернуть управление. Поверх этого даёт типизированный контейнер с методами-адресами: AmqpURL для клиентов, HttpURL для менеджмента. Дальше остаётся только код теста: топология, публикация, чтение. По счёту это четвёртый модуль курса: базы и брокеры закрывают большую часть реальных зависимостей сервисного кода. Разница с ручной сборкой примерно та же, что между запуском теста и настройкой стенда руками перед каждым прогоном.

Почему testcontainers rabbitmq, а не GenericContainer#

GenericContainer умеет поднять образ rabbitmq, но про брокера не знает ничего: переменные RABBITMQ_DEFAULT_USER и RABBITMQ_DEFAULT_PASS, порты 5672 и 15672, ожидание готовности — всё руками. С готовностью у брокера отдельная засада: порт слушается рано, узел стартует не один заход, и соединение, принятое слишком рано, рвётся с загадочной ошибкой. Модуль закрывает это разом. Он знает про порты брокера, подставляет учётку в окружение и пишет брокеру собственный конфигурационный файл, где заданы пользователь и пароль, — путь до него передаётся через переменную RABBITMQ_CONFIG_FILE. Готовность ловится по строке из лога, а не по открытому порту: порт готов раньше, чем маршрутизация. Взамен вы получаете типизированный RabbitMQContainer с полями учётки и методами-адресами, и rabbitmq тесты go перестают зависеть от того, кто в каком порядке поднял инфраструктуру.

Каркас: rabbitmq.Run с опциями

import (
    "context"
    "log"
    "testing"

    "github.com/testcontainers/testcontainers-go"
    "github.com/testcontainers/testcontainers-go/modules/rabbitmq"
)

func TestMain(m *testing.M) {
    ctx := context.Background()
    rmq, err := rabbitmq.Run(ctx,
        "rabbitmq:3-management-alpine",
        rabbitmq.WithAdminUsername("app"),
        rabbitmq.WithAdminPassword("secret"),
    )
    if err != nil {
        log.Fatalf("rabbitmq не поднялся: %v", err)
    }
    defer testcontainers.TerminateContainer(rmq)

    m.Run()
}

rabbitmq.Run принимает образ первым аргументом и возвращает типизированный RabbitMQContainer с ошибкой. Опции WithAdminUsername и WithAdminPassword превращаются в переменные RABBITMQ_DEFAULT_USER и RABBITMQ_DEFAULT_PASS — учётку, которую модуль подставит в AmqpURL. Образ берите с суффиксом management: в нём включён mgmt-плагин с HTTP-API. TerminateContainer в defer уберёт контейнер, даже если тесты упадут.

rabbitmq.Run: опции, порты и ожидание готовности#

Вход в модуль — функция rabbitmq.Run(ctx, образ, opts ...testcontainers.ContainerCustomizer): те же кастомайзеры ядра плюс собственные опции модуля. WithAdminUsername и WithAdminPassword задают учётку администратора брокера; ничего не передали — получите классических guest и guest. Опции попадают сразу в три места: в окружение контейнера, в конфигурационный файл и в поля контейнера — AmqpURL соберёт адрес именно с ними. Порты модуль публикует сам: 5672 для AMQP, 5671 для AMQP поверх TLS, 15672 и 15671 для HTTP-менеджмента — выбирать и пробрасывать их вручную не нужно. Готовность ожидается по строке Server startup complete из логов: брокер печатает её, когда узел полностью стартовал, с таймаутом в минуту на весь запуск — на холодной машине первый прогон тянет образ и занимает заметное время. Для TLS есть WithSSL: сертификаты монтируются в контейнер, конфиг переключается на SSL-листенер, а к ожиданию добавляется строка про запущенный TLS-листенер. Тонкие флаги брокера через опции модуля не меняются — для этого свой конфиг, но тестам обычно хватает дефолтов. Заметьте, как модуль держит одну учётку согласованной во всех местах сразу: поменяли WithAdminPassword — новый пароль окажется и в окружении, и в конфиге, и в адресе из AmqpURL, рассинхрона не бывает.

Живое место: строка Server startup complete

Живое место: брокер пишет Server startup complete один раз — в отличие от Postgres из четвёртого урока, никаких двойных стартов. Модуль ждёт её как регулярное выражение с таймаутом в минуту. Отсюда практика: не оборачивайте rabbitmq.Run в короткий context.WithTimeout — на холодной машине с загрузкой образа минута не предел, а убитый посреди старта контейнер оставит мусор. Проверить ожидание легко руками: поднимите брокер контейнером, выведите логи и найдите эту строку сами — заодно узнаете, сколько секунд старт занимает на вашей машине.

Тест «сообщение доставлено»: от Dial до Delivery

// rmq — *rabbitmq.RabbitMQContainer из TestMain
url, err := rmq.AmqpURL(ctx)
if err != nil {
    t.Fatalf("amqp url: %v", err)
}

conn, err := amqp.Dial(url)
if err != nil {
    t.Fatalf("dial: %v", err)
}
defer conn.Close()

ch, err := conn.Channel()
if err != nil {
    t.Fatalf("channel: %v", err)
}

if err := ch.ExchangeDeclare("orders", "fanout", true, false, false, false, nil); err != nil {
    t.Fatalf("exchange: %v", err)
}
q, err := ch.QueueDeclare("", false, false, true, false, nil)
if err != nil {
    t.Fatalf("queue: %v", err)
}
if err := ch.QueueBind(q.Name, "", "orders", false, nil); err != nil {
    t.Fatalf("bind: %v", err)
}

msgs, err := ch.Consume(q.Name, "tester", true, false, false, false, nil)
if err != nil {
    t.Fatalf("consume: %v", err)
}

if err := ch.PublishWithContext(ctx, "orders", "", false, false,
    amqp.Publishing{ContentType: "text/plain", Body: []byte("order-42")}); err != nil {
    t.Fatalf("publish: %v", err)
}

select {
case d, ok := <-msgs:
    if !ok {
        t.Fatal("канал доставки закрыт")
    }
    if string(d.Body) != "order-42" {
        t.Fatalf("тело: %s", d.Body)
    }
case <-time.After(5 * time.Second):
    t.Fatal("сообщение не доставлено за 5 секунд")
}

AmqpURL отдаёт адрес с маппированным портом и учёткой из опций, amqp.Dial открывает соединение, Channel — сессию поверх него. Анонимная очередь с exclusive=true живёт ровно до закрытия соединения: имя брокер даёт сам, мусора в брокере не остаётся. Consume возвращает Go-канал доставки, select с таймаутом превращает тишину брокера в падение с ясным текстом вместо вечного зависания теста.

AmqpURL и amqp.Dial: подключение из теста#

RabbitMQContainer.AmqpURL(ctx) возвращает amqp://логин:пароль@хост:порт — логин и пароль берутся из опций Run, хост и порт из PortEndpoint по порту 5672/tcp, то есть из маппированного адреса Docker. Порт меняется от запуска к запуску: тест не дерётся с соседями за фиксированный 5672, параллельные прогоны живут спокойно. Ручная сборка адреса — классическая ошибка: захардкоженный localhost:5672 работает на одном ноутбуке и падает в CI. Подключайтесь клиентом amqp091-go: amqp.Dial(url) открывает соединение, conn.Channel() — канал, и это вся подготовка перед тестом. Канал в AMQP — лёгкая сессия поверх соединения: топология и публикации гоняются через каналы, и для теста одного хватает. Есть парные методы AmqpsURL и HttpsURL для TLS-вариантов и HttpURL для менеджмента — о нём отдельный разговор ниже. Соединение закрывайте Close-ом в defer, иначе тестовый процесс может повиснуть на выходе. Открывать ли соединение на каждый тест или одно на пакет — вопрос масштаба: один пакет на TestMain дешевле, но упавшее состояние канала тянется между тестами, поэтому топологию и публикации лучше держать на своём канале в каждом тесте.

Проверка: учётка по умолчанию

rabbitmq.Run вызвали без WithAdminUsername и WithAdminPassword. Что окажется в URL, который вернёт AmqpURL?

Топология теста: обменник, очередь, биндинг#

Сам тест собирается из трёх шагов AMQP: топология, публикация, чтение. Топологию объявляют через канал: ExchangeDeclare создаёт обменник — fanout рассылает всем подписанным очередям, direct маршрутизирует по ключу, topic — по шаблону ключа; QueueDeclare заводит очередь; QueueBind привязывает её к обменнику по ключу маршрутизации. Для теста удобно заводить анонимную очередь с пустым именем и exclusive=true: брокер сам даст уникальное имя, а умрёт очередь вместе с соединением — топология теста не утечёт в соседние тесты. Публикация — PublishWithContext с Publishing: имя обменника, ключ маршрутизации, флаг mandatory и тело. Чтение — Consume: он отдаёт Go-канал с Delivery, в цикле или select-е читаете Body и подтверждаете Ack, если autoAck выключен. Отдельно про durable: флаг означает, что объект переживёт рестарт брокера, а сообщение с DeliveryMode 2 — переживёт вместе с ним; в тестах это нужно редко, но помните о правиле — durable-очередь может биндиться только в durable-обменник. Тест «сообщение доставлено» — это select с таймаутом: пришло — сравните тело с отправленным; тишина пять секунд — t.Fatal с человеческим текстом. Так проверяется вся цепочка: обменник, биндинг, очередь, доставка — то, что мок никогда не покажет.

Опции и методы модуля RabbitMQ#

Опция или метод	За что отвечает	Пример
WithAdminUsername	логин администратора; без опции — guest	WithAdminUsername("app")
WithAdminPassword	пароль администратора; без опции — guest	WithAdminPassword("secret")
WithSSL	TLS: сертификаты монтируются, конфиг переключается на SSL-листенер	WithSSL(settings)
AmqpURL	amqp://логин:пароль@хост:порт для AMQP-клиентов	rmq.AmqpURL(ctx)
AmqpsURL	тот же адрес поверх TLS, порт 5671	rmq.AmqpsURL(ctx)
HttpURL	http://хост:порт — HTTP-API и интерфейс менеджмента	rmq.HttpURL(ctx)

HttpURL: mgmt-плагин и его грабли#

HttpURL(ctx) возвращает http://хост:порт — адрес HTTP-API и веб-интерфейса менеджмента на порту 15672. Через него тесты проверяют то, что не видно из AMQP: списки очередей, глубину, соединения и каналы — обычным http.Get из стандартной библиотеки, API отдаёт JSON. Грабля одна, но больная: mgmt-плагин должен быть включён в образе. Порт 15672 модуль публикует всегда, поэтому HttpURL вернёт адрес в любом случае — но на образе без management внутри контейнера на этом порту никто не слушает, и первый же запрос упадёт с отказом в соединении. Ошибка всплывёт не в Run и не в HttpURL, а на первом HTTP-запросе, и выглядит как проблема сети, хотя дело в выборе образа. Берите образы с суффиксом management — они и стоят в примерах модуля. Второй нюанс: HttpURL отдаёт адрес без учётки — авторизация делается Basic-аутентификацией той же парой из опций, а guest подходит только с дефолтными настройками.

Проверка: отказ в соединении по HttpURL

Тест получил rmq.HttpURL(ctx) и сделал http.Get — соединение отклонено. Контейнер поднялся без ошибок. В чём причина?

Что вернёт AmqpURL

Прочитайте код и выберите, что он выведет:

rmq, err := rabbitmq.Run(ctx, "rabbitmq:3-management-alpine",
    rabbitmq.WithAdminUsername("app"),
    rabbitmq.WithAdminPassword("secret"),
)
// err == nil

url, err := rmq.AmqpURL(ctx)

Типовые ошибки, на которых ломаются первые прогоны. Первая — авто-удаление: очередь с autoDelete=true удаляется, когда от неё отцепляется последний читатель; в тесте, где Consume закрывается раньше публикации, сообщение уходит в никуда — брокер его выбрасывает. Ошибка плавает: с быстрым читателем всё зелёное, в CI — красное. Вторая — exclusive-очереди: очередь с exclusive=true доступна только своему соединению, чтение из второго закончится отказом доступа, а сама очередь умрёт вместе с первым соединением. Третья — повторное объявление топологии с другими параметрами: ExchangeDeclare того же имени, но с другим типом или флагом durable закрывает канал ошибкой 406 PRECONDITION_FAILED, и все дальнейшие операции на канале падают. Четвёртая — autoAck=false без Ack: брокер копит неподтверждённые доставки, и очередь раздувается прямо во время прогона. Пятая — короткий контекст вокруг Run: старт с загрузкой образа легко выходит за секунды, убитый посреди старта контейнер оставляет после себя мусор.

Чек-лист запуска брокера в тесте#

rabbitmq.Run вместо GenericContainer: окружение, порты и конфиг уже настроены
образ с суффиксом management — иначе mgmt-плагина и HttpURL не будет
WithAdminUsername и WithAdminPassword — учётка для AmqpURL и Basic-аутентификации
подключение — amqp.Dial по AmqpURL; канал через conn.Channel()
топология теста — ExchangeDeclare, QueueDeclare, QueueBind; анонимная exclusive-очередь для изоляции
доставка — Consume и select с таймаутом; без таймаута тихий тест висит вечно
TerminateContainer в defer — уборка независимо от результата теста

Соберите запуск брокера

Восстановите вызовы: старт контейнера, учётка, адрес AMQP и подключение клиента.

rmq, err := rabbitmq.(ctx, "rabbitmq:3-management-alpine", rabbitmq.("app"), rabbitmq.WithAdminPassword("secret"), ) url, err := rmq.(ctx) conn, err := amqp.(url)

Расставьте минимальный тест по порядку

Соберите путь теста: контекст, старт брокера, проверка ошибки, уборка, адрес, соединение.

Строки

Ваша программа

Нажимайте на строки слева, чтобы собрать программу

Карточки: модуль RabbitMQ

Термин

rabbitmq.Run

нажмите, чтобы перевернуть

Определение

Вход в модуль: образ первым аргументом, дальше кастомайзеры. Возвращает типизированный RabbitMQContainer с ошибкой; дефолтная учётка — guest и guest, порты 5672 и 15672 публикуются сами.

1 / 6

Про песочницу: нужен локальный Docker

Честно: в браузерной песочнице этот урок не оживить — Docker там нет, а брокеру нужен живой демон. Рабочая схема — локально: установленный Go, запущенный Docker и обычный go test в терминале. Первая сборка скачает образ на пару сотен мегабайт, дальше он кешируется. И держите демон запущенным: ошибка про подключение к Docker — это не про тест, а про окружение. Прогоните пакет дважды: второй запуск покажет реальную скорость старта брокера на вашей машине. Если тест падает только на CI, а локально зелёный — в девяти случаях из десяти дело в таймаутах и холодном кеше образов, а не в коде.

Что дальше: брокер в контейнере — половина дела, вторая — читаемый тест поверх него. Предыдущий урок про Kafka — модуль Kafka: продюсер и консьюмер — показал ту же идею на другом брокере, сравнение двух подходов поучительно само по себе. Глубже в сам RabbitMQ — курс rabbitmq: обменники, биндинги и подтверждения там разобраны отдельно от тестов. Про комбинации ожиданий — двенадцатый урок wait-стратегии: когда сервис готов, а обзор всего пути — курс testcontainers от установки до CI. И проверка на честность: вынесите топологию из теста в TestMain и запустите два теста параллельно — поведение анонимных очередей и разделяемого обменника расскажет об изоляции больше, чем абзац теории.

Песочница урока

Редактируйте код и запускайте его прямо здесь — интерпретатор работает в браузере, ничего устанавливать не нужно. За каждый запуск — +2 XP.

Вывод

Запустите код, чтобы увидеть вывод.

Коротко: что запомнить

rabbitmq.Run поднимает настоящий брокер: WithAdminUsername и WithAdminPassword задают учётку, дефолт — guest и guest.
AmqpURL отдаёт amqp://логин:пароль@хост:порт с маппированным портом; подключение — amqp.Dial и conn.Channel().
Топология объявляется каналом — ExchangeDeclare, QueueDeclare, QueueBind; тест доставки — Consume, select с таймаутом и Ack.
HttpURL ведёт в mgmt-плагин на 15672: нужен образ с management, без него запросы падают с отказом соединения.
Грабли: autoDelete теряет сообщения без читателя, повторное объявление топологии роняет канал ошибкой 406.

Домашнее задание

Соберите пакет с TestMain: rabbitmq.Run с management-образом и своей учёткой, подключение через AmqpURL и amqp.Dial. Напишите тест «сообщение доставлено» через fanout-обменник и анонимную очередь с select-ом и таймаутом. Затем сломайте его осознанно: поставьте очереди autoDelete=true и закройте Consume до публикации — разберитесь по логам брокера, что стало с сообщением. В конце добавьте проверку через HttpURL: GET по пути /api/queues с Basic-аутентификацией — убедитесь, что очередь видна в менеджменте и её глубина совпадает с ожиданиями. Финальный шаг: запустите два теста параллельно с t.Parallel и посмотрите, как анонимные очереди разводят их трафик.

Вопросы и ответы по уроку

Можно ли поднять брокера через GenericContainer без модуля?

Технически да: задайте переменные RABBITMQ_DEFAULT_USER и RABBITMQ_DEFAULT_PASS, опубликуйте порты 5672 и 15672 и соберите wait-стратегию по логу сами. Но потеряете типизированный RabbitMQContainer с готовыми AmqpURL и HttpURL, монтирование конфига с учёткой и учтённые дефолты. Модуль — тот же GenericContainer внутри, уже собранный под брокера: меньше кода в тестах и меньше флейков на старте, потому что готовность здесь ловится по логу, а не по порту. Образ поднимется один и тот же — разница в объёме вашего кода.

Почему порт в AmqpURL не 5672?

Docker маппирует порт контейнера на свободный порт хоста, и он меняется от запуска к запуску. AmqpURL делает эту работу через PortEndpoint: адрес всегда актуален, а параллельные прогоны не дерутся за один порт. Захардкоженный 5672 в тесте — классическая ошибка переноса кода с локального брокера на контейнерный.

Канал падает с 406 при объявлении обменника. Что случилось?

Это PRECONDITION_FAILED: обменник или очередь уже существуют с другими параметрами — другим типом, флагом durable или autoDelete. AMQP закрывает канал, и все операции на нём дальше падают, поэтому ошибка выглядит как сбой в совершенно другом месте теста. В тестах такое бывает при общей топологии в TestMain и повторном объявлении в тесте: держите параметры топологии в одном месте или используйте уникальные имена на каждый прогон.

Термины урока

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