DaJet MQ - надёжный обмен данными

Очереди создаются на одном экземпляре SQL Server в базе данных dajet-mq.

Очереди расположены на разных экземплярах SQL Server.

Клиент-источник отправляет сообщение у себя на сервере 192.168.0.1 в так называемую внешнюю очередь. Адрес внешней очереди настраивается в таблице маршрутов экзепляра SQL Server клиента-источника. Service Broker берёт на себя доставку сообщения в локальную очередь клиента-получателя на удалённом сервере 192.168.0.2. В случае успешного помещения сообщения в локальную очередь клиента-получателя Service Broker подтверждает доставку. Для этого также необходима настройка таблицы маршрутов на экземпляре SQL Server клиента-получателя.

Для взаимодействия друг с другом SQL Server'а должны открыть специальный порт Service Broker.

Использование DaJet MQ значительно упрощает настройку таблиц маршрутов на обоих серверах SQL Server.

Код 1С для открытия порта Service Broker на одном из серверов:

DaJet = Обработки.DaJetMQClient.Создать();

АдресСервера = "zhichkin\sqlexpress";
Пользователь = "sa";
Пароль       = "sa";
Провайдер    = "MSOLEDBSQL";
БазаДанных   = "master"; // Подключение к базе данных "master" SQL Server

СтрокаПодключения = DaJet.ОткрытьПодключение(АдресСервера, Пользователь, Пароль, Провайдер, БазаДанных);

НомерПортаServiceBroker = 4022; // Номер порта по умолчанию
	
DaJet.ВключитьУдалённоеВзаимодействие(НомерПортаServiceBroker);
	
DaJet.ЗакрытьПодключение();

Код 1С для настройки маршрута для внешней очереди (предварительно нужно открыть порты Service Broker - см. код выше):

DaJet = Обработки.DaJetMQClient.Создать();
	
ПараметрыОтправителя = Новый Структура();
ПараметрыОтправителя.Вставить("Провайдер", "MSOLEDBSQL");
ПараметрыОтправителя.Вставить("АдресХоста", "192.168.0.1");
ПараметрыОтправителя.Вставить("АдресСервера", "sql-server-sender");
ПараметрыОтправителя.Вставить("БазаДанных", Неопределено);
ПараметрыОтправителя.Вставить("ИмяПользователя", "sa");
ПараметрыОтправителя.Вставить("ПарольПользователя", "sa");
	
ПараметрыПолучателя = Новый Структура();
ПараметрыПолучателя.Вставить("Провайдер", "MSOLEDBSQL");
ПараметрыПолучателя.Вставить("АдресХоста", "192.168.0.2");
ПараметрыПолучателя.Вставить("АдресСервера", "sql-server-receiver");
ПараметрыПолучателя.Вставить("БазаДанных", Неопределено);
ПараметрыПолучателя.Вставить("ИмяПользователя", "sa");
ПараметрыПолучателя.Вставить("ПарольПользователя", "sa");

 // Очередь получателя - внешняя по отношению к серверу 192.168.0.1 (sql-server-sender)
ИдентификаторВнешнейОчереди = "514B05BB-CD37-4D17-828C-AFFB5161380B/queue/ЗаказыПокупателей";

DaJet.СоздатьМаршрут(ПараметрыОтправителя, ПараметрыПолучателя, ПолучательИдентификаторОчереди);

Для облегчения настройки можно использовать обработку dajet-mq-1c-client.epf, которая входит в комплект поставки.

Внимание! Все нижеописанные операции можно выполнить программно!

1. Открыть обработку "dajet-mq-1c-client.epf" и выбрать опцию "Добавить сервер".

2. Ввести данные сервера SQL Server, на котором нужно установить DaJet MQ.

Если используется Windows аутентификация, то имя пользователя и пароль указывать не надо - подключение к SQL Server будет выполнено под пользователем, от которого работает сервер 1С.

Если настройка схемы "точка-точка" не планируется, то заполнять поле "Адрес хоста" не обязательно.

3. После нажатия кнопки "Сохранить" сервер SQL Server будет добавлен в список.

4. Выделяем узел с названием добавленного ранее SQL Server и выбираем опцию меню "Установить DaJet MQ".

5. В случае успешной установки видим внизу следующее сообщение:

6. Выделяем узел "Локальные очереди" и выбираем опцию "Добавить локальную очередь".

