--- title: "Fanout: рассылка всем подписчикам" description: "Fanout exchange в amqp091-go: рассылка всем подписчикам, своя очередь потребителя, broadcast и цена широковещания. Rabbitmq pub sub на Go с примерами." course: "RabbitMQ в Go: очереди сообщений на amqp091-go" url: "https://golangmaster.ru/courses/rabbitmq/lessons/12-fanout-exchange/" published: "2026-09-19" updated: "2026-09-21" minutes: 16 --- # Fanout: рассылка всем подписчикам *Курс «RabbitMQ в Go: очереди сообщений на amqp091-go», урок 12 из 25. ~16 минут. Обновлено 21.09.2026. Источник: https://golangmaster.ru/courses/rabbitmq/lessons/12-fanout-exchange/* Когда одно событие нужно всем — письму, аналитике и аудиту — его раздаёт fanout exchange: обменник, который копирует каждое сообщение во все привязанные очереди, не глядя на routing key. В этом уроке собираем rabbitmq pub sub на amqp091-go: свою очередь подписчика, продюсера без маршрутов и цену широковещания. Разберём, почему опоздавший подписчик теряет историю, чем именованная очередь отличается от анонимной и когда broadcast превращается в умножитель нагрузки. Заказ оформлен, и это событие нужно трём системам сразу: сервису писем — чтобы отправить подтверждение, аналитике — для воронки, аудиту — для журнала. Наивное решение — три публикации в три очереди из кода продюсера — быстро обрастает проблемами: добавилась четвёртая система, и в код заказа правят все, кто при нём присутствовал; у системы-подписчика отпуск, а её очередь продолжает наполняться чужими обязанностями. Паттерн, который это лечит, называется 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 убьёт очередь вместе с соединением — годится для временных сессий вроде просмотра потока в отладке, но не для подписки, которую жалко терять при каждом рестарте. ### Подписчик со своей очередью ```go 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 — и три системы уже держат копию каждая в своей очереди. ### Одна публикация — копии всем ```go 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 из тринадцатого урока. Соседи по маршрутизации: [одиннадцатый урок](/courses/rabbitmq/lessons/11-obmenniki-direct/) дал direct и привязки, на которых стоит fanout, [тринадцатый про topic](/courses/rabbitmq/lessons/13-topic-exchange/) добавит фильтрацию по маскам, [четырнадцатый про headers](/courses/rabbitmq/lessons/14-headers-alternativy/) — маршрутизацию по атрибутам и alternate exchange. Выбор очереди для подписчика опирается на [пятый урок про QueueDeclare](/courses/rabbitmq/lessons/05-ochered-prostaya/), дубли и идемпотентность — на [восьмой про ack](/courses/rabbitmq/lessons/08-ack-nack/). Та же модель рассылки в другом брокере — [pub/sub Redis](/courses/go-redis/lessons/14-pub-sub/): каналы вместо обменников и то же правило «опоздал — не увидел». События из этого урока станут payload'ом в [пятнадцатом уроке про JSON](/courses/rabbitmq/lessons/15-json-serializaciya/), карта курса — на [странице курса](/courses/rabbitmq/). ### Проверка: механика fanout **Вопрос:** К exchange orders типа fanout привязаны три очереди. Публикация ch.PublishWithContext(ctx, "orders", "", false, false, msg). Сколько копий создаст брокер? - Одну: fanout отдаёт сообщение первой свободной очереди - ✅ Три: по копии каждой привязанной очереди - Ноль: для fanout нужен непустой routing key - Зависит от prefetch потребителей Fanout игнорирует ключ и раздаёт копию каждой привязанной очереди — три очереди, три копии. Prefetch управляет темпом потребления, но не количеством копий; пустой ключ для fanout — норма, а не ошибка. ### Проверка: очереди подписчиков **Вопрос:** Сервис аналитики запущен в двух инстансах и должен обработать каждое событие ровно один раз на команду. Как расставить очереди? - Каждый инстанс объявляет свою exclusive-очередь и привязывает к обменнику - ✅ Оба инстанса встают в одну именованную durable-очередь, привязанную к обменнику - Каждый инстанс привязывает свою durable-очередь: брокер поделит события пополам - Очередь не нужна: fanout доставит события прямо в горутину потребителя Одна именованная очередь на сервис: инстансы делят поток, как воркеры с prefetch из десятого урока, и каждое событие обработает ровно один процесс. Отдельная очередь каждому инстансу дала бы копию каждой — двойной учёт; exclusive умерла бы вместе с соединением. ### Что напечатает продюсер? ```go // топология: 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) ``` **Что выведет программа?** - ✅ true дважды: отвязка не ошибка издателя, evt-2 получит только очередь a - false на второй публикации: обменник без части привязок возвращает ошибку - true дважды, но evt-2 не получит никто: обменник запомнил состав подписчиков - true дважды, evt-2 получат обе очереди: fanout кэширует участников Обе публикации возвращают nil: издатель не знает состав подписок, а fanout раздаёт копии по текущему списку привязок. После отвязки b сообщение придёт только a; если бы отвязали обе, evt-2 был бы выброшен без единой ошибки. ### Соберите подписчика fanout Допишите объявление обменника, очередь и привязку подписчика. ```go 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) ``` Подсказка: Обменник объявляет ExchangeDeclare, очередь — QueueDeclare, а привязку к fanout делает QueueBind с пустым ключом. Правильные ответы в пропусках: `ExchangeDeclare`, `QueueDeclare`, `QueueBind` ### Соберите подписку на broadcast Расставьте строки в порядке выполнения: соединение, канал, обменник, очередь, привязка, чтение. Правильный порядок строк: 1. `conn, err := amqp.Dial(url)` 2. `ch, err := conn.Channel()` 3. `err = ch.ExchangeDeclare("orders", "fanout", true, false, false, false, nil)` 4. `q, err := ch.QueueDeclare("orders.analytics", true, false, false, false, nil)` 5. `err = ch.QueueBind(q.Name, "", "orders", false, nil)` 6. `d := <-msgs` ### Карточки: fanout и pub/sub - **fanout exchange** — Обменник-копир: отправляет сообщение каждой привязанной очереди, routing key игнорирует. - **своя очередь подписчика** — Очередь на независимого потребителя: полный поток и свой темп ack. Конкуренты одного сервиса делят её. - **QueueBind к fanout** — Привязка с пустым ключом: фильтрации нет, ключ ни на что не влияет. - **обменник и история** — Сообщение без привязанных очередей выбрасывается; догнать пропущенное fanout не может. - **autoDelete-очередь** — Удаляется после отключения последнего потребителя; привязка уходит вместе с ней. - **цена broadcast** — Каждая привязка — полная копия потока: диск, память и consumer_timeout у каждого подписчика свои. ## Коротко: что запомнить - 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 продюсер знает один обменник, а состав подписчиков меняется привязками на стороне брокера, без правки издателя. Ручной цикл по очередям требует перечислить всех получателей в коде, синхронизировать деплои и обрабатывать частичные сбои: первые три очереди приняли, четвёртая недоступна — что делать с транзакцией?