Руководство RabbitMQ - Часть 4. Routing - маршрутизация по ключу (direct)

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

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

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

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

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

В части 3 fanout слал все логи всем подписчикам. Здесь добавим выбор: подписка только на подмножество. Например, критические ошибки - в «файл» (отдельный исполнитель), на «консоль» - все уровни.

Binding key и direct exchange

Привязка (binding) - связь очереди с типом exchange. У неё может быть binding key - в КлиентRMQ это параметр КлючМаршрутизации у ОбъявитьОчередь.

У fanout binding key игнорируется. У direct алгоритм простой: сообщение попадает в те очереди, у которых binding key точно совпадает с routing key публикации.

Как на схеме выше: у QS21; (файл) binding key error, у QS22; (консоль) - info, warning и error. Сообщение с routing key error → в обе очереди; info / warning → только QS22;; любой другой ключ - отброшен.

Несколько очередей могут иметь один и тот же binding key - тогда direct ведёт себя как fanout для этого ключа.

Система логов: уровень важности как ключ

Вместо logs (fanout) берём очередь direct_logs типа direct. Именованный exchange объявляем через ОМ_РМКУ_ManagementAPI.ОбъявитьExchange (Management HTTP API) или через Management UI.

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

При публикации routing key = уровень важности: info, warning, error.

ТочкаОбмена = "direct_logs";
Уровень = "error";
ТекстСообщения = "Run. Run. Or it will explode.";

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

Подписчик объявляет temporary-очередь и для каждого интересующего уровня - отдельную привязку (повторный ОбъявитьОчередь с тем же именем очереди и другим ключом):

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

// ещё уровни на ту же очередь
КлиентRMQ.ОбъявитьОчередь(Канал, ИмяОчереди, "warning", ТочкаОбмена, Ложь, Истина, Истина);
// info - по желанию

Только error: одна привязка с ключом error. Все уровни - три привязки (info, warning, error).

Реализация

Издатель:

ТочкаОбмена = "direct_logs";
ОМ_РМКУ_ManagementAPI.ОбъявитьExchange(ТочкаОбмена, "direct");
КлиентRMQ.ОпубликоватьСообщение(Канал, ТекстСообщения, Уровень, ТочкаОбмена, Истина);

Подписчик (выбранные уровни):

Как в части 3: у подписчика лога АвтоматическоеПодтверждение = Истина - своя temporary-очередь и копия сообщения, не общая задача work queue (часть 2).

ТочкаОбмена = "direct_logs";
ОМ_РМКУ_ManagementAPI.ОбъявитьExchange(ТочкаОбмена, "direct");
Уровни = СтрРазделить("warning,error", ",");

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

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

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

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

Как в части 3: при ручном демо сначала слушатели с нужными binding key, потом emit.

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

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

// smoke: direct_logs U94; подписчик только error + подписчик все уровни U94; emit error/info
ОМ_ТестRMQ_Lessons.Урок4_RoutingВОднойИБ();

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

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

… | Старт | очередь: часть4_routing | данные: Урок4_RoutingВОднойИБ
… | Отправка в очередь | очередь: часть4_routing | данные: error | error:a1b2c3d4
… | Отправка в очередь | очередь: часть4_routing | данные: info | info:a1b2c3d4
… | Получение из очереди | очередь: часть4_routing | данные: подписчик error | error:a1b2c3d4
… | Получение из очереди | очередь: часть4_routing | данные: подписчик все | error:a1b2c3d4
… | Получение из очереди | очередь: часть4_routing | данные: подписчик все | info:a1b2c3d4
… | Окончание | очередь: часть4_routing | данные: Урок4_RoutingВОднойИБ

У подписчик error - одна строка (только error). У подписчик все - две; порядок error/info внутри «все» может поменяться - важнее набор тел, не порядок строк.

Результат

Перешли с «всем всё» на выборочную доставку: direct-exchange и точное совпадение binding key ↔ routing key.

Дальше - часть 5: шаблоны ключей (* / #), exchange topic.

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

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