Руководство RabbitMQ - Часть 1. Hello World

Введение

RabbitMQ - брокер сообщений: он принимает и пересылает сообщения. Можно представить его как почтовое отделение: когда вы опускаете письмо в почтовый ящик, почтальон в итоге доставит его получателю. В этой аналогии RabbitMQ - и ящик, и отделение, и почтальон.

Отличие от настоящей почты: RabbitMQ не работает с бумагой - он принимает, хранит и пересылает двоичные блоки данных - сообщения (messages).

RabbitMQ и система обмена сообщениями в целом используют некоторый жаргон.

• Producing - просто отправка. Программа, которая отправляет сообщения, - производитель (producer):

• Очередь (queue) - «почтовый ящик» в RabbitMQ. Сообщения могут храниться только внутри очереди. Очередь ограничена памятью и диском хоста - по сути большой буфер.

Несколько производителей могут писать в одну очередь; несколько потребителей - читать из одной очереди.

• Consuming - по смыслу получение. Потребитель (consumer) в основном ждёт сообщения:

• Обменник (exchange) - точка, через которую сообщение входит в брокер. Напрямую в очередь сообщение не кладут: производитель публикует в exchange, а тот по правилам (и ключу маршрутизации) решает, в какие очереди отправить. В этой части хватит обменника по умолчанию (пустая строка в API); типы exchange разберём позже.
• Виртуальный хост (virtual host, vhost) - логическая «площадка» внутри одного RabbitMQ: свои очереди, обменники, привязки и права пользователей. По умолчанию используется vhost / (в строке подключения AMQP его пишут как %2f). Идея близка к virtual hosts в веб-сервере: один процесс - несколько изолированных пространств имён. Подробнее: документация vhosts.

Производитель, потребитель и брокер не обязаны быть на одном хосте; в реальных системах обычно так и есть. Одно приложение может быть и производителем, и потребителем.

Когда RabbitMQ уместен в 1С. Типовой обмен данными, HTTP-сервисы и планы обмена закрывают много задач «внутри» контура 1С. Брокер имеет смысл, когда нужна асинхронная развязка: отправитель не ждёт обработку, несколько независимых получателей, общий буфер при пиках нагрузки, интеграция с системами вне платформы. Если достаточно синхронного вызова между двумя базами или регламентного обмена по плану - усложнять стенд RabbitMQ обычно не стоит. В серии разбираем API КлиентRMQ и паттерны брокера; выбор «нужен ли брокер в вашем проекте» остаётся за архитектурой задачи.

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

В примерах используется компонента kovalevdmv/1CRabbitMQ: обращения через обработку КлиентRMQ в составе конфигурации.

Код примеров собран в расширении RMQ_lessons (malikov-pro/1CRabbitMQ) для запуска через YAxUnit. HTTP Management API реализован в модуле ОМ_РМКУ_ManagementAPI , использует БОНД_Коннектор. Параметры брокера в общем модуле ОМ_РМКУ_Настройки.

Сценарии тестирования можно вызывать и из консоли кода (например bsl_console) в серверном контексте.

На стенде 1С компонента запускается в клиент-серверной базе.

Установка и запуск RabbitMQ

В примерах используется я версия 4.x. Основной стенд для прогона примеров - брокер на localhost. Установка на Windows Installing on Windows, для RabbitMQ 4.3.4 нужен Erlang 27.x, после установки нужно включить плагин Management UI: rabbitmq-plugins enable rabbitmq_management (документация).

После установки и запуска сервиса откройте Management UI: http://localhost:15672. Если интерфейс открывается и вход проходит - можно идти дальше.

Учтите что localhost RMQ задается относительно сервера 1С. Если сервис установлен на другом ПК, то поменяйте настройки на нужные и учитывайте что guest/guest подключаются только с localhost.

Hello World!

Напишем сценарий на 1С: отправить одно сообщение и получить его обратно. Это «Hello World» обмена сообщениями.

На схеме «P» - производитель, «C» - потребитель, посередине - очередь hello.

Производитель пишет в очередь "hello". Потребитель читает из неё.

Ниже разбираем шаги по отдельности; собранные сценарии - в модуле ОМ_ТестRMQ_Lessons (см. «Запуск решения»).

Отправка

Первая процедура отправит одно сообщение в очередь. В расширении это ОМ_ТестRMQ_Lessons.Урок1_Отправитель - ниже разбор по шагам. Сначала соединение с RabbitMQ:

