Обменники: direct exchange и routing key

Дефолтный обменник из пятого урока работал как почтовый ящик: публикация с пустым именем обменника уходила прямо в очередь с тем же именем, и на этом маршрутизация заканчивалась. Как только очередей становится несколько, схема трещит: продюсер обязан знать имена всех очередей, добавление нового потребителя означает правку кода издателя, а разослать одно событие трём командам — это цикл по очередям в продакшн-коде. AMQP решает это отдельной сущностью — exchange, обменником: продюсер публикует в обменник с routing key, очереди привязаны к обменнику с binding key, и брокер сам раздаёт копии по правилам привязки. Издатель знает один адрес — обменник; потребители настраивают свою видимость привязками, не трогая код продюсера. В этом уроке разбираем типы обменников, сигнатуру ExchangeDeclare из amqp091-go и поведение direct — самого предсказуемого из четырёх, — а заодно находим место, где сообщения пропадают молча.

Rabbitmq exchange: четыре типа и дефолтный#

Обменник в AMQP 0-9-1 — именованная сущность брокера, у которой есть тип, и именно тип определяет правило раздачи копий. Direct отправляет сообщение тем очередям, чей binding key в точности равен routing key публикации. Fanout игнорирует ключ и шлёт копию каждой привязанной очереди. Topic сравнивает ключ с шаблоном из звёздочек и решёток. Headers маршрутизирует по заголовкам сообщения, не глядя на ключ. Обменник по умолчанию с пустой строкой в имени — особый случай: это прямой алиас на каждую очередь, ключ равен имени очереди, и создаёт его сам брокер. Ещё в каждом vhost живут предустановленные amq.direct, amq.fanout, amq.topic и amq.headers — ими удобно пользоваться для быстрых экспериментов, но под задачу обменник объявляют свой: имя, тип и durable вы выбираете в коде. Каждому типу в курсе отведён свой урок: здесь direct, дальше fanout, topic и headers.

Direct exchange go: ExchangeDeclare и QueueBind#

Прямую маршрутизацию direct exchange go собирают двумя вызовами. Первый — ch.ExchangeDeclare(name, kind, durable, autoDelete, internal, noWait, args) error: объявляет обменник events типа direct; durable — переживает рестарт брокера, как durable-очередь из девятого урока; autoDelete — удаляется, когда отвяжется последняя очередь; internal — принимает публикации только от обменников, клиент напрямую в него писать не может; noWait — не ждать ответа сервера; args — табличка с редкими настройками, об одной из них, alternate exchange, поговорим в четырнадцатом уроке. Объявление идемпотентно: тот же вызов с теми же аргументами ничего не ломает, а попытка переобъявить существующий exchange другим типом закрывает канал с PRECONDITION_FAILED — та же механика, что у QueueDeclare. Второй вызов — ch.QueueBind(name, key, exchange, noWait, args): привязывает очередь к обменнику с binding key. Именно пара «обменник и ключ», а не имя очереди, становится адресом подписки потребителя.

Топология direct и публикация по меткам

package main

