Fanout: рассылка всем подписчикам

Заказ оформлен, и это событие нужно трём системам сразу: сервису писем — чтобы отправить подтверждение, аналитике — для воронки, аудиту — для журнала. Наивное решение — три публикации в три очереди из кода продюсера — быстро обрастает проблемами: добавилась четвёртая система, и в код заказа правят все, кто при нём присутствовал; у системы-подписчика отпуск, а её очередь продолжает наполняться чужими обязанностями. Паттерн, который это лечит, называется pub/sub: продюсер публикует событие один раз в точку обмена, а подписчики сами решают, что им слушать. В RabbitMQ роль точки обмена выполняет fanout exchange: копия уходит каждой привязанной очереди, и появление нового подписчика не меняет ни строчки в издателе. Соберём такую схему на amqp091-go и посмотрим на её слабые места — у широковещания они свои.

Fanout exchange: копия каждой привязанной очереди#

Fanout — самый простой тип обменника: при доставке он смотрит только на список привязок и копирует сообщение каждой очереди из него. Routing key при этом игнорируется полностью: публикуйте с пустой строкой или с меткой — раздача одинаковая; привязки к fanout тоже делают с пустым ключом, чтобы не создавать ложное впечатление фильтрации. Объявление знакомо по одиннадцатому уроку, меняется второй аргумент: ch.ExchangeDeclare("orders", "fanout", true, false, false, false, nil). Идемпотентность, PRECONDITION_FAILED при смене типа, durable-семантика — всё то же самое: тип у обменника один, это не direct с расширением, а отдельная сущность. В management UI fanout виден сразу: на его странице нет поля ключа при публикации, только список привязок. Именно этой простотой fanout ценен как каркас: когда появляется нужда фильтровать, схему переводят на topic из следующего урока, но структура «обменник плюс очереди» не меняется.

Своя очередь подписчика: изоляция потока#

Главное решение в подписчике — какую очередь привязать к обменнику, и ошибка здесь стоит истории событий. Общее правило: у каждого независимого потребителя своя очередь. Сервис писем привязывает очередь orders.emails, аналитика — orders.analytics; каждая получает полный поток событий и подтверждает его в своём темпе из восьмого урока. Если два инстанса одного сервиса должны делить нагрузку, они встают в одну очередь, как воркеры из седьмого, — но это всё ещё одна подписка сервиса, а не две. Очередь именуют по сервису и назначению: orders.emails, orders.analytics; имя делает привязку видимой в UI и объясняет, кто владелец. Анонимные очереди QueueDeclare с пустым именем тоже легальны: брокер сгенерирует имя, а флаг exclusive убьёт очередь вместе с соединением — годится для временных сессий вроде просмотра потока в отладке, но не для подписки, которую жалко терять при каждом рестарте.

Подписчик со своей очередью

package main

import (
        "context"
        "log"
        "os"
        "os/signal"

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

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("orders", "fanout", true, false, false, false, nil)
        if err != nil {
                log.Fatal(err)
        }

        q, err := ch.QueueDeclare("orders.analytics", true, false, false, false, nil)
        if err != nil {
                log.Fatal(err)
        }
        err = ch.QueueBind(q.Name, "", "orders", false, nil)
        if err != nil {
                log.Fatal(err)
        }

        ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt)
        defer stop()

        msgs, err := ch.ConsumeWithContext(ctx, q.Name, "analytics-sub",
                false, false, false, false, nil)
        if err != nil {
                log.Fatal(err)
        }

        for d := range msgs {
                log.Printf("событие: %s (redelivered=%v)", d.Body, d.Redelivered)
                if err := d.Ack(false); err != nil {
                        log.Println("ack failed:", err)
                }
        }
        log.Println("подписка закрыта")
}

Подписчик сам собирает свою часть топологии: fanout-обменник orders, очередь orders.analytics и привязка с пустым ключом. Ключ тут ни на что не влияет, но параметр обязателен — договорённость команды: пустая строка в привязках к fanout. Конкуренты-инстансы аналитики встают в ту же очередь по имени и делят поток, как воркеры с prefetch из десятого урока.

Какую очередь привязать к fanout#

Вариант	Поведение	Когда уместен
именованная durable	переживает рестарты, копит события, пока владелец выключен	боевые подписки сервисов
анонимная exclusive	умирает вместе с соединением, имя генерирует брокер	временные сессии и отладка
autoDelete-очередь	удаляется после отключения последнего потребителя	эфемерные представления, дашборды
общая очередь разных сервисов	каждый получит лишь часть копий	ошибка дизайна: у каждого сервиса свой поток

Опоздавший подписчик: обменник не помнит историю

