Руководство RabbitMQ - Часть 2. Work Queues - рабочие очереди

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

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

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

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

Для имитации работы исполнителя использую ВызватьПаузу() - работает с версии платформы 8.3.25

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

В части 1 мы отправляли и получали сообщение из именованной очереди. Здесь сделаем рабочую очередь (Work Queue) - она распределяет трудоёмкие задачи между несколькими исполнителями (workers).

Идея простых очередей задач (Task Queues) такая: не делать тяжёлую работу сразу и не ждать окончания выполнения. Задачу планируют к выполнению: упаковывают в сообщение и кладут в очередь. Фоновый исполнитель забирает сообщения и выполняет работу. Если исполнителей несколько - нагрузка делится между ними. Так не приходится «морозить» пользователя, пока идёт долгий расчёт или обработка.

В части 1 слали «Hello World!». Теперь - строки, которые обозначают сложные задачи. Реальной работы (изменение размера картинок, печать PDF) нет: притворимся, что заняты, через ВызватьПаузу. Сложность - число точек в строке (до 10): каждая точка - 0,5 с «работы». Например, Задача 2.......... займёт пять секунд, а Задача 3. - полсекунды.

Исполнителей в 1С можно запускать поднять двумя способами: отдельными клиентами или несколько фоновых заданий в одной ИБ, в примере используются несколько фоновых.

Подключение и канал к брокеру

Два уровня API, которые легко смешать:

• Подключение (ПодключитьсяКСерверу) - сессия клиента к брокеру (TCP / AMQP): на ней учётные данные и vhost. Параметры - из ОМ_РМКУ_Настройки.СтрокаПодключения().
• Канал (СоздатьКанал) - логический канал внутри подключения. На нём объявляют очереди, публикуют и потребляют. Настройки вроде предзагрузки (сколько неподтверждённых сообщений держать у получателя) задают на канале, а не на подключении.

Типичный цикл: подключились U94; создали канал U94; работа U94; при необходимости ЗакрытьКанал U94; обязательно ОтключитьсяОтСервера.

Несколько каналов на одном подключении удобны, когда нужны разные настройки (у исполнителя - предзагрузка = 1) или изоляция ошибок канала. Предзагрузку для рабочей очереди разберём после подтверждений.

Отправка задач

Публикуем несколько сообщений (не одно «Hello World!»). Число точек в теле - «вес» задачи (1..10): каждая точка - 0,5 с паузы у исполнителя. Веса задаю массивом, рядом с тяжёлыми кладу несколько лёгких, чтобы в логе было видно - пока один исполнитель тянет задачу на 8-10 точек, второй успевает обработать две-три коротких.

В расширении веса - ОМ_РМКУ_Настройки.ВесаЗадачЧасть2(): 1, 10, 1, 1, 2, 8, 1.

Очередь task_queue и долговечность

Очередь hello из части 1 уже могла быть объявлена с другими параметрами - брокер не даст «тихо» поменять объявление. Для рабочей очереди берём новое имя task_queue: durable + quorum через хелпер из части 1.

Сообщения помечаем как постоянные (СохранитьСообщениеНаДиске = Истина): при рестарте брокера задача с большей вероятностью сохранится. Полной гарантии нет (есть окно до записи на диск), более жёсктий вариант описан в подтверждение публикаций.

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

ВесаЗадач = ОМ_РМКУ_Настройки.ВесаЗадачЧасть2(); // 1, 10, 1, 1, 2, 8, 1
Номер = 0;
Для Каждого Вес Из ВесаЗадач Цикл
	
	Номер = Номер + 1;
	ТекстСообщения = "Задача " + Номер + Лев("..........", Вес);
	СохранитьСообщениеНаДиске = Истина; // persistent
	
	Ответ = КлиентRMQ.ОпубликоватьСообщение(
		Канал, ТекстСообщения, ИмяОчереди, "", Истина, СохранитьСообщениеНаДиске);
	
	Если КлиентRMQ.ЭтоОшибка(Ответ) Тогда
		ВызватьИсключение СтрШаблон("Ошибка отправки: %1", Ответ.Текст);
	КонецЕсли;
	
	// журнал обмена - РС rmq_ОчередьОбменаRMQ 
	ОМ_РМКУ_Настройки.ЗафиксироватьОтправку(ИмяОчереди, ТекстСообщения, "Урок2_Отправитель");
	ОМ_РМКУ_Настройки.СообщитьКонтрольнуюТочку("Отправка в очередь", ТекстСообщения, ИмяОчереди);
	
