Подготовка окружения
Компонента kovalevdmv/1CRabbitMQ, обработка КлиентRMQ. Брокер - как в части 1.
Код примеров собран в расширении RMQ_lessons (malikov-pro/1CRabbitMQ) для запуска через YAxUnit.
Те же сценарии можно вызывать из консоли кода в серверном контексте - например bsl_console и аналоги.
На чём сосредоточена часть
В части 2 рабочая очередь отдавала каждую задачу ровно одному исполнителю. Здесь наоборот: одно сообщение получают все слушатели. Паттерн называют publish/subscribe (издатель–подписчик).
Чтобы показать идею, соберём простую «систему логирования»: одна программа шлёт сообщения лога, несколько получателей их принимают. Каждая запущенная копия получателя видит одни и те же сообщения - можно одного писать в файл, другого смотреть «на экране» (в ЖР / РС).
По сути сообщения рассылаются (broadcast) всем подписчикам.
Точки обмена (exchanges)
В частих 1–2 мы писали «в очередь». Полная модель RabbitMQ чуть точнее: производитель никогда не кладёт сообщение напрямую в очередь. Он публикует только в точку обмена (exchange). Exchange решает, в какие очереди положить сообщение - по своему типу и привязкам.
Кратко:
- Producer - отправляет сообщения;
- Queue - буфер;
- Consumer - получает;
- Exchange - точка входа: принял от producer U94; раздал по правилам.
Типы точек обмена: direct, topic, headers, fanout. В этой части рассмотрим fanout: рассылает каждое сообщение во все привязанные очереди. Как раз для логгера.
Добавим точку обмена logs.
// один раз перед публикацией / подпиской (идемпотентно при тех же параметрах)
ТочкаОбмена = ОМ_РМКУ_Настройки.ИмяОбменникаЧасть3(); // "logs"
ОМ_РМКУ_ManagementAPI.ОбъявитьExchange(ТочкаОбмена, "fanout");
Раньше в ОпубликоватьСообщение мы передавали пустую ТочкаОбмена - это обменник по умолчанию: сообщение шло в очередь с именем из ключа маршрутизации. Теперь публикуем в наш logs:
// Подключение и Канал - как в [части 1](//infostart.ru/1c/articles/2196530/)
// Exchange «logs» уже объявлен через Management API (тип fanout)
ТочкаОбмена = "logs";
ТекстСообщения = "info: Hello World!";
Ответ = КлиентRMQ.ОпубликоватьСообщение(
Канал, ТекстСообщения, "", ТочкаОбмена, Истина);
Ключ маршрутизации при публикации указать нужно, но для fanout его значение игнорируется.
Если к exchange никто не привязан, сообщение пропадёт - для логгера это нормально: некому слушать - некуда писать.
Временные очереди и привязки
В work queue имя очереди (task_queue) было общим для всех исполнителей. Для подписчика лога другое:
- при каждом подключении - своя пустая очередь (имя может выбрать сервер);
- после отключения подписчика очередь должна исчезнуть (
exclusive/ auto-delete).
В КлиентRMQ объявление очереди и привязка к exchange - один вызов ОбъявитьОчередь: пустое имя U94; серверное имя; ТочкаОбмена + ключ - binding.
// временная exclusive + auto-delete, привязка к fanout «logs»
Очередь = КлиентRMQ.ОбъявитьОчередь(
Канал,
"", // имя выберет сервер (amq.gen-…)
"", // binding key; для fanout не важен
"logs", // ТочкаОбмена
Ложь, // не durable
Истина, // exclusive
Истина); // auto-delete
ИмяОчереди = Очередь.ИмяОчереди;
Связь «exchange U94; очередь» и есть binding (привязка). С этого момента logs копирует входящие сообщения в нашу очередь.
В Management UI на вкладке Bindings при двух подписчиках увидите две очереди, обе связанные с logs.
Реализация
Издатель:
ТочкаОбмена = "logs";
ОМ_РМКУ_ManagementAPI.ОбъявитьExchange(ТочкаОбмена, "fanout");
Подключение = КлиентRMQ.ПодключитьсяКСерверу(URI);
Канал = КлиентRMQ.СоздатьКанал(Подключение);
КлиентRMQ.ОпубликоватьСообщение(Канал, ТекстСообщения, "", ТочкаОбмена, Истина);
КлиентRMQ.ЗакрытьКанал(Канал);
КлиентRMQ.ОтключитьсяОтСервера(Подключение);
Подписчик:
Четвёртый параметр СоздатьПолучателя - АвтоматическоеПодтверждение (no_ack). Здесь ставим Истина. В части 2 для work queue так нельзя было: задача должна пережить падение исполнителя и вернуться в общую очередь. У логгера другая модель: у каждого подписчика своя temporary-очередь, сообщение - копия broadcast, потеря одной копии не «съедает» задачу у остальных. Учебный приёмник только пишет в ЖР / РС - ручной ack здесь усложняет код без выигрыша. Если подписчик начнёт делать критичную работу (запись в ИБ с гарантией), переходите на Ложь + ПодтвердитьСообщение, как в части 2.
ТочкаОбмена = "logs";
ОМ_РМКУ_ManagementAPI.ОбъявитьExchange(ТочкаОбмена, "fanout");
Подключение = КлиентRMQ.ПодключитьсяКСерверу(URI);
Канал = КлиентRMQ.СоздатьКанал(Подключение);
Очередь = КлиентRMQ.ОбъявитьОчередь(Канал, "", "", ТочкаОбмена, Ложь, Истина, Истина);
// auto-ack = Истина - см. пояснение выше (не как work queue в части 2)
Получатель = КлиентRMQ.СоздатьПолучателя(
Канал, Очередь.ИмяОчереди, "подписчик_1", Истина,,,, ТаймаутОжиданияСек);
Пока КлиентRMQ.СледующееСообщение(Получатель) Цикл
ДанныеСообщения = КлиентRMQ.ДанныеСообщения();
// … ЗафиксироватьПолучение / Сообщить …
КонецЦикла;
КлиентRMQ.ОтменитьПолучателя(Получатель);
КлиентRMQ.ЗакрытьКанал(Канал);
КлиентRMQ.ОтключитьсяОтСервера(Подключение);
Запуск решения
Как в части 2 сначала запускам слушателей потом отправляем сообщения иначе broadcast «на лету» не будет видно.
YAxUnit - тесты набора «Урок 3» Урок3_PubSubВОднойИБ
Консоль кода (серверный контекст) - те же сценарии:
// smoke: exchange logs U94; два temporary-подписчика U94; broadcast U94; оба получают одно тело
ОМ_ТестRMQ_Lessons.Урок3_PubSubВОднойИБ();
// или вручную: два слушателя (ФЗ / сеансы по скелету «Подписчик» выше) U94; потом emit
ОМ_ТестRMQ_Lessons.Урок3_Отправитель();
ОМ_ТестRMQ_Lessons.Урок3_Отправитель("warning: disk low");
Результат работы: в регистре rmq_ОчередьОбменаRMQ - исходящие Отправлено и входящие у обоих подписчиков. В ЖР / окне сообщений после Урок3_PubSubВОднойИБ примерно:
… | Старт | очередь: часть3_pubsub | данные: Урок3_PubSubВОднойИБ
… | Отправка в очередь | очередь: часть3_pubsub | данные: broadcast:a1b2c3d4
… | Получение из очереди | очередь: часть3_pubsub | данные: подписчик 1 | broadcast:a1b2c3d4
… | Получение из очереди | очередь: часть3_pubsub | данные: подписчик 2 | broadcast:a1b2c3d4
… | Окончание | очередь: часть3_pubsub | данные: Урок3_PubSubВОднойИБ
При ручном emit двух сообщений у каждого подписчика будет по две строки «Получение» - тела одинаковые у подписчик 1 и подписчик 2.
Результат
Настроили рассылку: fanout-exchange, временные очереди подписчиков, binding. Каждое опубликованное сообщение уходит всем слушателям.
Дальше - часть 4: слушать не всё подряд, а подмножество по ключу.
Ссылки на остальные части
- Часть 1. Hello World
- Часть 2. Work Queues - рабочие очереди
- Часть 3. Publish/Subscribe - одно сообщение многим (fanout) (эта статья)
- Часть 4. Routing - маршрутизация по ключу (direct)
- Часть 5. Topics - маршрутизация по шаблону
- Часть 6. RPC - удалённый вызов процедур
- Часть 7. Publisher Confirms - надёжная публикация
- Часть 8. Streams - Hello World
- Часть 9. Streams - отслеживание смещения
Благодарю за внимание.
Создано совместно с Cursor Grok 4.5
Вступайте в нашу телеграмм-группу Инфостарт