Руководство RabbitMQ - Часть 5. Topics - маршрутизация по шаблону

Подготовка окружения

Компонента kovalevdmv/1CRabbitMQ, обработка КлиентRMQ. Брокер - как в части 1.

Код примеров собран в расширении RMQ_lessons (malikov-pro/1CRabbitMQ) для запуска через YAxUnit.

Те же сценарии можно вызывать из консоли кода в серверном контексте - например bsl_console и аналоги.

На чём сосредоточена часть

В части 4 direct фильтровал по точному ключу (error, info). Нельзя было сказать: «все логи от kern» и одновременно «все critical, откуда угодно».

Для нескольких измерений в ключе (как источник и уровень важности в syslog) нужен exchange типа topic.

Topic exchange: слова, * и #

Routing key в topic - не произвольная строка, а слова через точку: kern.critical, stock.usd.nyse, quick.orange.rabbit (до 255 байт). Binding key - в той же форме, плюс два спецсимвола:

• * (звезда) - ровно одно слово;
• # (решётка) - ноль или больше слов.

Классический пример оригинала - ключ из трёх слов <скорость>.<цвет>.<вид>:

quick.orange.rabbit попадёт в очереди с .orange. и ..rabbit. quick.brown.fox - никуда. lazy.orange.new.rabbit (четыре слова) совпадёт с lazy.#.

> Topic умеет притворяться другими типами: binding # W76; fanout; bindings без */# W76; direct.