КонецЦикла;

В расширении это ОМ_ТестRMQ_Lessons.Урок2_Отправитель.

Циклическая диспетчеризация (round-robin)

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

По умолчанию RabbitMQ отдаёт каждое следующее сообщение следующему получателю по кругу (round-robin): в среднем поровну.

Подтверждение сообщения

Задача у исполнителя может идти несколько секунд. Если брокер считает сообщение обработанным сразу после выдачи, а исполнитель упадёт посередине - задача потеряна. Мы хотим другое: упал исполнитель - задача вернулась в очередь и ушла другому.

В части 1 параметр АвтоматическоеПодтверждение у СоздатьПолучателя (в AMQP это no_ack) часто оставляют Истина: брокер удаляет сообщение сразу после доставки. Для рабочей очереди так недопустимо.

АвтоматическоеПодтверждение = Ложь;

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

// … обработка задачи …

Ответ = КлиентRMQ.ПодтвердитьСообщение(Канал, ДанныеСообщения.Тег);

> Забытое подтверждение. Сообщения, выданные исполнителю, но ещё не подтверждённые копятся увеличивая расход памяти брокера. Проверяйте через Management UI счётчики готовых к выдаче и неподтверждённых. Таймаут подтверждения у брокера по умолчанию ~30 мин - Delivery Acknowledgement Timeout.

Справедливая раздача (предзагрузка)

Round-robin всё ещё может быть несправедливым. Два исполнителя: все нечётные задачи тяжёлые, чётные - лёгкие. Один постоянно занят, второй почти простаивает. Брокер об «весе» не знает и по кругу отдаёт каждое следующее сообщение следующему получателю - вслепую.

Когда подтверждение ручное, можно ограничить предзагрузку: сколько неподтверждённых сообщений брокер держит у одного получателя. В КлиентRMQ это третий параметр СоздатьКанал - КоличествоПредзагруженныхНеПодтвержденныхСообщений.

При значении 1 следующее сообщение не отдадут, пока не подтверждено текущее и свободный исполнитель получит сообщение раньше занятого. При использовании АвтоматическоеПодтверждение = Истина параметр предзагрузки не влияет на распределение.

КоличествоПредзагруженныхНеПодтвержденныхСообщений = 1;
Канал = КлиентRMQ.СоздатьКанал(
	Подключение, Ложь, КоличествоПредзагруженныхНеПодтвержденныхСообщений);

Собираем исполнителя вместе

Полный код - в расширении: ОМ_ТестRMQ_Lessons.Урок2_ИсполнительВФоне. Здесь - скелет того, что собрали выше.

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

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

Пока КлиентRMQ.СледующееСообщение(Получатель) Цикл
	
	ДанныеСообщения = КлиентRMQ.ДанныеСообщения();
	ТекстСообщения = Строка(ДанныеСообщения.Данные);
	
	ЧислоТочек = СтрЧислоВхождений(ТекстСообщения, ".");
	ВызватьПаузу(ЧислоТочек * ПаузаНаТочкуМс); // «работа»
	
	КлиентRMQ.ПодтвердитьСообщение(Канал, ДанныеСообщения.Тег);
	ОМ_РМКУ_Настройки.СообщитьКонтрольнуюТочку(
		"Получение из очереди",
		СтрШаблон("%1 | %2", ПрефиксИсполнителя, ТекстСообщения), // «исполнитель N | Задача …»
		ИмяОчереди);
	
КонецЦикла;

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

С ручным подтверждением и предзагрузкой = 1 получается рабочая очередь.

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

Сначала запускаем исполнителей с прослушиванием очереди после публикуем пакет сообщений, это нужно чтобы фоновые задания исполнителей не закончили работу до запуска отправителя и продемонстрировать разбор "на лету".

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

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

// round-robin с паузой (ФЗ + ВызватьПаузу), одной процедурой:
ОМ_ТестRMQ_Lessons.Урок2_RoundRobinВОднойИБ();

