Очереди сообщений

Подсистема использует очереди сообщений для надёжной асинхронной доставки данных.

Принцип работы

┌─────────────────┐                    ┌─────────────────┐
│  Регистрация    │                    │  Обработка      │
│  сообщения      │───────────────────▶│  фоновым        │
│  (синхронно)    │   Очередь          │  заданием       │
└─────────────────┘                    └─────────────────┘

Преимущества:

• Операция регистрации быстрая и не блокирует пользователя
• Отказоустойчивость — сообщения не теряются при сбоях
• Масштабируемость — несколько фоновых заданий обрабатывают очередь параллельно

Регистры очередей

инт_ОчередьИсходящихСообщений

Основная очередь для исходящих сообщений.

Измерение	Тип	Описание
ИдентификаторСообщения	УникальныйИдентификатор	Уникальный ID сообщения

Ресурс	Тип	Описание
ПотокДанных	СправочникСсылка.инт_ПотокиДанных	Поток обработки
ИсходныеДанные	Строка	Сериализованное представление исходных данных (ЗначениеВСтрокуВнутр)
СформированноеСообщение	ХранилищеЗначения	Готовое сообщение

инт_ОчередьОтправкиИсходящихСообщений

Очередь для рассылки сообщений подписчикам (сообщение × подписчик).

Измерение	Тип	Описание
ИдентификаторСообщения	УникальныйИдентификатор	ID сообщения
Подписчик	СправочникСсылка.инт_Подписчики	Получатель

инт_ОчередьВходящихСообщений

Очередь входящих сообщений от внешних систем.

Измерение	Тип	Описание
ИдентификаторСообщения	УникальныйИдентификатор	ID сообщения

Ресурс	Тип	Описание
ПотокДанных	СправочникСсылка.инт_ПотокиДанных	Поток обработки
ВходящееСообщение	ХранилищеЗначения	Десериализованные данные сообщения

Реквизит	Тип	Описание
ДатаСоздания	Дата (дата и время, индексируется)	Момент регистрации сообщения в очереди

Регистрация сообщения

Базовый вызов

ИдентификаторСообщения = РегистрыСведений.инт_ОчередьИсходящихСообщений.ЗарегистрироватьСообщение(
    ИсходныеДанные,   // Ссылка на объект или ФиксированнаяСтруктура
    ПотокДанных,      // СправочникСсылка.инт_ПотокиДанных
    РегистрироватьДубль  // Булево, по умолчанию Ложь
);

Параметры

Параметр	Тип	Описание
ИсходныеДанные	Ссылка/ФиксированнаяСтруктура	Данные для формирования сообщения
ПотокДанных	СправочникСсылка	Поток для обработки
РегистрироватьДубль	Булево	Ложь — дубли игнорируются

Возвращаемое значение

• УникальныйИдентификатор — ID зарегистрированного сообщения
• Неопределено — если регистрация не выполнена (поток неактивен, дубль)

Пример использования

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

Защита от дублей

Подсистема автоматически рассчитывает хэш исходных данных и хранит его в регистре инт_ХешиОбъектов.

При РегистрироватьДубль = Ложь:

• Если хэш не изменился — сообщение не регистрируется
• Если хэш изменился — регистрируется новое сообщение

// Принудительная регистрация дубля
РегистрыСведений.инт_ОчередьИсходящихСообщений.ЗарегистрироватьСообщение(
    ДокументСсылка,
    ПотокДанных,
    Истина  // РегистрироватьДубль = Истина
);

Статусы сообщений

Исходящие сообщения

Перечисление инт_СтатусыИсходящихСообщений:

Статус	Описание
Новый	Зарегистрировано, ожидает формирования
ФормированиеСообщения	Формируется фоновым заданием
ОшибкаФормирования	Ошибка при выполнении обработчика
ГотовоКОтправке	Сформировано, ожидает рассылки
ПомещеноВОчередьОтправки	Добавлено в очередь для подписчиков
Отправлено	Доставлено всем подписчикам
ОшибкаОтправки	Ошибка при отправке

Статусы рассылки (по подписчикам)

