Это вторая часть, первая здесь
1. Введение
В первой статье мы рассмотрели архитектуру асинхронного обмена, в этой статье рассмотрим практический пример.
2. Архитектура практического решения
2.1 Общая схема взаимодействия
[1С-Отправитель] → [Broker] → очередь → [1С-Получатель]
Кратко:
* отправитель просто отдает
* брокер получает, хранит, передаёт дальше по запросу, гарантирует доставку
* получатель периодически опрашивает брокера, получает, воспроизводит
3. Формирование сообщения в 1С
3.1 Структура передаваемого сообщения
Структура сообщения от отправителя брокеру выяглядит следующим образом:
{
"exchange_id": "ИдентификаторОбменаУникальныйВнутриКанала",
"channel": "ИмяКанала",
"data": "ПередаваемыеДанные",
"proirity": "Приоритет"
}
Поле exchange_id - это идентификатор обмена, уникальный в рамках одного канала между получателем и отправителем.
Поле "data" содержит json-строку, кототрая представляет собой данные передаваемого объекта, это json в json-е, так сделано чтобы брокер не парсил лишнее, он распознает только идентификатор обмена, данные и приоритет, что именно в data - брокеру знать не нужно. exchange_id дублируется в data при отправке, т.к. он нужен получателю и передавать единым сообщением намного удобнее.
3.2 Архитектура отправителя
Для реализации функции отправителя понадобится регистр сведений Очередь следующей стркутуры:
Измерения:
- Ссылка, тип ДокументСсылка, СправочникСсылка
Ресурсы:
- exchange_id
- data
- Статус
- Приоритет
Приоритет позволяет упорядочить обмен, чтобы сначала уходили объекты не требующие наличия зависимостей, т.е. сначала элементы справлчников, а потом уже документы. Это позволяет снизить нагрузку на получателя, который после каждой порции данных должен проверять наличие необходимых зависимостей, чтобы воспроизвести очередной объект.
Регистрацию объекта в очереди можно производить либо в событии ПриЗаписи/ПриПроведении, либо периодически брать из таблицы изменений плана обмена, в рассматриваемом примере данные помещаются обработкой, это сделано в демонстрационных целях.
Json-данные объекта и его exchange_id лучше (но не обязательно) генерировать сразу, при помещении в очередь, так нагрузка на систему будет равномерно распределена во времени. Пример формата exchange_id я приводил в первой статье, продублирую ещё раз:
```
guid#version#type
```
Пример:
```
ad178c88-1b07-4234-11f1-063fc697887f#AAAA-BFAA#Документ.ПриходнаяНакладная
```
Теперь займёмся генерацией data. Рассмотрим формирование data на максимально простом примере - на справочнике Склады, в котором есть только стандартные реквизиты Код и Наименование. код следующий:
Функция Справочник_Склады(Знач ЭлементСсылка) Экспорт
Запрос = Новый Запрос;
Запрос.УстановитьПараметр("Ссылка", ЭлементСсылка);
Запрос.УстановитьПараметр("ИмяТаблицы", "Справочник.Склады");
Запрос.Текст =
"ВЫБРАТЬ
| Спр.Ссылка КАК Ссылка,
| &ИмяТаблицы КАК typ,
| Спр.ВерсияДанных КАК ver,
| ПРЕДСТАВЛЕНИЕ(УНИКАЛЬНЫЙИДЕНТИФИКАТОР(Спр.Ссылка)) КАК uid,
| Спр.Код КАК Код,
| Спр.Наименование КАК Наименование
|ИЗ
| Справочник.Склады КАК Спр
|ГДЕ
| Спр.Ссылка = &Ссылка";
Пакет = Запрос.ВыполнитьПакет();
Если Пакет[0].Пустой() Тогда
Возврат Неопределено;
КонецЕсли;
ДанныеОбъекта = Новый Структура;
// Шапка
Шапка = Новый Массив;
Выборка = Пакет[0].Выбрать();
Пока Выборка.Следующий() Цикл
Структура = Новый Структура;
Структура.Вставить("typ");
Структура.Вставить("ver");
Структура.Вставить("uid");
Структура.Вставить("Код");
Структура.Вставить("Наименование");
Структура.Вставить("Комментарий");
ЗаполнитьЗначенияСвойств(Структура, Выборка);
Шапка.Добавить(Структура);
КонецЦикла;
ДанныеОбъекта.Вставить("Шапка", Шапка);
Возврат Новый Структура("exchange_id, data, channel, priority",
exchange_id(Шапка[0].uid, Шапка[0].ver, Шапка[0].typ),
ИСТ_Сериализация.ТиповойСериализаторJSON(ДанныеОбъекта),
ИСТ_ОбщийМодульПовтИсп.ИмяКанала(),
0);
КонецФункции
Поля запроса uid, ver и typ используются для формирования exchange_id. Если нет необходимости отправлять каждую версию получателю, то в качестве ver можно всегда указывать 0 или любую другую константу. Если брокер увидит exchange_id, который был получен ранее, то он его проигнорирует и не передаст получателю, что избавит получателя от лишних телодвижений. На выходе в data получается примерно такой json:
{
"Шапка": [
{
"typ": "Справочник.Склады",
"uid": "c6978878-063f-11f1-ad17-8c881b074234",
"ver": "AAAABwAAAAA=",
"Код": 1,
"Наименование": "Склад 1"
}
]
}
После генерации данные помещаются в регистр очереди и ждут отправки, которая происходит регламентным заданием, а в нашем случае по нажатию кнопки в обработке.
3.3 Код отправки из 1С
Отправка сообщений производится через WebSocket-соединение. Код следующий:
Процедура ОтправитьСообщенияНаСервере() Экспорт
ДанныеКОтправке = ДанныеДляОтправки();
Если ДанныеКОтправке.Пустой() Тогда
Возврат;
КонецЕсли;
Клиент = WebSocketКлиенты.НайтиПредопределенный(Метаданные.WebSocketКлиенты.WebSocketSender);
Если Клиент = Неопределено Тогда
Возврат;
КонецЕсли;
ТекущееСоединение = Клиент.Подключить();
Если ТекущееСоединение = Неопределено Тогда
Возврат;
КонецЕсли;
СостояниеСоединения = Клиент.ПолучитьСостояниеСоединения();
КоличествоПопыток = 5;
Пока КоличествоПопыток > 0 И СостояниеСоединения <> СостояниеWebSocketСоединения.Открыто Цикл
ИСТ_ВебСокетыВызовСервера.Пауза(1000);
КоличествоПопыток = КоличествоПопыток - 1;
СостояниеСоединения = Клиент.ПолучитьСостояниеСоединения();
КонецЦикла;
Если СостояниеСоединения <> СостояниеWebSocketСоединения.Открыто Тогда
Возврат;
КонецЕсли;
ИСТ_ВебСокетыВызовСервера.ЗаписатьВЛог("ОтправитьСообщенияНаСервере", Клиент.Ключ, Новый Структура("Описание", "Старт: " + ТекущаяДатаСеанса()));
СтатусОтправлен = Перечисления.ИСТ_СтатусыОчереди.Отправлен;
Выборка = ДанныеКОтправке.Выбрать();
Пока Выборка.Следующий() Цикл
ТекущееСоединение.ОтправитьСообщение(Выборка.data);
РегистрыСведений.ИСТ_Очередь.ИзменитьСтатус(Выборка.Ссылка, СтатусОтправлен);
КонецЦикла;
ИСТ_ВебСокетыВызовСервера.ЗаписатьВЛог("ОтправитьСообщенияНаСервере", Клиент.Ключ, Новый Структура("Описание", СтрШаблон("Финиш: %1, отправлено: %2", ТекущаяДатаСеанса(), Выборка.Количество())));
КонецПроцедуры
4. Приём сообщений на стороне брокера
Брокер принимает сообщения и первым делом регистрирует их у себя, одновременно проверяет что exchange_id не был загружен ранее, если был, то проигнорирует его. Практически сразу же после получения брокер готов отдавать данные получателю, делает он это только после входящего запроса, когда от получателя приходит приглашение, например "ReadyForReceiving". После этого брокер начинает передачу всех не переданнных ранее сообщений и отмечает в своей базе факт передачи.
5. Обработка на стороне получателя (1С)
5.1 Схема работы получателя
При получении сообщения получатель не пытается его сразу же парсить, а просто складывает в регистр сведений СообщенияКЗагрузке, с примерно такой структурой:
Измерения:
- ИдентификаторСообщения (exchange_id)
Ресурсы:
- Сообщение (Сипрка - data в виде json-строки)
- Загружено (Булево)
- Приоритет (Число)
После окончания приема сообщений от брокера получатель начинает поочередно, согласно приоритета, воспроизводить объекты из json - производит парсинг и проверку наличия необходимых зависимостей, если всё в порядке, то объект создается и записывается в базу. Если каике-то заивисимости отсутствуют, то получатель ничего не делает, очередная попытка произвести загрузку будет выполнена после очередной порции сообщений. Поэтому важно организовать отправку сообщений на стороне отправителя таким образом, чтобы получатель как можно раньше получал необходимые данные, особенно если они нужно для других объектов. Если грубо обобщить - то сначала нужно отправлять справочники, а потом документы.
5.2 Код получения
Сначала отправляется приглашение на приме сообщений:
Процедура ЗапроситьСообщенияНаСервере() Экспорт
Клиент = WebSocketКлиенты.СоздатьКлиента(Метаданные.WebSocketКлиенты.WebSocketReceiver, ИСТ_ОбщийМодульПовтИсп.ИмяКанала());
Если Клиент = Неопределено Тогда
Возврат;
КонецЕсли;
ТекущееСоединение = Клиент.Подключить();
Если ТекущееСоединение = Неопределено Тогда
Возврат;
КонецЕсли;
СостояниеСоединения = Клиент.ПолучитьСостояниеСоединения();
КоличествоПопыток = 20;
Пока КоличествоПопыток > 0 И СостояниеСоединения <> СостояниеWebSocketСоединения.Открыто Цикл
ИСТ_ВебСокетыВызовСервера.Пауза(1000);
КоличествоПопыток = КоличествоПопыток - 1;
СостояниеСоединения = Клиент.ПолучитьСостояниеСоединения();
КонецЦикла;
Если СостояниеСоединения <> СостояниеWebSocketСоединения.Открыто Тогда
Возврат;
КонецЕсли;
ИСТ_ВебСокетыВызовСервера.ЗаписатьВЛог("ЗапроситьСообщенияНаСервере", Клиент.Ключ, Новый Структура("Описание", "Старт: " + ТекущаяДатаСеанса()));
ТекущееСоединение.ОтправитьСообщение(СтрШаблон("ReadyForReceiving:%1", ИСТ_ОбщийМодульПовтИсп.ИмяКанала()));
КонецПроцедуры
Далее, в обработчике события при получении сообщения происходит загрузка:
Процедура ПриПолученииСообщения_TestChannel(Соединение, Сообщение) Экспорт
Если СтрНайти(Сообщение, "{") > 0 Тогда
АтрибутыСообщения = ИСТ_Сериализация.ТиповойДесериализаторJSON(Сообщение);
ИСТ_ВебСокетыВызовСервера.ДобавитьСообщениеКЗагрузке(АтрибутыСообщения);
Иначе
ИСТ_ВебСокетыВызовСервера.ЗаписатьВЛог("ИСТ_ОбработчикиСобытийПолучателя.ПриПолученииСообщения_TestChannel", Соединение.Ключ, Сообщение);
КонецЕсли;
КонецПроцедуры
Воспроизведение (загрузка) объектов
После очередной порции сообщений получатель всегда пытается загрузить то что получил.
&НаСервереБезКонтекста
Процедура ЗагрузитьСообщенияНаСервере()
Запрос = Новый Запрос;
Запрос.УстановитьПараметр("channel", ИСТ_ОбщийМодульПовтИсп.ИмяКанала());
Запрос.Текст =
"ВЫБРАТЬ
| ИСТ_СообщенияКЗагрузке.exchange_id КАК exchange_id,
| ИСТ_СообщенияКЗагрузке.data КАК data
|ИЗ
| РегистрСведений.ИСТ_СообщенияКЗагрузке КАК ИСТ_СообщенияКЗагрузке
|ГДЕ
| НЕ ИСТ_СообщенияКЗагрузке.Загружено
| И ИСТ_СообщенияКЗагрузке.channel = &channel
|
|УПОРЯДОЧИТЬ ПО
| ИСТ_СообщенияКЗагрузке.priority";
Выборка = Запрос.Выполнить().Выбрать();
Пока Выборка.Следующий() Цикл
Если СообщениеЗагружено(Выборка.data) Тогда
РегистрыСведений.ИСТ_СообщенияКЗагрузке.СообщениеЗагружено(Выборка.exchange_id);
КонецЕсли;
КонецЦикла;
КонецПроцедуры
&НаСервереБезКонтекста
Функция СообщениеЗагружено(Знач Сообщение)
Чтение = Новый ЧтениеJSON;
Чтение.УстановитьСтроку(Сообщение);
Структура = ПрочитатьJSON(Чтение);
Если Не Структура.Свойство("Шапка") Тогда
Возврат Ложь;
КонецЕсли;
Данные = Структура.Шапка[0];
Если Данные.typ = "Справочник.Контрагенты" Тогда
Возврат ИСТ_ЗагрузкаСообщений.ЗагрузитьКонтрагента(Структура);
ИначеЕсли Данные.typ = "Справочник.Номенклатура" Тогда
Возврат ИСТ_ЗагрузкаСообщений.ЗагрузитьНоменклатуру(Структура);
ИначеЕсли Данные.typ = "Справочник.Склады" Тогда
Возврат ИСТ_ЗагрузкаСообщений.ЗагрузитьСклад(Структура);
ИначеЕсли Данные.typ = "Документ.ПриходнаяНакладная" Тогда
Возврат ИСТ_ЗагрузкаСообщений.ЗагрузитьПриходнуюНакладную(Структура);
Иначе
Возврат Ложь;
КонецЕсли;
КонецФункции
...
Функция ЗагрузитьСклад(Знач Структура) Экспорт
РеквизитыШапки = Структура.Шапка[0];
УИД = Новый УникальныйИдентификатор(РеквизитыШапки.uid);
ЭлементСсылка = Справочники.Склады.ПолучитьСсылку(УИД);
Если СсылкаСуществует(ЭлементСсылка) Тогда
Возврат Истина;
КонецЕсли;
ЭлементОбъект = Справочники.Склады.СоздатьЭлемент();
ЭлементОбъект.УстановитьСсылкуНового(Справочники.Склады.ПолучитьСсылку(УИД));
ЭлементОбъект.Наименование = РеквизитыШапки.Наименование;
ЭлементОбъект.Записать();
Возврат Истина;
КонецФункции
6. Работа с зависимостями, идемпотентность и защита от дублей, роль брокера
Тезисно проговорю ещё раз архитектуру, если где-то повторюсь, заранее приношу извинения:
- отправитель только отправляет, может формировать отправку по приоритету, но это необязательно, вся ответственность за полноту отправляемых данных лежит именно на нём
- брокер - получает, ведёт журнал обмена, недопускает повторную отправку, отправляет по запросу получателя
- получатель запрашивает, сначала просто сохраняет сообщения в сыром виде, потом пытается распарсить и воспроизвести
- получатель, в случае отсутствия какой-то зависимости ничего не запрашивает целенаправленно, просто ждёт очередную порцию данных
- и отправитель и получатель общаются исключительно с брокером
7. Производительность
Для тестирования производительности выгрузки в демонстрационной конфигурации были созданы:
- Приходные накладные - 1000 шт
- Количество строк в одном документе - 50
Все значения ссылочных типов были заполнены рандомно
Выгрузка
Выгрузка в файл:
23 сек
Выгрузка брокеру:
3 сек
Запрос сообщений
Запрос сообщений (получение от брокера):
2 сек
Загрузка в базе-получателе
Из регистра сообщений к загрузке:
2 сек
Из файла:
17 сек
Как видно, разница достаточно ощутимая, на больших данных она может иметь решающее значение.
Это всё, что я хотел показать, исходники брокера можно взять [здесь](https://github.com/edex-dev/edex), база, на которой велась разработка и тестирование, приложена к статье.
Как попробовать?
Брокер (должен быть установлен Go, лучше последней версии):
1. Скачиваем репоизторий
2. Собираем с помощью Makefile, либо командой:
3. Запускаем edex.exe
go build -o edex.exe
Можно запустить без предварительной сборки:
go run .
1С-отправитель:
1. Скачиваем, создаём демо-базу
2. Заполняем тестовыми данными (Основное - Сервис - Управление тестовыми данными)
3. Создаем сообщения на отправку (Отправка - Сервис - Обмен сообщения через WebSocket - Создать сообщения на отправку)
4. Отправляем сообщений (Отправка - Сервис - Обмен сообщения через WebSocket - Отправить сообщения)
1С-получатель:
1. Создаём ещё одну демо-базу
2. Запрашиваем сообщения (Отправка - Сервис - Обмен сообщения через WebSocket - Запросить сообщения). На этом шаге заполнится регистр сведений "Сообщения к загрузке"
3. Загружаем сообщения (Отправка - Сервис - Обмен сообщения через WebSocket - Загрузить сообщения)
Вступайте в нашу телеграмм-группу Инфостарт