Кейс: сервис уведомлений выкатили на день позже продюсера, и команда ожидала увидеть вчерашние заказы — их нет. Fanout — почтальон, а не архив: сообщение, опубликованное без привязанных очередей, доставить некуда, и оно выбрасывается так же, как немаршрутизируемое в direct из одиннадцатого урока. История появляется только у привязанных очередей: привяжете durable-очередь до первого события — получите полный поток. Отсюда два правила запуска: топологию подписчик объявляет до старта потребления, а для сценариев «догнать пропущенное» fanout не подходит — нужен журнал, как Streams в Redis, или хранилище событий поверх очереди. Близкий случай из практики: exclusive-очередь отладочной сессии умерла вместе с обрывом, а продюсер продолжал публиковать — подписка молча пропустила всё это время.

Rabbitmq pub sub: продюсер без маршрутов#

Продюсерская сторона rabbitmq pub sub получается короче прямой: адрес один, ключа нет. Публикация идёт в обменник orders с пустой строкой в routing key — параметр обязателен, значение безразлично. Флаг mandatory из шестого урока вместе с fanout осмысленного сочетания не даёт: у fanout нет «немаршрутизируемых» ключей, копии либо уходят всем привязанным, либо не уходят никому, если привязок нет вовсе, — и этот последний случай mandatory не поймает, возврата не будет. Событие лучше публиковать в компактном конверте: идентификатор, тип, время, полезная нагрузка; сериализация — тема пятнадцатого урока. DeliveryMode выбирается по цене вопроса: amqp.Persistent для событий, которые нельзя терять при рестарте брокера, из девятого урока. Один вызов PublishWithContext — и три системы уже держат копию каждая в своей очереди.

Одна публикация — копии всем

err := ch.ExchangeDeclare("orders", "fanout", true, false, false, false, nil)
if err != nil {
        log.Fatal(err)
}

ctx := context.Background()
body := []byte("order.created ord-1042")
err = ch.PublishWithContext(ctx, "orders", "", false, false,
        amqp.Publishing{
                ContentType:  "text/plain",
                Body:         body,
                DeliveryMode: amqp.Persistent,
        })
if err != nil {
        log.Fatal(err)
}

Публикация в fanout: обменник первым аргументом, пустая строка в routing key. Копия уйдёт в каждую привязанную очередь — emails, analytics, audit — независимо от того, какие потребители онлайн: очереди durable, копии лягут в них и дождутся владельцев.

Цена broadcast: умножение нагрузки и дубли#

Широковещание умножает: каждая новая привязка добавляет брокеру полную копию потока, и при дюжине подписчиков fanout превращается в дюжину очередей с одинаковыми данными, диском и consumer_timeout из восьмого урока на каждого. Подписка «на всякий случай» здесь дороже, чем в direct: мёртвая очередь с привязкой продолжает копить persistent-события и однажды упирается в лимиты. Дубли появляются и внутри: событие, опубликованное дважды после обрыва соединения, уйдёт обеим системам по два раза — идемпотентность подписчика, знакомая по at-least-once из восьмого урока, обязательна. Обратная ловушка — фильтрация на потребителе: подписчик получает всё, выбрасывает девяносто процентов, трафик и диск работают впустую; это сигнал, что схеме нужен topic из тринадцатого урока. Правило баланса: fanout — для событий, нужных половине систем и больше; для срезов потока — маски.

Обслуживание привязок: чистка и autoDelete#

Привязки к fanout живут на брокере и не зависят от потребителей, поэтому за ними нужен уход. Мёртвые привязки ищутся на странице обменника в management UI — там же можно отвязать очередь вручную; в скриптах то же делает rabbitmqctl list_bindings по vhost. Для очередей, которые должны исчезать вместе с потребителем, есть autoDelete: очередь удаляется после отключения последнего потребителя, и привязка уходит вместе с ней — панель подписок чистится сама. Тем же свойством обладает autoDelete-обменник: последний QueueUnbind стирает его. Для боевых потоков это редко нужно: durable-обменник с durable-очередями переживает всё и ведёт себя предсказуемее, а эфемерные схемы оставляют дашбордам и отладке. Спорная практика — отдельный fanout на каждый тип события: имён становится много, а разница с одним потоком и масками из тринадцатого урока исчезает; компромисс — один поток событий и тип обменника под способ фильтрации.

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

Fanout проверяется только на живом брокере: в песочнице доступны fmt, strings, strconv, math, sort, errors и time, сети нет, и код урока запускается локально против Docker-контейнера из второго урока. Откройте два терминала с разными подписчиками и одну публикацию: обе очереди получат копию одновременно. На вкладке обменника в management UI нажмите Publish message с пустым ключом — так удобно проверять привязки без кода. Интерактивы ниже проверят механику копирования и выбор очереди.