Тип headers здесь не разбираем - для логгера достаточно topic. Кратко: маршрутизация идёт не по routing key, а по заголовкам сообщения (таблица ключU94;значение в свойствах AMQP). Привязка задаёт набор заголовков и правило x-match: all (должны совпасть все) или any (достаточно одного). Именованный headers-exchange создаётся так же, как остальные - ОМ_РМКУ_ManagementAPI.ОбъявитьExchange("имя", "headers"), на брокере есть и встроенный amq.headers. Публикация в КлиентRMQ - через параметр Заголовки у ОпубликоватьСообщение`. Подробнее: Headers exchange.

Логгер: <источник>.<уровень>

Рабочее допущение руководства: ключ из двух слов (в оригинале facility / severity). Exchange - topic_logs, объявляем через ОМ_РМКУ_ManagementAPI.ОбъявитьExchange или через Management UI.

ТочкаОбмена = ОМ_РМКУ_Настройки.ИмяОбменникаЧасть5(); // "topic_logs"
ОМ_РМКУ_ManagementAPI.ОбъявитьExchange(ТочкаОбмена, "topic");

Публикация:

// Подключение и Канал - как в [части 1](//infostart.ru/1c/articles/2196530/)
// Exchange «topic_logs» уже объявлен через Management API (тип topic)
ТочкаОбмена = "topic_logs";
КлючМаршрутизации = "kern.critical";
ТекстСообщения = "A critical kernel error";

КлиентRMQ.ОпубликоватьСообщение(
	Канал, ТекстСообщения, КлючМаршрутизации, ТочкаОбмена, Истина);

Подписка - temporary-очередь + одна или несколько привязок с шаблонами:

ТочкаОбмена = "topic_logs";
ОМ_РМКУ_ManagementAPI.ОбъявитьExchange(ТочкаОбмена, "topic");

// все логи
ОчередьВсе = КлиентRMQ.ОбъявитьОчередь(
	Канал, "", "#", ТочкаОбмена, Ложь, Истина, Истина);

// только kern.*
ОчередьKern = КлиентRMQ.ОбъявитьОчередь(
	Канал, "", "kern.*", ТочкаОбмена, Ложь, Истина, Истина);

// kern.* и *.critical на одну очередь - повторный ОбъявитьОчередь
Очередь = КлиентRMQ.ОбъявитьОчередь(
	Канал, "", "kern.*", ТочкаОбмена, Ложь, Истина, Истина);
КлиентRMQ.ОбъявитьОчередь(
	Канал, Очередь.ИмяОчереди, "*.critical", ТочкаОбмена, Ложь, Истина, Истина);

Собираем вместе

Скелет подписчика ниже; связный прогон и лог - в «Запуск решения». Код - ОМ_ТестRMQ_Lessons.

Как в части 3: АвтоматическоеПодтверждение = Истина у логгера уместен (своя очередь + копия broadcast); для work queue в части 2 нужен ручной ack.

ТочкаОбмена = "topic_logs";
ОМ_РМКУ_ManagementAPI.ОбъявитьExchange(ТочкаОбмена, "topic");
Шаблоны = СтрРазделить("kern.*,*.critical", ",");

Очередь = Неопределено;
Для Каждого Шаблон Из Шаблоны Цикл
	Если Очередь = Неопределено Тогда
		Очередь = КлиентRMQ.ОбъявитьОчередь(
			Канал, "", Шаблон, ТочкаОбмена, Ложь, Истина, Истина);
	Иначе
		КлиентRMQ.ОбъявитьОчередь(
			Канал, Очередь.ИмяОчереди, Шаблон, ТочкаОбмена, Ложь, Истина, Истина);
	КонецЕсли;
КонецЦикла;

Получатель = КлиентRMQ.СоздатьПолучателя(
	Канал, Очередь.ИмяОчереди, "подписчик_topic", Истина,,,, ТаймаутОжиданияСек);

Пока КлиентRMQ.СледующееСообщение(Получатель) Цикл
	ДанныеСообщения = КлиентRMQ.ДанныеСообщения();
	// КлючМаршрутизации + тело
КонецЦикла;

Поэкспериментируйте с ключами длиннее двух слов - код ничего жёстко не предполагает про формат.

Запуск решения

Как в части 4: при ручном демо сначала слушатели с нужными шаблонами binding, потом отправка сообщений.

YAxUnit - тесты набора «Урок 5» Урок5_TopicsВОднойИБ

Консоль кода (серверный контекст) - те же сценарии:

// smoke: topic_logs U94; kern.* + *.critical U94; emit kern.info / kern.critical / app.critical
ОМ_ТестRMQ_Lessons.Урок5_TopicsВОднойИБ();

// или вручную: слушатели (ФЗ / сеансы по скелету «Подписчик» выше, разные шаблоны)
// U94; потом emit по скелету «Издатель» (routing key = <источник>.<уровень>)

Результат работы: в регистре rmq_ОчередьОбменаRMQ - исходящие Отправлено и входящие у подписчиков. Журнал сценария - с логическим именем часть5_topics. В ЖР / окне сообщений после Урок5_TopicsВОднойИБ примерно:

… | Старт | очередь: часть5_topics | данные: Урок5_TopicsВОднойИБ
… | Отправка в очередь | очередь: часть5_topics | данные: kern.info | kern.info:a1b2c3d4
… | Отправка в очередь | очередь: часть5_topics | данные: kern.critical | kern.critical:a1b2c3d4
… | Отправка в очередь | очередь: часть5_topics | данные: app.critical | app.critical:a1b2c3d4
… | Получение из очереди | очередь: часть5_topics | данные: подписчик kern | kern.info:a1b2c3d4
… | Получение из очереди | очередь: часть5_topics | данные: подписчик kern | kern.critical:a1b2c3d4
… | Получение из очереди | очередь: часть5_topics | данные: подписчик critical | kern.critical:a1b2c3d4
… | Получение из очереди | очередь: часть5_topics | данные: подписчик critical | app.critical:a1b2c3d4
… | Окончание | очередь: часть5_topics | данные: Урок5_TopicsВОднойИБ

У подписчик kern нет app.critical. У подписчик critical нет kern.info. Порядок двух строк внутри одной метки может чуть съехать - важнее набор тел по меткам.

Результат

Настроили гибкую маршрутизацию по шаблону: topic-exchange, * и # в binding key.

Дальше - часть 6: запрос–ответ (RPC) и свойства сообщения (reply_to, correlation_id).

Ссылки на остальные части

• Часть 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