В рамках одного из проектов нам потребовалось организовать стабильный и масштабируемый обмен сообщениями между различными информационными системами. Для решения этой задачи мы выбрали RabbitMQ — популярный брокер сообщений, реализующий концепции асинхронного взаимодействия.

В этой статье я расскажу, как устроен RabbitMQ, какие сущности в нем используются, как мы организовали маршрутизацию сообщений и как интегрировали его с платформой 1С, используя внешнюю компоненту PinkRabbitMQ. Все примеры взяты из промышленного проекта Таипит.

Описание функциональности

RabbitMQ это брокер сообщений, позволяющий реализовывать асинхронные обмены сообщениями между различными информационными системами. Устанавливается на отдельный сервер и имеет веб интерфейс (можно ознакомится по ссылке). В веб интерфейсе можно управлять настройками, о которых речь пойдет ниже.
Перейдем к устройству основных сущностей которые будут нами используются и их описанию.

Виртуальный хост (Virtual host)

Виртуальные хосты позволяют изолировать проекты или окружения внутри одного инстанса RabbitMQ. На нашем проекте, например, были подняты два VHost’а:

• taipit — для основного контура разработки

• taipit-test — песочница для индивидуальных тестов разработчиков

Точка обмена (Exchange) Точка обмена принимает сообщения и маршрутизирует их в нужные очереди в зависимости от заданного routing key и binding. Это центральный компонент, определяющий, куда «полетит» сообщение.

Очереди (Queues) 

Очереди хранят доставленные сообщения. Одна очередь может быть привязана к нескольким Exchange’ам и получать сообщения по различным ключам маршрутизации.
Если несколько очередей подписаны на один и тот же routing key, каждое сообщение копируется в каждую из них — это удобно для мультикаста данных, например, НСИ из MDM в несколько систем.

На проекте мы использовали отдельную очередь под каждую ИС.

Связи (Binding) 

Binding задает правило маршрутизации от Exchange к очереди. Основан на routing key — специальном свойстве, содержащемся в сообщении.

Например:

• очередь dv.kaz принимает из exchange data сообщения с ключами dv.kaz и dv.all

Ключ маршрутизации (Routing key) 

Это строка, по которой происходит выбор нужных очередей. Позволяет гибко управлять маршрутизацией. Возможны шаблоны (dv.*, *.kaz), что особенно полезно при широковещательных рассылках.

Сообщение (Message) 

Собственно полезная нагрузка, которую мы передаем между системами. Может содержать произвольные данные и дополнительные метаданные (об этом ниже).

Как работает маршрутизация сообщений

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

Интеграция с 1С

Для подключения 1С к RabbitMQ мы использовали внешнюю компоненту PinkRabbitMQ. Вся интеграционная логика была реализована в отдельной подсистеме проекта Таипит.

Основные методы компоненты:

• Connect — подключение к серверу

• BasicPublish — отправка сообщений

• BasicConsume — подписка на очередь

• BasicConsumeMessage — чтение одного сообщения

• BasicAck — подтверждение получения

• BasicCancel — отмена подписки

Также есть методы для создания очередей, exchange’ов и связей, что потенциально позволяет динамически настраивать структуру обмена прямо из 1С.

Схема в общем виде реализованная на проекте

Примеры работы с компонентой

Для соединения с компонентой используется метод Connect 

Для отправки сообщений используется метод BasicPublish

Для чтения сообщений из очереди используется метод BasicConsume

Для получения сообщения используется метод BasicConsumeMessage

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

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

Примеры использования разработанных методов по работе с компонентой

Получение данных

Отправка сообщений реализована с помощью подсистемы очереди обработки данных, получение через отдельное регламентное задание. Важный момент, на отправку и получение мы открываем соединение один раз на все время жизни регламента, далее получаем/отправляем сообщения, при этом если сообщений нет, то не разрываем соединение, а по таймауту пробуем получить/отправить сообщения заново, таким образом не тратим время на подключение. Для реализации ожидания используется компонента "Пауза".

Использование ключей маршрутизации в очередях

На примерах очередях показанных выше будет описан механизм работы маршрутизации сообщений с помощью ключей маршрутизации.

В ключах маршрутизации можно использовать маску для получателей, к пример по маске "*" очередь будет получать сообщения от всех отправителей. Если использовать маску "dv.*" то сообщение будет получено от все отправителей начинающимся с "dv.". На данном проекте есть одна мастер система Казначейство (dv.kaz), из нее все другие информационные системы должны получать информацию по НСИ. Для этого при отправки таких сообщений передается ключ маршрутизации "dv.all", так же во всех очередях в связях указан этот ключ маршрутизации, таким образом все информационные базы получаю информацию о НСИ.

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

Использование свойств сообщения RMQ

Далее приведены только те свойства, которые были использованы нами на проекте.

MessageId — идентификатор сообщения, формируется в базе отправителе, и через данное свойство передается в приемник, для того что бы однозначно сопоставлять сообщения в разных информационных системах.Type — тип, явно обозначающий какой «объект» передается в теле сообщения, используется, что бы не обращаясь к телу сообщения можно было понять по типу, что нужно делать с сообщением.ReplyTo — передается ключ маршрутизации равный самой очереди, в случае когда на отправленное сообщение необходимо отправить ответ в базу отправителя.CorrelationId — при отправки ответного сообщения передаем идентификатор сообщения родителя, для того что бы можно было понять ответ к какому сообщению нам пришел.

RabbitMQ оказался отличным решением для построения распределенной шины обмена сообщениями между различными ИС. Благодаря использованию routing key’ев, виртуальных хостов и возможности масштабирования, он позволил организовать гибкую и отказоустойчивую архитектуру интеграции. А внешняя компонента для 1С закрыла вопрос взаимодействия с платформой без сторонних сервисов или middleware.

Если вы строите интеграцию 1С с внешними системами, советую обратить внимание на подобный подход — он уже доказал свою надежность в боевых условиях.