Руководство RabbitMQ - Часть 6. RPC - удалённый вызов процедур

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

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

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

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

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

В части 2 задачи уходили исполнителям «в одну сторону». Здесь другой паттерн: вызвать работу на удалённой стороне и дождаться результата - Remote Procedure Call (RPC).

Учебный сервис - числа Фибоначчи: клиент шлёт n, сервер отвечает значением fib(n). Реальной тяжести нет; важна схема запрос–ответ.

> Когда RPC уместен. Легко перепутать локальный вызов с медленным удалённым - система становится непредсказуемой. Делайте удалённость очевидной, документируйте зависимости, думайте об ошибках и таймаутах. Если сомневаетесь - асинхронный конвейер (как work queue) часто проще сопровождать, чем блокирующий RPC.

Свойства сообщения и заголовки

Протокол AMQP задаёт набор свойств, которые едут вместе с телом. В КлиентRMQ их собирает НовыеСвойстваСообщения(). Для RPC важны:

Persistent delivery (СохранитьСообщениеНаДиске) вы уже видели в части 2 - это не поле структуры свойств, а отдельный параметр публикации.

Пользовательские заголовки - через ДобавитьЗаголовок; на приёме - ДанныеСообщения.Заголовки. Их можно использовать для трассировки (например x-source = 1C), не смешивая с correlation_id / reply_to / message_id.

Для журнала обмена (rmq_ОчередьОбменаRMQ) кладите один UUID в message_id: при отправке и при получении ЗафиксироватьОтправку / ЗафиксироватьПолучение берут его как Идентификатор записи - исходящий и входящий потоки получат один ключ.

Корреляция = Строка(Новый УникальныйИдентификатор);
ИдСообщения = ОМ_РМКУ_Настройки.НовыйИдентификаторСообщения();
// content_type, encoding, priority, correlation_id, reply_to, expiration, message_id
Свойства = КлиентRMQ.НовыеСвойстваСообщения(
	"", "", "", Корреляция, ИмяОчередиОтвета, "", ИдСообщения);
Заголовки = КлиентRMQ.ДобавитьЗаголовок("x-source", "Урок6_Клиент");

КлиентRMQ.ОпубликоватьСообщение(
	Канал, "4", "rpc_queue", "", Истина, Ложь, Свойства, Заголовки);
ОМ_РМКУ_Настройки.ЗафиксироватьОтправку("rpc_queue", "4", "Клиент", Заголовки, Свойства);

Очередь ответа и correlation_id

Клиент создаёт exclusive callback-очередь (серверное имя). В запросе указывает её в reply_to и уникальный correlation_id.

Одна callback-очередь на клиента эффективнее, чем новая на каждый вызов. Тогда по correlation_id отличаем ответы: чужой или повторный - пропускаем (сервер мог переобработать запрос после сбоя до ack).

Итог потока:

1. Клиент объявляет callback-очередь.
2. Публикует в rpc_queue с reply_to + correlation_id.
3. Сервер читает rpc_queue, считает результат, публикует в очередь из reply_to с тем же correlation_id.
4. Клиент читает callback-очередь, пока не найдёт совпадение по correlation_id.

Сервер (исполнитель RPC)

Очередь запросов - именованная, durable (как рабочие очереди частей). Предзагрузка = 1, если серверов несколько. Ручной ack после отправки ответа.

Канал = КлиентRMQ.СоздатьКанал(Подключение, Ложь, 1); // prefetch = 1
ИмяОчереди = "rpc_queue";
Очередь = ОМ_РМКУ_Настройки.ОбъявитьОчередь(КлиентRMQ, Канал, ИмяОчереди);

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

Пока КлиентRMQ.СледующееСообщение(Получатель) Цикл
	
	Запрос = КлиентRMQ.ДанныеСообщения();
	N = Число(Запрос.Данные);
	Результат = Фибоначчи(N); // учебная рекурсия - только маленькие N
	
	Корреляция = Запрос.СвойстваСообщения.correlation_id;
	// content_type, encoding, priority, correlation_id, reply_to, …
	ОтветныеСвойства = КлиентRMQ.НовыеСвойстваСообщения("", "", "", Корреляция, "");
	
	КлиентRMQ.ОпубликоватьСообщение(
		Канал,
		Строка(Результат),
		Запрос.СвойстваСообщения.reply_to, // routing key = имя callback-очереди
		"",                                 // default exchange
		Истина,
		Ложь,
		ОтветныеСвойства);
	
	КлиентRMQ.ПодтвердитьСообщение(Канал, Запрос.Тег);
	