Перечисление инт_СтатусыРассылкиИсходящихСообщений:

Статус	Описание
Новый	Ожидает отправки
ВПроцессеОтправки	Отправляется
Отправлено	Успешно доставлено
ОшибкаОтправки	Ошибка, будет повтор

Регистры статусов

инт_ТекущийСтатусИсходящихСообщений

Хранит текущий статус каждого сообщения:

// Получение текущего статуса
Статус = РегистрыСведений.инт_ТекущийСтатусИсходящихСообщений.ПолучитьСтатус(
    ИдентификаторСообщения
);

инт_ИсторияСтатусовИсходящихСообщений

Хранит полную историю изменения статусов:

Измерение	Описание
ИдентификаторСообщения	ID сообщения
ДатаИзменения	Метка времени

Ресурс	Описание
Статус	Новый статус
Комментарий	Текст ошибки или примечание

Удаление сообщений

// Удаление сообщения и всей связанной информации
РегистрыСведений.инт_ОчередьИсходящихСообщений.УдалитьСообщениеИзОчередиПоИдентификатору(
    ИдентификаторСообщения
);

Удаляются записи из:

• инт_ОчередьИсходящихСообщений
• инт_ТекущийСтатусИсходящихСообщений
• инт_ИсторияСтатусовИсходящихСообщений
• инт_ОчередьОтправкиИсходящихСообщений

Регламентная очистка

Регламентное задание инт_ОчисткаУстаревшихСообщений удаляет успешно отправленные сообщения старше заданного срока.

Первым шагом задание вызывает инт_СервисныеДействия.УдалитьОсиротевшиеСтатусыВходящихСообщений() — удаляет записи инт_ТекущийСтатусВходящихСообщений (и историю статусов), для которых нет записи в инт_ОчередьВходящихСообщений или не заполнен идентификатор сообщения. Такие «осиротевшие» статусы не могут быть обработаны и раньше блокировали обработку входящей очереди. Процедуру можно вызвать и вручную. Количество удалённых записей пишется в журнал регистрации (событие ПодсистемаИнтеграции.ОчисткаУстаревшихСообщений).

Срок хранения задаётся реквизитом УдалятьОбработанныеСообщенияЧерез потока данных. Сообщение попадает в регистр инт_СообщенияКУдалению в момент, когда становится обработанным. Если срок хранения задали позже, чем сообщения были обработаны, дату удаления им проставляет фоновое задание. Оно запускается из формы элемента потока данных, когда поток записывают с изменённым сроком хранения.

Задание запускается только при изменении значения срока: повторная запись потока с тем же сроком регистрацию не перезапускает. Если запустить задание не удалось (в журнал регистрации попадает запись события ПодсистемаИнтеграции.СообщенияКУдалению), регистрацию повторяют изменением срока хранения — например, задав другое значение и вернув прежнее.

Получение данных сообщения

// Получение данных сообщения по идентификатору
ДанныеСообщения = РегистрыСведений.инт_ОчередьИсходящихСообщений.ПолучитьДанныеОчередиПоИдентификатору(
    ИдентификаторСообщения,
    "ИсходныеДанные, ПотокДанных, СформированноеСообщение"  // Список полей через запятую
);

// Результат: Структура
// - ИсходныеДанные: Ссылка
// - ПотокДанных: СправочникСсылка
// - СформированноеСообщение: Соответствие (десериализованное)

Рекомендации

Момент регистрации

Регистрируйте сообщения в обработчиках проведения документов или подписках на события. Это гарантирует, что сообщение будет создано только для успешно записанных данных.

Мониторинг очередей

Настройте алерты на метрику pde_queue_length для раннего обнаружения проблем с обработкой очередей. Подробнее о метриках и настройке алертов см. Мониторинг и метрики.

Большие объёмы

При массовой регистрации сообщений (загрузка данных) используйте пакетную обработку и контролируйте размер очереди.

Следующие шаги

• Многопоточность — параллельная обработка очередей
• Валидация — проверка сообщений
• Подписчики — типы подписчиков
• Мониторинг и метрики — Prometheus-метрики и алерты