Процедура Урок1_Отправитель() Экспорт
	
	КлиентRMQ = Обработки.КлиентRMQ.Создать();
	URI = ОМ_РМКУ_Настройки.СтрокаПодключения();
	ИмяОчереди = ОМ_РМКУ_Настройки.ИмяОчередиЧасть1(); // "hello"
	ТекстСообщения = "Hello World!";
	
	Подключение = КлиентRMQ.ПодключитьсяКСерверу(URI);
	Если КлиентRMQ.ЭтоОшибка(Подключение) Тогда
		КлиентRMQ.ОтключитьсяОтСервера(Подключение);
		ВызватьИсключение "Ошибка подключения к RabbitMQ";
	КонецЕсли;
	
	Попытка
		
		Канал = КлиентRMQ.СоздатьКанал(Подключение);
		Если КлиентRMQ.ЭтоОшибка(Канал) Тогда
			ВызватьИсключение "Ошибка создания канала";
		КонецЕсли;
		
		// … объявление очереди и публикация - ниже
		
	Исключение
		КлиентRMQ.ОтключитьсяОтСервера(Подключение);
		ВызватьИсключение;
	КонецПопытки;
	
	КлиентRMQ.ОтключитьсяОтСервера(Подключение);
	
КонецПроцедуры

Мы подключены к брокеру по параметрам из СтрокаПодключения(), в URI указан vhost /.

Объявление очереди

Перед отправкой очередь должна существовать. Иначе RabbitMQ отбросит сообщение. Создадим очередь hello.

Очередь hello - именованная и общая, для таких очередей в курсе берём тип quorum - это подходит и для большинства задач интеграций с 1С. Это не универсальный тип на все случаи и сам по себе не обеспечивает высокую доступность: на одном брокере quorum переживает рестарт процесса (данные на диске), но не даёт отказоустойчивости кластера. Временные «личные» очереди (classic + exclusive) появятся в части 3, stream - в части 9.

В расширении объявление очереди вынесено в хелпер ОбъявитьОчередь (durable + quorum):

ИмяОчереди = ОМ_РМКУ_Настройки.ИмяОчередиЧасть1(); // "hello"
Очередь = ОМ_РМКУ_Настройки.ОбъявитьОчередь(КлиентRMQ, Канал, ИмяОчереди);

Эквивалент «вручную» через API компоненты:

Аргументы = Новый Массив;
Аргументы.Добавить(КлиентRMQ.НовыйАргумент("x-queue-type", "quorum", "строка"));

Очередь = КлиентRMQ.ОбъявитьОчередь(
	Канал, ИмяОчереди, "", "",
	Истина, // durable
	Ложь,   // exclusive
	Ложь,   // auto-delete
	Аргументы);

ОбъявитьОчередь идемпотентен при тех же параметрах. Если очередь уже есть с другими параметрами (например, раньше создали classic) - брокер вернёт ошибку. Удалите её через HTTP Management API (или через UI) и объявите снова:

ОМ_РМКУ_ManagementAPI.УдалитьОчередь(ИмяОчереди);
Очередь = КлиентRMQ.ОбъявитьОчередь(
	Канал, ИмяОчереди, "", "",
	Истина, Ложь, Ложь, Аргументы);

Публикация

Сообщение всегда проходит через обменник. Пока достаточно обменника по умолчанию, поэтому передаём пустую строку в ТочкаОбмена: брокер кладёт сообщение в очередь, имя которой совпадает с ключом маршрутизации (routing_key, здесь это ИмяОчереди). Типы exchange - в части 3 (Publish/Subscribe, fanout), маршрутизация по ключу - в части 4 (direct), шаблоны ключей - в части 5 (topic).

ТекстСообщения = "Hello World!";
ИмяОчереди = "hello"; // или ОМ_РМКУ_Настройки.ИмяОчередиЧасть1()
ТочкаОбмена = "";     // обменник по умолчанию; routing_key = имя очереди
Исполнитель = "Урок1_Отправитель";

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

Если КлиентRMQ.ЭтоОшибка(Ответ) Тогда
	КлиентRMQ.ОтключитьсяОтСервера(Подключение);
	ВызватьИсключение СтрШаблон("Ошибка отправки: %1", Ответ.Текст);
КонецЕсли;

ИдИсходящего = ОМ_РМКУ_Настройки.ЗафиксироватьОтправку(ИмяОчереди, ТекстСообщения, Исполнитель);

После публикации сообщение лежит в очереди hello. Пока потребитель ещё не забрал его, это видно в Management UI (Queues → hello → Ready / Total) или через HTTP Management API - об этом ниже, перед запуском решения.

Закрытие соединения

По документации КлиентRMQ соединение нужно всегда закрывать через ОтключитьсяОтСервера - иначе накапливаются неиспользуемые подключения (утечка ресурсов). В примерах этого части делаем «одна операция - одно подключение»: подключились → работа → отключились. В дальнейшем соединение можно переиспользовать между вызовами, закрывая канал (ЗакрытьКанал).

КлиентRMQ.ОтключитьсяОтСервера(Подключение);

Получение

Вторая процедура получает сообщения из очереди и выводит их. В расширении это ОМ_ТестRMQ_Lessons.Урок1_Получатель.