// или вручную
Для Номер = 1 По ОМ_РМКУ_Настройки.КоличествоИсполнителейЧасть2() Цикл // по умолчанию 2
	ПараметрыЗадания = Новый Массив;
	ПараметрыЗадания.Добавить(Номер); // U94; метка «исполнитель N» в РС / ЖР
	ПараметрыЗадания.Добавить(0);     // лимит жизни, сек; 0 = из настройки
	ФоновыеЗадания.Выполнить(
		"ОМ_ТестRMQ_Lessons.Урок2_ИсполнительВФоне",
		ПараметрыЗадания);
КонецЦикла;

// когда слушатели уже в очереди - публикуем пакет
ОМ_ТестRMQ_Lessons.Урок2_Отправитель();

Результат работы: В регистре rmq_ОчередьОбменаRMQ - исходящие записи Отправлено и входящие Подтверждено. В ЖР / окне сообщений после round-robin примерно такой лог - у продюсера строка на каждую задачу, у каждого исполнителя - после ack:

05.08.2026 19:16:36 | Старт | очередь: task_queue | данные: Урок2_Отправитель
05.08.2026 19:16:35 | Старт | очередь: task_queue | данные: Урок2_RoundRobinВОднойИБ | исполнителей=2 | задач=7 | лимитФЗ=40 с
05.08.2026 19:16:36 | Отправка в очередь | очередь: task_queue | данные: Задача 1.
05.08.2026 19:16:36 | Отправка в очередь | очередь: task_queue | данные: Задача 2..........
05.08.2026 19:16:36 | Отправка в очередь | очередь: task_queue | данные: Задача 3.
05.08.2026 19:16:36 | Отправка в очередь | очередь: task_queue | данные: Задача 4.
05.08.2026 19:16:36 | Отправка в очередь | очередь: task_queue | данные: Задача 5..
05.08.2026 19:16:36 | Отправка в очередь | очередь: task_queue | данные: Задача 6........
05.08.2026 19:16:36 | Отправка в очередь | очередь: task_queue | данные: Задача 7.
05.08.2026 19:16:36 | Окончание | очередь: task_queue | данные: Урок2_Отправитель
05.08.2026 19:16:36 | Получение из очереди | очередь: task_queue | данные: исполнитель 2 | Задача 1.
05.08.2026 19:16:36 | Старт | очередь: task_queue | данные: исполнитель 1
05.08.2026 19:16:36 | Старт | очередь: task_queue | данные: исполнитель 2
05.08.2026 19:16:37 | Получение из очереди | очередь: task_queue | данные: исполнитель 2 | Задача 3.
05.08.2026 19:16:37 | Получение из очереди | очередь: task_queue | данные: исполнитель 2 | Задача 4.
05.08.2026 19:16:38 | Получение из очереди | очередь: task_queue | данные: исполнитель 2 | Задача 5..
05.08.2026 19:16:41 | Получение из очереди | очередь: task_queue | данные: исполнитель 1 | Задача 2..........
05.08.2026 19:16:41 | Получение из очереди | очередь: task_queue | данные: исполнитель 1 | Задача 7.
05.08.2026 19:16:42 | Окончание | очередь: task_queue | данные: исполнитель 1
05.08.2026 19:16:42 | Получение из очереди | очередь: task_queue | данные: исполнитель 2 | Задача 6........
05.08.2026 19:16:43 | Окончание | очередь: task_queue | данные: исполнитель 2
05.08.2026 19:16:43 | Окончание | очередь: task_queue | данные: Урок2_RoundRobinВОднойИБ | бэклогСнят=0 | приростИсполнителей=7

Пока исполнитель 2 закрывает задачу на 10 точек (W76;5 с), у исполнитель 1 в логе подряд несколько коротких ack - это как раз эффект prefetch=1. Точный порядок строк может чуть съехать изR09;за планировщика ФЗ, важнее расклад по меткам и что лёгкие не ждут окончания «гири».

Таймаут у СоздатьПолучателя (ТаймаутПолучателяСекЧасть2 = 1) - это ожидание пустой очереди, а не «время на обработку». Пока в Ready есть сообщения, СледующееСообщение возвращает их сразу; секунда нужна только чтобы корректно выйти из цикла, когда пакет разобран.

Результат

Настроили рабочую очередь: quorum task_queue, persistent сообщения, ручное подтверждение, несколько фоновых исполнителей, предзагрузка = 1 на канале.

Дальше - часть 3: доставка одного сообщения многим получателям (Publish/Subscribe).

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

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