import (
        "context"
        "log"

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

type Task struct {
        Kind string
        Body string
}

func main() {
        conn, err := amqp.Dial("amqp://guest:guest@localhost:5672/")
        if err != nil {
                log.Fatal(err)
        }
        defer conn.Close()

        ch, err := conn.Channel()
        if err != nil {
                log.Fatal(err)
        }

        err = ch.ExchangeDeclare("tasks", "direct", true, false, false, false, nil)
        if err != nil {
                log.Fatal(err)
        }

        routes := map[string]string{
                "emails": "email",
                "push":   "push",
        }
        for q, key := range routes {
                if _, err := ch.QueueDeclare(q, true, false, false, false, nil); err != nil {
                        log.Fatal(err)
                }
                if err := ch.QueueBind(q, key, "tasks", false, nil); err != nil {
                        log.Fatal(err)
                }
        }

        ctx := context.Background()
        tasks := []Task{
                {Kind: "email", Body: "добро пожаловать"},
                {Kind: "push", Body: "статус заказа"},
                {Kind: "sms", Body: "код подтверждения"},
        }
        for _, t := range tasks {
                err := ch.PublishWithContext(ctx, "tasks", t.Kind, false, false,
                        amqp.Publishing{
                                ContentType:  "text/plain",
                                Body:         []byte(t.Body),
                                DeliveryMode: amqp.Persistent,
                        })
                if err != nil {
                        log.Fatal(err)
                }
                log.Println("опубликовано:", t.Kind, t.Body)
        }
}

Топология и публикация в одном месте: exchange tasks типа direct, две durable-очереди, каждая со своим ключом. Сообщение с ключом email получит только emails, push — только push, а sms не достанется никому: привязки с таким ключом нет, и брокер отбросит доставку. Ключ лежит во втором аргументе PublishWithContext — в дефолтном обменнике там было имя очереди, здесь стала меткой маршрута.

Четыре типа обменников рядом#

Тип	Правило доставки	Типовая задача
direct	binding key равен routing key в точности	адресная доставка: email, push, sms по метке
fanout	всем привязанным очередям, ключ игнорируется	broadcast: сброс кэша, аудит, события для всех
topic	ключ против шаблона со звёздочками и решётками	события по маске: order., .created, order.#
headers	совпадение заголовков по x-match, ключ не смотрят	маршрутизация по атрибутам: формат, версия
пустое имя	алиас: ключ равен имени очереди	точечная доставка в известную очередь

Опечатка в ключе: сообщение исчезает без ошибки

Кейс: команда переименовала метку маршрута с email на mail в продюсере, а привязку забыла. PublishWithContext вернул nil, логи чистые, а письма стоят: сообщение с ключом, к которому не привязана ни одна очередь, брокер выбрасывает сразу — это штатное поведение AMQP, а не сбой. В шестом уроке уже была лекарство со стороны издателя: флаг mandatory вместе с NotifyReturn возвращает недоставленное в приложение. Со стороны топологии помогает дисциплина объявлений: обменник и привязки выполняются одним кодом топологии при старте обеих сторон, а для промахов заводят очередь-ловушку — механику покажет четырнадцатый урок про alternate exchange.

Маршрутизация по ключу: что видит брокер#

Сравнение в direct строковое и точное: брокер берёт routing key публикации и ищет привязки с таким же binding key, без префиксов и масок, регистр учитывается. Из этого следуют четыре правила, на которых строятся все схемы. Одно сообщение может уйти в несколько очередей: две очереди с ключом email получат по копии. Одна очередь может держать несколько привязок: QueueBind по очереди с ключами email и welcome — сообщения по любому из них придут. Привязки живут на стороне брокера и переживают рестарт вместе с durable-обменником, поэтому топологию меняют осознанно: QueueUnbind отвязывает очередь, а повторный QueueBind с новым ключом старую привязку не удаляет. Ключ — это контракт между командами: строка вида domain.entity.action с точками-разделителями, которыми всерьёз распорядится topic-обменник из тринадцатого урока; direct точки не интерпретирует и сравнивает ключ целиком.

Одна очередь — два ключа, один ключ — две очереди

// одна очередь подписана на два ключа сразу
if err := ch.QueueBind("notifications", "email", "tasks", false, nil); err != nil {
        log.Fatal(err)
}
if err := ch.QueueBind("notifications", "welcome", "tasks", false, nil); err != nil {
        log.Fatal(err)
}

// вторая очередь делит ключ с первой: получит копию того же сообщения
if err := ch.QueueBind("audit", "email", "tasks", false, nil); err != nil {
        log.Fatal(err)
}

Публикация с ключом email уйдёт и в notifications, и в audit — по копии на каждую подходящую привязку. Ключ welcome достанется только notifications. QueueBind не заменяет предыдущие привязки очереди, а добавляет новые: чтобы маршрут исчез, нужен QueueUnbind с теми же аргументами.

Аргументы ExchangeDeclare: durable, autoDelete, internal#

Аргументы объявления определяют жизнь обменника, и проходят их в порядке важности. Durable хранит метаданные обменника на диске: вместе с durable-очередями и persistent-сообщениями из девятого урока это полная цепочка выживания; недолговечный обменник исчезнет при рестарте брокера, а привязки — вместе с ним. AutoDelete удаляет обменник после отвязки последней очереди: удобно для временных схем, рискованно для боевых — простой без потребителей стирает топологию, и первый же publisher после паузы пишет в пустоту. Internal делает обменник транзитным: публикации принимаются только от других обменников, чаще всего в связке с alternate exchange из четырнадцатого урока. NoWait экономит рейс до сервера, но ошибку объявления вы увидите позже, через закрытие канала по NotifyClose, — в коде этого курса ответ сервера ждём. Args принимает amqp.Table; список настроек обменника короткий, в отличие от очередей, где живут dead letter аргументы шестнадцатого урока.

Какой тип обменника выбрать под задачу#

Выбор сужается до одного вопроса: как потребители описывают свою видимость. Метки-адреса без иерархии — direct: перечислили ключи, получили ровно их. Всем всё — fanout, но каждая очередь тянет полную копию потока. Иерархия и маски — topic. Маршрутизация по данным сообщения, а не по строке ключа, — headers: вариант редкий, зато честный, когда атрибутов несколько и они опциональны. Спорное наблюдение из ревью: команды любят topic за гибкость и заводят его там, где хватает direct, а потом половина привязок висит мёртвым грузом, и никто не помнит, чей это маршрут. Direct честнее: каждая привязка — осознанный контракт. Практический критерий: если множество ключей известно на этапе дизайна — берите direct; если оно растёт по мере развития продукта, а потребители хотят срезы потока — topic. Сменить тип у существующего обменника нельзя: только новое имя и переезд.

Код урока — локально, интерактивы — на странице

Маршрутизация происходит внутри брокера, поэтому браузерной песочнице с её набором fmt, strings, strconv, math, sort, errors и time — без сети — обменник не показать: код урока запускается локально против Docker-контейнера из второго урока. Раскройте вкладку Exchanges в management UI: там видны ваш tasks, дефолтный и предустановленные amq.*, а клик по обменнику открывает форму публикации с полем routing key — удобно проверять привязки руками. Интерактивы ниже проверят сигнатуры и поведение ключей.

Правила обменников#

Продюсер публикует в обменник, а не в очередь; адрес подписки — пара «обменник и ключ».
ExchangeDeclare идемпотентен при тех же аргументах; смена типа закрывает канал PRECONDITION_FAILED.
Direct сравнивает binding key и routing key как строки: точно, с учётом регистра, без префиксов.
Копия уходит каждой подходящей привязке; одна очередь может держать несколько ключей.
Сообщение без подходящей привязки выбрасывается молча — mandatory с NotifyReturn или alternate exchange.
Durable-обменник переживает рестарт только вместе с durable-очередями и persistent-сообщениями.

Что дальше: шестой урок уже показал mandatory и NotifyReturn — издательскую половину защиты от промахов, девятый объяснил durable, на который опирается durable-обменник. Широковещательный вариант того же механизма — двенадцатый урок про fanout, маски по ключам — тринадцатый про topic, маршрутизация по заголовкам и очередь-ловушка для промахов — четырнадцатый. Похожий переход «один адрес вместо списка очередей» вы могли видеть в pub/sub Redis — там роль обменника играет канал. Объявление очередей, к которым всё это привязывается, осталось в пятом уроке, карта курса — на странице курса.

Проверка: дефолтный обменник

Публикация ch.PublishWithContext(ctx, "", "tasks", false, false, msg) — куда попадёт сообщение?

Проверка: привязки direct

Очередь notifications привязана к exchange tasks с ключом email, очередь audit — тоже с ключом email. Публикация с routing key email. Что произойдёт?

Что напечатает продюсер?

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

err := ch.ExchangeDeclare("events", "direct", true, false, false, false, nil)
if err != nil {
        log.Fatal(err)
}
_, err = ch.QueueDeclare("clicks", true, false, false, false, nil)
if err != nil {
        log.Fatal(err)
}
// привязку забыли
err = ch.PublishWithContext(ctx, "events", "click", false, false,
        amqp.Publishing{Body: []byte("click-1")})
fmt.Println(err == nil)

Соберите топологию direct

Допишите объявление обменника, привязку очереди и ключ публикации.

err := ch.("tasks", "direct", true, false, false, false, nil) if err != nil { log.Fatal(err) } err = ch.("emails", "email", "tasks", false, nil) if err != nil { log.Fatal(err) } err = ch.PublishWithContext(ctx, "tasks", , false, false, amqp.Publishing{Body: []byte("привет")})

Соберите публикацию через direct exchange

Расставьте строки в порядке выполнения: соединение, канал, обменник, очередь, привязка, публикация.

Строки

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

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

Карточки: exchange и маршрутизация

Термин

ExchangeDeclare

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

Определение

Объявление обменника: имя, тип, durable, autoDelete, internal, noWait, args. Идемпотентно при тех же аргументах; смена типа закрывает канал.

1 / 6

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

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

Вывод

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

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

Продюсер пишет в обменник с routing key; очереди получают копии через привязки QueueBind.
Direct сравнивает ключи точно: копия уходит каждой очереди с совпавшей привязкой.
Дефолтный обменник — алиас на очереди: ключ равен имени очереди, привязки не нужны.
Сообщение без подходящей привязки исчезает молча: mandatory с NotifyReturn или alternate exchange.
ExchangeDeclare идемпотентен при тех же аргументах; переобъявление другим типом закрывает канал.
Одна очередь — несколько ключей, один ключ — несколько очередей: оба варианта штатные.
Выбор типа: метки — direct, всем — fanout, маски — topic, атрибуты — headers.

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

Соберите обменник notifications типа direct и три очереди: emails с ключом email, sms с ключом sms и копилку audit с привязками email и sms. Публикуйте сообщения по одной метке и проверяйте в management UI, где оказались копии. Затем отправьте сообщение с ключом voice, к которому привязок нет, включите mandatory с NotifyReturn из шестого урока и убедитесь, что возврат приходит в горутину, а не в err от публикации.

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

Можно ли переобъявить exchange с другим типом?

Нет: тип фиксируется первым объявлением, повторное объявление с другим типом закрывает канал PRECONDITION_FAILED — той же ошибкой, что и перекраска durable-очереди из девятого урока. Путь один: новый обменник с новым именем, переключение издателей и перевод привязок, поэтому тип выбирают до первого деплоя.

Почему не публиковать напрямую в очередь, если имя известно?

Можно: дефолтный обменник работает именно так, и для скриптов этого хватает. По мере роста схемы продюсер начинает знать имена всех очередей и меняться вместе с ними; именованный обменник прячет топологию за одним адресом, а привязки перестраиваются без правки издателя. Практичное правило: дефолтный — для экспериментов, именованный — для всего, что живёт дольше одного дня.

Чем binding key отличается от routing key?

Routing key — атрибут публикации, его ставит продюсер. Binding key — атрибут привязки, его выбирает владелец очереди при QueueBind. В direct брокер сравнивает строки в точности; в topic сравнение идёт по правилам масок из тринадцатого урока, а в fanout routing key не смотрят вовсе.

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

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