Организация обмена с помощью шины сообщений MSMQ

С выходом восьмой версии 1С в руки разработчиков попали сразу несколько совершенно замечательных вещей, которые позволяет строить сложные распределенные системы с участием 1С. Во-первых, сериализация в XML, а во вторых - планы обмена. Сериализация предоставила возможность представлять данные хранящиеся в базе 1С в нейтральном формате, пригодном для дальнейшей обработке любой системой, а планы обмена позволили описывать перечень данных для которых будут регистрироваться изменения, и которые будут выгружаться. При этом выгрузка, загрузка и регистрация ведется в разрезе логических узлов обмена. Узлы обмена в свою очередь позволяют легко реализовать схему обмена один к одному или один ко многим.

По сути единственная вещь, которая отдана на реализацию разработчикам - это собственно транспорт. А вот тут все оказалось довольно банально - почти во всех случаях все сводится к обмену файлами. Файлы выгружаются на FTP или любое другое общедоступное место одним из узлов обмена и загружаются другим. Такая схема имеет ряд достоинств, прежде всего довольно высокую надежность и простоту реализации. Но она и не лишена недостатков, прежде всего связанных с масштабируемостью и производительностью, особенно если требуется организовать какую-то произвольную обработку выгружаемых данных для их последующей загрузки в систему отличную от 1С. 

Уже достаточно давно люди пришли к осознанию того, что для интеграции разнородных сильнораспределенных систем требуется некая общая точка, интеграционное ПО, которое будет гарантировать доставку, обеспечивать возможность промежуточной обработки, да и вообще скроет разнообразные системы друг от друга. И при так взгляде на вещи, обмен файлами через общую шару уже не является идеальным. 