Правила fanout#

Fanout копирует сообщение каждой привязанной очереди; routing key игнорируется.
У каждой подписки своя очередь; конкуренты одного сервиса делят её, как воркеры.
Очередь привязывают до первого события: обменник историю не хранит.
Именованная durable — для боевых подписок, exclusive — для отладки и временных сессий.
Каждая привязка — полная копия потока: подписка «на всякий случай» дороже, чем кажется.
Фильтрация на потребителе — сигнал переезда на topic из тринадцатого урока.

Соседи по маршрутизации: одиннадцатый урок дал direct и привязки, на которых стоит fanout, тринадцатый про topic добавит фильтрацию по маскам, четырнадцатый про headers — маршрутизацию по атрибутам и alternate exchange. Выбор очереди для подписчика опирается на пятый урок про QueueDeclare, дубли и идемпотентность — на восьмой про ack. Та же модель рассылки в другом брокере — pub/sub Redis: каналы вместо обменников и то же правило «опоздал — не увидел». События из этого урока станут payload'ом в пятнадцатом уроке про JSON, карта курса — на странице курса.

Проверка: механика fanout

К exchange orders типа fanout привязаны три очереди. Публикация ch.PublishWithContext(ctx, "orders", "", false, false, msg). Сколько копий создаст брокер?

Проверка: очереди подписчиков

Сервис аналитики запущен в двух инстансах и должен обработать каждое событие ровно один раз на команду. Как расставить очереди?

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

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

// топология: fanout exchange "orders",
// очереди a и b привязаны
err = ch.PublishWithContext(ctx, "orders", "", false, false,
        amqp.Publishing{Body: []byte("evt-1")})
fmt.Println(err == nil)

// затем очередь b отвязали и удалили
err = ch.PublishWithContext(ctx, "orders", "", false, false,
        amqp.Publishing{Body: []byte("evt-2")})
fmt.Println(err == nil)

Соберите подписчика fanout

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

err := ch.("orders", "fanout", true, false, false, false, nil) if err != nil { log.Fatal(err) } q, err := ch.("orders.analytics", true, false, false, false, nil) if err != nil { log.Fatal(err) } err = ch.(q.Name, "", "orders", false, nil)

Соберите подписку на broadcast

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

Строки

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

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

Карточки: fanout и pub/sub

Термин

fanout exchange

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

Определение

Обменник-копир: отправляет сообщение каждой привязанной очереди, routing key игнорирует.

1 / 6

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

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

Вывод

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

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

Fanout копирует каждое сообщение всем привязанным очередям; routing key игнорируется.
У независимого подписчика своя очередь; инстансы одного сервиса делят её, как воркеры.
Обменник истории не хранит: очередь привязывают до первого события.
Именованная durable — для боевых подписок, exclusive и autoDelete — для временных сессий.
Каждая привязка — полная копия потока; мёртвые привязки копят диск и требуют чистки.
Фильтрация на потребителе — сигнал переезда на topic; broadcast оставляют событиям для всех.

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

Соберите fanout-обменник events и два подписчика: один с durable-очередью orders.emails, второй с анонимной exclusive. Остановите первый подписчик, опубликуйте три события, запустите его снова и убедитесь, что копии дождались его в очереди, а второй подписчик, перезапущенный позже, получил только новые события. Затем привяжите третью очередь, заметьте в management UI рост глубины при остановленном потребителе и оцените, не пора ли такой подписке маски topic из следующего урока.

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

Нужен ли routing key при публикации в fanout?

Параметр обязателен, но значение не играет роли: fanout раздаёт копии по списку привязок и ключ не читает. В команде обычно договариваются публиковать пустую строку, чтобы код не намекал на несуществующую фильтрацию; метка в ключе не сломает доставку, но собьёт с толку читающего.

Как не потерять события, если подписчик временно отключён?

Привязать именованную durable-очередь до первого события: копии будут копиться в ней, пока владелец выключен, и он догонит поток после запуска — при условии, что очередь не переполнена лимитами. Анонимные exclusive-очереди для этого не годятся: они умирают вместе с соединением, и всё, что не доставлено, продюсер уже не вернёт.

Чем fanout отличается от публикации в несколько очередей вручную?

Адресом и живучестью схемы: в fanout продюсер знает один обменник, а состав подписчиков меняется привязками на стороне брокера, без правки издателя. Ручной цикл по очередям требует перечислить всех получателей в коде, синхронизировать деплои и обрабатывать частичные сбои: первые три очереди приняли, четвёртая недоступна — что делать с транзакцией?

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

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