7. Задаём имя очереди и жмём кнопку "ОК". В случае успешного создания локальной очереди она отображается в интерфейсе.

8. Можно выполнять отправку и получение сообщений.

Внимание! Все нижеописанные операции можно выполнить программно!

1. Установить DaJet MQ на обоих серверах SQL Server.

2. Открыть порт Service Broker на обоих серверах SQL Server, если они ещё не открыты.

При этом важно указать в настройках сервера SQL Server параметр "Адрес хоста" (см. инструкцию по установке DaJet MQ).

Если сервера SQL Server находятся на одном и том же физическом сервере, то порты Service Broker у них должны быть разные!

Нужно повторить следующие операции для каждого из серверов:

2.1. Выделить узел сервера и выбрать опцию меню "Открыть внешний порт".

2.2. В случае успешного от крытия порта видим следующее сообщение:

2. Создать локальную очередь на сервере-получателе. По отношению ко второму серверу это будет внешняя очередь. См. инструкцию по установке DaJet MQ выше.

3. Выделить узел "Внешние очереди (маршруты)" сервера-отправителя и выбрать опцию меню "Добавить внешнюю очередь".

4. В открывшемся окне настройки маршрута заполнить необходимые параметры и нажать кнопку "Создать маршрут".

5. В случае успешного создания маршрута видим следующее сообщение:

Список серверов и очередей изменится следующим образом:

6. Можно отправлять сообщения с сервера-отправителя в очередь сервера-получателя.

В данном конкретном случае подключаемся к серверу "zhichkin" и отправляем сообщение во внешнюю очередь "58094149-D544-4AA5-A1E0-0370E0473A8A/queue/test".

Получаем сообщения на сервере "zhichkin\sqlexpress" в его локальной очереди "58094149-D544-4AA5-A1E0-0370E0473A8A/queue/test".

1. Сохраняем скрипт установки install-dajet-mq-database.sql из одноимённого макета обработки dajet-mq-1c-client.epf во внешний файл.

2. Выполняем скрипт на сервере SQL Server, где хотим установить DaJet MQ.

Внимание!

При данном варианте установки могут быть нюансы с подключением к базе данных DaJet MQ от имени пользователя, под которым работает сервер 1С.

Чтобы не было проблем, добавьте пользователя сервера 1С во владельцы базы данных DaJet MQ.

// ************************************************************
// * Создание локальной очереди сообщений "ЗаказыПокупателей" *
// ************************************************************

DaJet = Обработки.DaJetMQClient.Создать();

АдресСервера = "zhichkin\sqlexpress";
Пользователь = "sa";
Пароль       = "sa";
Провайдер    = "MSOLEDBSQL";

СтрокаПодключения = DaJet.ОткрытьПодключение(АдресСервера, Пользователь, Пароль, Провайдер);

ИмяОчереди = "ЗаказыПокупателей";
ИдентификаторОчереди = DaJet.СформироватьИдентификаторОчередиСообщений(ИмяОчереди);

Если Не DaJet.ОчередьСуществует(ИдентификаторОчереди) Тогда

    ИдентификаторОчереди = DaJet.СоздатьОчередьСообщений(ИмяОчереди);

КонецЕсли;
	
DaJet.ЗакрытьПодключение();

// ************************************************************
// * Удаление локальной очереди сообщений "ЗаказыПокупателей" *
// ************************************************************

DaJet = Обработки.DaJetMQClient.Создать();

АдресСервера = "zhichkin\sqlexpress";
Пользователь = "sa";
Пароль       = "sa";
Провайдер    = "MSOLEDBSQL";

СтрокаПодключения = DaJet.ОткрытьПодключение(АдресСервера, Пользователь, Пароль, Провайдер);

ИмяОчереди = "ЗаказыПокупателей";

Обработка.УдалитьОчередьСообщений(ИмяОчереди);
	
DaJet.ЗакрытьПодключение();

DaJet = Обработки.DaJetMQClient.Создать();

АдресСервера = "zhichkin\sqlexpress";
Пользователь = "sa";
Пароль       = "sa";
Провайдер    = "MSOLEDBSQL";

СтрокаПодключения = DaJet.ОткрытьПодключение(АдресСервера, Пользователь, Пароль, Провайдер);