КонецЦикла;

Клиент

ОчередьОтветов = КлиентRMQ.ОбъявитьОчередь(Канал, "", "", "", Ложь, Истина, Истина);
ИмяОчередиОтвета = ОчередьОтветов.ИмяОчереди;
Корреляция = Строка(Новый УникальныйИдентификатор);
ИдСообщения = ОМ_РМКУ_Настройки.НовыйИдентификаторСообщения();
// content_type, encoding, priority, correlation_id, reply_to, expiration, message_id
Свойства = КлиентRMQ.НовыеСвойстваСообщения(
	"", "", "", Корреляция, ИмяОчередиОтвета, "", ИдСообщения);
Заголовки = КлиентRMQ.ДобавитьЗаголовок("x-source", "Урок6_Клиент");

КлиентRMQ.ОпубликоватьСообщение(
	Канал, "4", "rpc_queue", "", Истина, Ложь, Свойства, Заголовки);

// idle-таймаут получателя - иначе цикл «сервер молчит» зависнет (как в Урок6_Вызвать)
ТаймаутОтветаСек = 25;
ПолучательОтвета = КлиентRMQ.СоздатьПолучателя(
	Канал, ИмяОчередиОтвета, "rpc_client", Истина,,,, ТаймаутОтветаСек);

Результат = Неопределено;
Пока КлиентRMQ.СледующееСообщение(ПолучательОтвета) Цикл
	Ответ = КлиентRMQ.ДанныеСообщения();
	Если Ответ.СвойстваСообщения.correlation_id = Корреляция Тогда
		Результат = Число(Ответ.Данные); // fib(4) = 3
		Прервать;
	КонецЕсли;
КонецЦикла;

Если Результат = Неопределено Тогда
	ВызватьИсключение "Таймаут ожидания RPC-ответа";
КонецЕсли;

КлиентRMQ.ОтменитьПолучателя(ПолучательОтвета);

Полный код клиента/сервера - в расширении (Урок6_Вызвать / Урок6_RpcВОднойИБ).

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

При ручном демо сначала RPC-сервер (слушает rpc_queue), потом клиент - иначе запрос останется в очереди без ответа.

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

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

// smoke: rpc_queue + reply-очередь U94; fib(6) U94; ответ 8 (одна ИБ, без ФЗ)
ОМ_ТестRMQ_Lessons.Урок6_RpcВОднойИБ();

// или вручную: сначала сервер в ФЗ, потом клиент
ПараметрыЗадания = Новый Массив;
ПараметрыЗадания.Добавить(0); // лимит жизни, сек; 0 = из настройки
ФоновыеЗадания.Выполнить("ОМ_ТестRMQ_Lessons.Урок6_СерверВФоне", ПараметрыЗадания);
Результат = ОМ_ТестRMQ_Lessons.Урок6_Вызвать(6); // U94; 8

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

… | Старт | очередь: rpc_queue | данные: Урок6_RpcВОднойИБ | fib(6)
… | Отправка в очередь | очередь: rpc_queue | данные: запрос fib(6)
… | Получение из очереди | очередь: rpc_queue | данные: сервер | fib(6)
… | Отправка в очередь | очередь: rpc_queue | данные: ответ fib=8
… | Получение из очереди | очередь: rpc_queue | данные: клиент | результат=8
… | Окончание | очередь: rpc_queue | данные: Урок6_RpcВОднойИБ | результат=8

В ответе - число Фибоначчи; в свойствах ответа - тот же correlation_id, что у запроса.

Результат

Собрали RPC поверх RabbitMQ: callback-очередь, reply_to / correlation_id, свойства и заголовки сообщения в КлиентRMQ.

Дальше - часть 7: подтверждение, что брокер принял публикацию (Publisher Confirms).

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

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