За что я люблю технологии 1С?
Это за широкую возможность интеграции с другими системами. А сколько возможностей открывают расширения, написанные на внешней библиотеке DLL. Предлагаю рассмотреть одну из таких возможностей - интеграции через брокер-сообщения.
Интеграция 1С в экосистеме
Я столкнулся с ситуацией, когда экосистема проектов написаны на разных платформах. У нас есть мобильное приложение, web-портал и несколько РИБ 1С баз и нам нужно настроить мгновенный обмен данными номенклатур между этими базами, желательно с минимальной задержкой. Я не буду говорить о структуре сообщения, которое нам нужно передать, это может быть XML, и JSON, что угодно. Давайте поговорим о способе передачи этих данных.
Требования:
большие объемы данных
оперативная выгрузка с минимальной задержкой
возможность принимать данные в нескольких потоках
Сейчас в 1С популярны способы интеграции:
HTTP и web-сервис на стороне 1С
подключение HTTP get/post запрос
синхронизация данных XML
ODATA соединение к базе 1С
внешние источники
Каждый из этих способов хорошо подходит для решения определенных задач, но не в нашей ситуации. Коротко опишу почему.
HTTP и web-сервис имеют ограничения на одновременное подключение, по умолчанию установлено ограничение на 10 параллельных запросов, можем увеличить количество по необходимости. Представьте, стороннее приложение начинает отправлять в цикле get/post запрос, этот лимит мгновенно заполнится, и начнут вылетать ошибки time-out. Эта война между приложениями никогда не закончится.
Еще одной причиной неэффективности таких способов является то, что мы не сможем выполнять оперативную выгрузку данных с минимальной задержкой. Эти способы не смогут обеспечить мгновенное получение данных из систем.
Решение этих задач я вижу в брокерах сообщений, таких как RabbiMQ, Kafka, NUTS, ActiveMQ. Мое внимание привлекла именно Kafka - непрерывная передача информации со smart-периферии (конечных устройств) в IoT-платформу.
Когда данные не только передаются, но и обрабатываются множеством клиентов, которые называются подписчиками (consumers).
В роли подписчиков выступают приложения и программные сервисы. Здесь имеют место отложенные вычисления, когда подписчиков меньше, чем сообщений от издателей - источников данных (producer). Сообщения (messages) записываются по разделам (partition), темам (topic) и хранятся в течение заданного периода. Подписчики сами опрашивают Kafka на предмет наличия новых сообщений и указывают, какие записи им нужно прочесть, увеличивая или уменьшая смещение к нужной записи.
Давайте перейдем к практике создания простого приложения на 1С для обмена данными. План создания архитектуры интеграции посредством Kafka будет следующий.
сначала напишем библиотеку DLL для установки соединения
чтобы стать подписчиком (consumers) и слушать топик на наличие новых данных, мы создадим «бессмертное» фоновое задание
для отправки данных в топик (producer) мы сделаем подписку в 1С на запись объектов, для мгновенной отправки
Поднять и настроить сервер Kafka не сложно, в интернете достаточно необходимой информации.
Напишем библиотеку DLL. Я выбрал язык программирования C# (опыта в этом языке у меня немного), вот мой пример:
using System; using System.Collections.Generic; using System.Text; using System.Threading; using System.Net.NetworkInformation; using System.Security.Cryptography; using System.Runtime.InteropServices; using Confluent.Kafka; using System.Linq; namespace Kafka1CConnect { [ComVisible(true)] [InterfaceType(ComInterfaceType.InterfaceIsIDispatch)] [Guid("764A91C9-7472-4DC6-810B-4EB31BE26DB9")] public interface Kafka1CEvent { [DispId(0x60020000)] string GetInfo(); } [ComVisible(true)] [ClassInterface(ClassInterfaceType.AutoDual)] [Guid("A6498157-CCC0-47B9-8B0F-40BBD3AFF096")] [ComSourceInterfaces(typeof(Kafka1CEvent))] [ProgId("Kafka1C.ConnectKafka")] public class ConnectKafka : IDisposable { public ConnectKafka() { // do nothing } ~ConnectKafka() { Dispose(); } private IProducer<string, string> Producer; private IConsumer<string, string> consumer; private CancellationTokenSource cts; public void Dispose() { } public void ProducerCreate(string ConnectString, int ComprType=1, int LingerMs=0, int mmb=1000000) { // do nothing var config = new ProducerConfig { BootstrapServers = ConnectString, Acks = Acks.All, LingerMs = LingerMs, MessageMaxBytes = mmb, CompressionType = (CompressionType) ComprType }; Producer = new ProducerBuilder<string, string>(config).Build(); } public string SendProducer(string topic, string table, string value) { var msg = new Message<string, string> { Key = table, Value = value }; //byte[] bytes = Encoding.UTF8.GetBytes("123123231"); //msg.Headers.Add("head", bytes); var res = Producer.ProduceAsync(topic, msg).GetAwaiter().GetResult(); return res.Status.ToString(); } public void ProducerClose() { Producer.Flush(TimeSpan.FromSeconds(5)); Producer.Dispose(); } public void ConsumerCreate(string ConnectString, string topic, string GroupId, bool eof=false) { var config = new ConsumerConfig { GroupId = GroupId, BootstrapServers = ConnectString, AutoOffsetReset = AutoOffsetReset.Earliest, EnablePartitionEof = eof, Debug = "cgrp" }; consumer = new ConsumerBuilder<string, string>(config).Build(); consumer.Subscribe(new string[] { topic }); } public string[] Consume() { var AnswerArr = new List<string>(); cts = new CancellationTokenSource(); var tokenCancel = cts.Token; try { var cr = consumer.Consume(tokenCancel); if (cr == null) { return null; } else { if (!cr.IsPartitionEOF) { AnswerArr.Add(cr.Topic + ""); AnswerArr.Add(cr.Key + ""); AnswerArr.Add(cr.Value + ""); AnswerArr.Add(cr.Timestamp.UtcDateTime.ToString()); } else { AnswerArr.Add("EOF"); } } } catch (ConsumeException e) { AnswerArr.Add(e.Error.Reason); } return AnswerArr.ToArray(); } public void ConsumeClose() { cts.Cancel(); consumer.Close(); } }}
В этой библиотеке основные методы это:
для чтения данных: ConsumerCreate(), Consume(), ConsumeClose()
для отправки данных: ProducerCreate(), SendProducer(), ProducerClose()
Теперь напишем фоновое задание с «бессмертным» циклом для непрерывного чтения данных:
kafka = Новый COMОбъект("Kafka1C.ConnectKafka"); kafka.ConsumerCreate("localhost","9092","test.topic","base01") Пока Истина Цикл ArrAns = kafka.Consume(); Если ArrAns = NULL или ArrAns = Неопределено Тогда Сообщить("Обработано " + Строка(к) + " записей!" ); Прервать; КонецЕсли; массив = ArrAns.Выгрузить(); Если массив.Количество()=4 Тогда Топик = массив[0]; // Топик (тема), с которого было прочитано сообщение Ключ = массив[0]; // Ключ сообщения Значение = массив[0]; // Сообщение Время = массив[0]; // Время отправки ИначеЕсли массив.Количество()=1 Тогда Сообщить(массив[0];); КонецЕсли; КонецЦикла; kafka = NULL;
Думаю, в этом коде ничего сложного, мы создаем экземпляр библиотеки DLL в коде 1С и пользуемся его методами.
Обратите внимание, хоть цикл и непрерывный, при вызове метода kafka.Consume() цикл становится на паузу, в ожидании нового сообщения, это позволяет уменьшить лишнюю нагрузку на процессоры 1С.
Так мы обеспечили мгновенную отправку данных с базы А при записи объекта и мгновенное чтение в базе Б.
Я показал самый простой пример применения брокер сообщений для интеграции внутри баз 1С и с другими базами. Это привело к развитию микросервисных систем с моментально быстрым с обменом данным. Все задачи решались на много быстрее. Также решили проблему логирования сообщений и хранение истории обмена для отказоустойчивой системы интеграции.
Целью данной статьи было не показать best-practices или не убеждать вас, уважаемые читатели, что для шины данных может использоваться только Apache Kafka, а предать свой опыт, для раздумья и развития вашего собственной экосистемы. Вы можете использовать любой брокер сообщения, для шины данных, главное правильно построить архитектуру и без "фанатизма".
В итоге мы применение нашей шины в таких проектах, как: MDM (быстрое согласование и контроль данных), DWH (сбор структурированных данных), CRM (быстрое согласование заявок).