Подключение и канал - как у отправителя. Очередь снова объявляем: мы не знаем, что запустят раньше, поэтому хорошая практика объявлять очередь в обеих программах.

ИмяОчереди = ОМ_РМКУ_Настройки.ИмяОчередиЧасть1();
Очередь = ОМ_РМКУ_Настройки.ОбъявитьОчередь(КлиентRMQ, Канал, ИмяОчереди);

Получение сообщений

Для «Hello World» достаточно разового чтения - ОдноСледующееСообщение: есть сообщение в очереди - вернёт Истина, нет - Ложь (без постоянной подписки и цикла ожидания). Долгоживущий consumer (СоздатьПолучателя + СледующееСообщение) понадобится в части 2 и дальше.

Исполнитель = "Урок1_Получатель";

Если КлиентRMQ.ОдноСледующееСообщение(Канал, ИмяОчереди) Тогда
	
	ДанныеСообщения = КлиентRMQ.ДанныеСообщения();
	
	Если КлиентRMQ.ЭтоОшибка(ДанныеСообщения) Тогда
		ОМ_РМКУ_Настройки.СообщитьКонтрольнуюТочку("Ошибка", ДанныеСообщения.Текст, ИмяОчереди);
	Иначе
		ИдВходящего = ОМ_РМКУ_Настройки.ЗафиксироватьПолучение(
			ИмяОчереди,
			ДанныеСообщения.Данные,
			Исполнитель,
			ДанныеСообщения.Тег);
		// после разового чтения сообщение нужно подтвердить
		КлиентRMQ.ПодтвердитьСообщение(Канал, ДанныеСообщения.Тег);
		ОМ_РМКУ_Настройки.ЗафиксироватьПодтверждение(ИмяОчереди, ИдВходящего);
	КонецЕсли;
	
Иначе
	
	ДанныеСообщения = КлиентRMQ.ДанныеСообщения();
	Если КлиентRMQ.ЭтоОшибка(ДанныеСообщения) Тогда
		ОМ_РМКУ_Настройки.СообщитьКонтрольнуюТочку("Ошибка", ДанныеСообщения.Текст, ИмяОчереди);
	КонецЕсли;
	
КонецЕсли;

Закрываем соединение:

КлиентRMQ.ОтключитьсяОтСервера(Подключение);

История операций и лог сценария

В расширении RMQ_lessons история обмена пишется в РС rmq_ОчередьОбменаRMQ, контрольные точки сценария - в ЖР и окно сообщений (событие КлиентRMQ). Реализация - в ОМ_РМКУ_Настройки. Формат строки лога: дата-время | операция | очередь: … | данные: ….

RabbitMQ Management API

Порт 15672 отдаёт не только веб-UI, но и HTTP API плагина rabbitmq_management: REST + JSON, Basic Auth той же учётной записью, что для UI. Management API не замена AMQP, а нужна для наблюдения и администрирования топологии: списки очередей и обменников, счётчики сообщений, создание exchange/queue/bind, policies. В расширении работа реализована в общем модуле ОМ_РМКУ_ManagementAPI

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

Сначала отправьте сообщение - оно останется в очереди hello (проверка через Management API / UI), затем заберите получателем (можно из другой информационной базы).

YAxUnit - тесты набора «Урок 1» (модуль ОМ_ТестRMQ_Lessons):

• ОМ_ТестRMQ_Lessons.Урок1_ОбменВОднойИБ - publish → HTTP-проверка hello → consume
• ОМ_ТестRMQ_Lessons.Урок1_Отправитель - публикация (+ HTTP: очередь в списке, messages ≥ 1)
• ОМ_ТестRMQ_Lessons.Урок1_Получатель - получение

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

ОМ_ТестRMQ_Lessons.Урок1_Отправитель(); // публикация + проверка Management API
ОМ_ТестRMQ_Lessons.Урок1_Получатель();  // получение
// или одной процедурой:
ОМ_ТестRMQ_Lessons.Урок1_ОбменВОднойИБ();

Результат работы: В регистре rmq_ОчередьОбменаRMQ - исходящая запись Отправлено и входящая Подтверждено. В ЖР / окне сообщений (событие КлиентRMQ) после успешного Урок1_ОбменВОднойИБ примерно такой лог:

… | Старт | очередь: hello | данные: Урок1_ОбменВОднойИБ
… | Отправка в очередь | очередь: hello | данные: Hello World!
… | Проверка очереди | очередь: hello | данные: да
… | Количество сообщений в очереди | очередь: hello | данные: 1
… | Получение из очереди | очередь: hello | данные: Hello World!
… | Окончание | очередь: hello | данные: Урок1_ОбменВОднойИБ

Результат

Мы научились отправлять и получать сообщение из именованной очереди. Дальше - часть 2: простая рабочая очередь (work queue).

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

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