В данной статье, мы рассмотрим пример интеграции двух баз 1С с помощью шины сообщений. В качестве шины сообщений будем использовать MSMQ (хотя для меня привычней JMS, но в этом случае ценность примера будет гораздо ниже, так как вряд ли кто-то будет поднимать ActiveMQ, для которого тоже есть адаптеры на C#, и заниматься противоестественным играми с быдложабкой).  Тема судя по всему свежа, c учетом того, что поиск msmq или jms на том же Инфостарте не дает никаких результатов.

Для начало немного теории, шина сообщений - это промежуточный механизм для обмена сообщениями, который обладает следующими свойствами:

1. Клиенты шины сообщений не взаимодействуют друг с другом непосредственно, они даже не знают друг о друге почти ничего.
2. Все взаимодействие ведется через шину сообщений, шина сообщений считается априори надежной и транзакционной.
3. Логическим структурами, с помощью которых клиенты шины сообщений обмениваются информацией, являются очереди и топики.
4. Отправитель помещает сообщение в некоторую очередь, а получатель забирает сообщения из некоторой очереди, при этом между ними может быть сколько угодно промежуточных узлов-ретрансляторов/преобразователей. Отправитель видит только место куда он должен поместить сообщение, а получатель - откуда забрать. При этом гарантируется что сообщение будет получено в нужном порядке и только одним получателем (кстати, несколько конкурирующих получателей позволяют организвовать параллельную обработку там где сообщения не зависят друг от друга).
5. Отправитель публикует сообщение в топик, при этом его получает все подписчики, слушающие топик в данный момент времени.
6. Некоторые шины сообщений имеют уже готовый набор обработчиков, которые позволяют фильтровать сообщений, разбивать поток сообщений и вновь агрегировать его, здесь стоит упомянуть проект Camel для ActiveMQ.

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

Но вернемся примеру. Итак, в нем мы реализуем обмен XML сообщениями между двумя базами 1С, для обмена мы будем использовать очереди сообщений MSMQ, при этом мы таки будем сохранять сообщения на файловую систему для организации следа, в отладочных целях и для организации возможности быстрой повторной выгрузки. Кроме того в данном пример даже в случае отсутствия изменения генерируется сообщение, для наглядности. Дополнительно мы реализует систему оповещения об ошибках возникших при обмене, используя протокол XMMP и любой клиент (в моем случае - Pidgin).

Общая схема будет выглядеть так:

Для начала нам потребуется установить MSMQ.

установка:

ocsetup.exe MSMQ-Container;MSMQ-Server;MSMQ-DCOMProxy /log:D:\msmq-install.log

удаление:

ocsetup.exe MSMQ-Container;MSMQ-Server;MSMQ-DCOMProxy /uninstall

под Windows XP также можно установить через установку и удаление программ.

Также нам потребуется некая тестовая конфигурация (приведена в архиве), в которой мы создадим план обмена с дополнительным строковым реквизитом, который назовем ОчередьДляПолучения, в этом реквизите будет хранится очередь, в которую будут помещаться все сообщения для данного узла:

Дальше будет необходимо создать и определить ряд констант (в D:\ExchangeOut выгружаются сообщения для следа):

Следующий момент состоит в том, что создать адаптеры для подключения к MSMQ и Jabber-серверу - они уже есть в архиве (в готовом виде и в виде исходников). Для обоих адаптеров определим одинаковый интерефейс, как пример того, что можно делать взаимозаменяемые адаптеры для разных систем обмена сообщениями (хотя jabber использовать не стоит из-за отсутствия транзакций, а вот JMS таки да).

using System;
using System.Runtime.InteropServices;

namespace MSMQAdapter
{
    [ComVisible(true)]
    public interface IAdapter
    {
        void SetParameter(string _name, string _value);
        void ClearParameters();

        void SendFile(string _text);
        void Send(string _text);
        bool HasMessage();
        string Receive();

        void Start();
        void Stop();

        void Begin();
        void Commit();
        void Rollback();
    }
}

Ну и наконец, перейдем к написанию обработки, которая будет осуществлять обмена по выбранному плану и узлу (на примере получения):    

Процедура Получение(adapter, узел)
    adapter.Begin();
    Попытка
        Пока adapter.HasMessage() Цикл
            xml = adapter.Receive();

            Если СтрДлина(xml) > 500 Тогда
                ЗаписьВЖуранал("Получение: "+Сред(xml, 1, 500));
            Иначе
                ЗаписьВЖуранал("Получение: "+xml);
            КонецЕсли;
            чтениеXML = Новый ЧтениеXML();
            чтениеXML.УстановитьСтроку(xml);

            чтениеСообщения = ПланыОбмена.СоздатьЧтениеСообщения();
            объектУзла = узел.ПолучитьОбъект();
            исходныйНомерПринятого = Неопределено;

            Попытка
                чтениеСообщения.НачатьЧтение(чтениеXML);
            Исключение
                Если Не ФлажокСтрогийПорядокСообщений Тогда
                    Если СтрЧислоВхождений(
                                    ОписаниеОшибки(),
                                    "Номер сообщения меньше или равен номеру ранее принятого сообщения"
                                          ) Тогда
                        //Автокоррекция номера принятого
                        исходныйНомерПринятого = объектУзла.НомерПринятого;
                        объектУзла.НомерПринятого = 0;
                        объектУзла.Записать();

                        чтениеXML.Закрыть();
                        чтениеXML = Новый ЧтениеXML();
                        чтениеXML.УстановитьСтроку(xml);
                        чтениеСообщения.НачатьЧтение(чтениеXML);
                    КонецЕсли;
                Иначе
                    ВызватьИсключение ОписаниеОшибки();
                КонецЕсли;
            КонецПопытки;

            Если чтениеСообщения.Отправитель <> узел.Ссылка Тогда
                ВызватьИсключение "Неверный узел отправителя. Ошибка маршрутизации.";
            КонецЕсли;

            НачатьТранзакцию();

            Попытка
                Пока ВозможностьЧтенияXML(чтениеXML) Цикл
                    данные = ПрочитатьXML(чтениеXML);
                    ЗаписьВЖуранал("Данные: "+данные);
                    //данные.ОбменДанными.Загрузка = Истина;
                    данные.Записать();
                    ПланыОбмена.УдалитьРегистрациюИзменений(
                                        чтениеСообщения.Отправитель,
                                        данные
                                                       );
                КонецЦикла;

                Если Не исходныйНомерПринятого = Неопределено Тогда
                    Если чтениеСообщения.НомерПринятого > исходныйНомерПринятого Тогда
                        объектУзла.НомерПринятого = чтениеСообщения.НомерПринятого;
                    Иначе
                        объектУзла.НомерПринятого = чтениеСообщения.НомерПринятого;
                    КонецЕсли;
                Иначе
                    объектУзла.НомерПринятого = чтениеСообщения.НомерПринятого;
                КонецЕсли;

                ЗафиксироватьТранзакцию();

                чтениеСообщения.ЗакончитьЧтение();
                чтениеXML.Закрыть();
            Исключение
                ВызватьИсключение ОписаниеОшибки();
            КонецПопытки;
        КонецЦикла;
        adapter.Commit();
    Исключение
        adapter.Rollback();
        ВызватьИсключение ОписаниеОшибки();
    КонецПопытки;
КонецПроцедуры

Обработка включена в обе базы в архиве.

Теперь, прежде чем начать тестирование, надо позаботиться о jabber-сервере (или закомментировать соответствующие строчки в обработке), я использовал OpenFire, поднятый в виртуалбоксе - такие извращения были связаны с тем, что библиотека agsXMPP, разыменовывает под вистой localhost не как 127.0.0.1, а как ::1 (IPv6), в то время как OpenFire там не висит.

Итак запуск.... http://www.youtube.com/watch?v=5wolQ2STjdY. На видео демонстрируются изменения в центральной базе и их отображение в периферийной после обмена, а также отправка сообщения об ошибке через XMPP.

Что еще можно сделать:

1. Вынести обмен из обработки в регламентное задание.
2. Оформить сообщение об ошибке в виде структурированного XML, и написать специальную программу клиент (или конфигурацию к 1С), которая будет их принимать, сохранять и определенным образом отображать.
3. Переделать интерфейс JabberAdapter'а.
4. Вообще отказаться от jabber'а и использовать MSMQ для транспорта сообщений об ошибках.

Открытые вопросы:

1. Насколько такой механизм будет эффективен при большом количестве периферийных баз и необходимости обмена с другими системами, отличным от 1С?
2. Насколько легко его будет контролировать?
3. Любые другие.

Архив со всеми файлами приведен здесь //infostart.ru/projects/4993/

P.S. С учетом того, что я ковыряю эску чуть более трех месяцев, в моем быдлокоде может быть какая-нибудь глупость - любая конструктивная критика приветствуется.

P.P.S. Если кому интересно во что такая система может вылиться в пределе, ищете в гугле "Enterprise Integration Patterns RUS.djvu" - ооочень занимательная книжка...