ИдентификаторОчереди = "514B05BB-CD37-4D17-828C-AFFB5161380B/queue/test";
ТелоСообщения = "Тестовое сообщение DaJet MQ";
Кодировка = "UTF-8";

DaJet.ОтправитьСообщение(ИдентификаторОчереди, ТелоСообщения, Кодировка);

DaJet.ЗакрытьПодключение();

DaJet = Обработки.DaJetMQClient.Создать();

АдресСервера = "zhichkin\sqlexpress";
Пользователь = "sa";
Пароль       = "sa";
Провайдер    = "MSOLEDBSQL";

СтрокаПодключения = DaJet.ОткрытьПодключение(АдресСервера, Пользователь, Пароль, Провайдер);

ИдентификаторОчереди = "514B05BB-CD37-4D17-828C-AFFB5161380B/queue/test";
Количество = 10; // Количество сообщений, получаемых из очереди за 1 раз
Таймаут = 5000; // Время реактивного ожидания появления сообщений в очереди, если она пуста (в миллисекундах)
Кодировка = "UTF-8";

DaJet.НачатьЧтениеСообщений(); // Открываем транзакцию чтения сообщений

МассивСообщений = DaJet.ПолучитьМассивСообщений(ИдентификаторОчереди, Количество, Таймаут, Кодировка);

Для Каждого Сообщение Из МассивСообщений Цикл
		
	ТелоСообщения = Сообщение.ТелоСообщения;
		
КонецЦикла;

DaJet.ЗафиксироватьЧтениеСообщений(); // Фиксируем транзакцию - сообщения будут удалены из очереди

DaJet.ЗакрытьПодключение();

DaJet = Обработки.DaJetMQClient.Создать();

АдресСервера = "zhichkin\sqlexpress";
Пользователь = "sa";
Пароль       = "sa";
Провайдер    = "MSOLEDBSQL";

СтрокаПодключения = DaJet.ОткрытьПодключение(АдресСервера, Пользователь, Пароль, Провайдер);

ИдентификаторОчереди = "514B05BB-CD37-4D17-828C-AFFB5161380B/queue/test";
Количество = 1; // Количество сообщений, получаемых из очереди за 1 раз
Таймаут = 1800000; // Время реактивного ожидания появления сообщений в очереди, если она пуста (30 минут)
Кодировка = "UTF-8";

КоличествоПолученных = 0;

Пока КоличествоПолученных = 0 Цикл

    DaJet.НачатьЧтениеСообщений(); // Открываем транзакцию чтения сообщений

    МассивСообщений = DaJet.ПолучитьМассивСообщений(ИдентификаторОчереди, Количество, Таймаут, Кодировка);

    КоличествоПолученных = МассивСообщений.Количество();

    Для Каждого Сообщение Из МассивСообщений Цикл
		
	    ТелоСообщения = Сообщение.ТелоСообщения;
		
    КонецЦикла;

    Если КоличествоПолученных = 0 Тогда
        DaJet.ОтменитьЧтениеСообщений(); // Закрытие транзакции обязательно!
    Иначе
        DaJet.ЗафиксироватьЧтениеСообщений(); // Фиксируем транзакцию - сообщения будут удалены из очереди
    КонецЕсли;

    КоличествоПолученных = 0; // Если хотим продолжить реактивное ожидание и обработку входящих сообщений

КонецЦикла;

DaJet.ЗакрытьПодключение();

DaJet = Обработки.DaJetMQClient.Создать();

АдресСервера = "zhichkin\sqlexpress";
Пользователь = "sa";
Пароль       = "sa";
Провайдер    = "MSOLEDBSQL";

СтрокаПодключения = DaJet.ОткрытьПодключение(АдресСервера, Пользователь, Пароль, Провайдер);

ИдентификаторОчереди = "514B05BB-CD37-4D17-828C-AFFB5161380B/queue/test";

// Неблокирующее получение данных - информация о получаемых в данный момент сообщениях тоже будет получена
ДанныеОчереди = DaJet.ПолучитьТекущееСостояниеОчереди(ИдентификаторОчереди);

DaJet.ЗакрытьПодключение();

Сообщить("Размер всех сообщений = " + ДанныеОчереди.РазмерСообщений); // в байтах
Сообщить("Общее количество сообщений = " + ДанныеОчереди.КоличествоСообщений);