За что я люблю технологии 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 (быстрое согласование заявок).