Подготовка окружения
Компонента 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
Вступайте в нашу телеграмм-группу Инфостарт