Децентрализованная система обмена сообщениями

Мир IT-разработок идет по спирали. Основатели UNIX считали, что пусть программ будет много, но каждая из них выполняет свою задачу на «отлично». В начале 2000х основным трендом были программы-комбайны, выполняющие все, что только можно и даже больше. Сейчас вектор направления разработок начал движение в обратную сторону. И если раньше для обмена данными использовался в основном стандартный поток ввода/вывода, то теперь из-за того, что системы делают все более распределенными, передачей данными между узлами занимаются специализированные интеграционные комплексы (англ. Message Bus или Message broker).

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

Пример реализации я хотел бы представить.

Немного терминологии: шина сообщений (message bus), брокер сообщений (message broker) — все это сходные (но далеко не одинаковые) понятия, обозначающие программный комплекс, осуществляющий прием, обработку и передачу данных от одного узла к другому.
Абонент – приложение, отправляющее или/и принимающее сообщение по согласованному протоколу.

Для начала кратко о системах с центральным узлом (включая системы с резервированием вида master-slave, master-master).
Типичные корпоративные системы: TibcoEMS, IBM MQ, JBoss и другие. Из системы с открытым исходным кодом: RabbitMQ, Apache ActiveMQ, Apollo, Redis. Есть даже облачные сервисы: IronMQ. Наиболее часто используемые протоколы: AMQP, STOMP.
Основная идея – абоненты подключаются к общему серверу (кластеру серверов), который маршрутизирует сообщения между подключенными клиентами.

Преимущества:

• Централизованная настройка;
• Легкость обеспечения шаблона «гарантированная доставка»;
• Наличие библиотек практически на всех языках программирования;
• Широкий выбор конкретной реализации;

Тем не менее присутствуют ряд недостатков:

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

Несмотря на все недостатки, «большой» бизнес использует именно этот тип брокеров сообщений, так как стоимость потери данных значительно выше, нежели стоимость покупки более мощного аппаратного обеспечения.

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

Другим подходом является обмен данными без брокера (англ. broker-less). Обычно такая архитектура требует специализированную библиотеку и/или дополнительного программного обеспечения на узле абонента.
Из корпоративного сегмента, насколько мне известно, есть только один продукт: TIBCO Rendezvous (если кто посоветует альтернативы, буду очень признателен).
Из некоммерческих систем можно указать ZeroMQ, не требующего центрального сервера. Однако, эта библиотека не предоставляет какой-либо абстракции над сетью и не редко приводит к написанию собственных централизованных систем, сводя на нет всю идею децентрализации.
Основная идея децентрализованной архитектуры сходна с идеей P2P: абонент передает данные другим абонентам, не используя общего координирующего сервера. (DHCP, DNS и прочее я не считаю, так как они находятся на другом слое OSI).

Можно выделить следующие преимущества такого подхода:

• Распределение нагрузки на множество узлов;
• Отказоустойчивость. Система будет работать, пока есть хотя бы один отправитель и один получатель;
• Потенциально более высокая скорость работы.

Из недостатков можно отметить:

• Отсутствие централизованного управления;
• Практически невозможно обеспечить гарантированную доставку;
• Низкая распространенность подобных систем в бизнес ИТ и отсутствие каких-либо стандартов.

Для реализации часто используется протокол UDP, как не требующий установления соединения. Также, используя UDP multicast (далее просто multicast) возможно очень просто реализовать шаблон PUB/SUB, т.е. когда узел-издатель (PUB) публикует/рассылает данные по указанному топику (теме) узлам-подписчикам (SUB). По такой технологии работает ММВБ при рассылке биржевых данных (FIX FAST) и множество других систем.

Рассмотрим реализацию такой системы. Требования следующие:

• Реализация шаблона PUB/SUB;
• Основное назначение — системы оповещения с небольшими (до 1КБ) сообщениями;
• Система должна работать без центрального сервера и независимо от получателей;
• Основная ОС — Linux 2.6 или выше.

Для начала возьмем простейший вариант. Используя один multicast адрес, будем отправлять сообщения всем абонентам с указанием имени топика. Абоненты должны фильтровать данные согласно индивидуальному набору подписок.

Определим содержание UDP пакета:

• Имя топика;
• Данные.

Алгоритм подписчика можно описать следующим образом:

1. Подключится к multicast группе:
   
   

       struct ip_mreq mreq;
       struct sockaddr_in sin;
       sin.sin_family = AF_INET;
       sin.sin_port = htons(PORT);
       sin.sin_addr.s_addr = ADDR;
       mreq.imr_multiaddr = addr;
       mreq.imr_interface.s_addr = htonl(INADDR_ANY);
       setsockopt(fd, SOL_SOCKET, SO_REUSEPORT, &optval, sizeof (optval));
       if (bind(fd, (struct sockaddr *) &sin, sizeof (struct sockaddr_in)) < 0) {
           perror("Bind");
           return -1;
       }
       if (setsockopt(fd, IPPROTO_IP, IP_ADD_MEMBERSHIP, &mreq, sizeof (mreq)) < 0) {
           perror("Join");
           return -2;;
       }
   

   
   
2. Получить сообщение;
3. Если топик сообщения не в списке интересующих, перейти на пункт 2;
4. Обработать сообщение;
5. Вернуться на пункт 2.

Работа издателя еще проще:

1. Добавить к сообщению имя топика;
2. Отправить на multicast адрес сообщение.

Этот алгоритм простой, рабочий но обладает неприятным моментом: при наличии большого количества трафика, на узлы идет много бесполезных данных, которые они будут вынуждены обрабатывать, прежде чем отбросить.

Снизим нагрузку на получателей, назначив разные multicast адреса для топиков. Для вычисления группы используем любое хэш преобразование, например CRC-32, и получим необходимый IP адрес.

Алгоритм подписчика:

1. Рассчитать хэш значения от интересующих топиков:
   
   

    unsigned int addr_value = 4009754625 + (crc32_hash(subject) % 16777215);
   

   
   
2. Подключится к полученным multicast адресам. Особенности работы с ними хорошо описаны в этой теме;
3. Получить сообщение;
4. Если топик сообщения не в списке интересующих нас, перейти на пункт 3;
5. Обработать сообщение;
6. Вернуться на пункт 3.

Издатель:

1. Добавить к сообщению имя топика;
2. Рассчитать хэш топика;
3. Отправить на полученный multicast адрес сообщение.

Так как диапазон доступных multicast групп составляет 16777214 адресов, то если хэш-функция хорошо подобрана, на 33 миллиона разных топиков будет около двух совпадений.
Так как в Linux возможно использовать корректно только один сокет на одну multicast группу, то для получения данных рекомендуется использовать epoll.

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

Реализацию и исходный код можно посмотреть здесь.

P.S.

Я очень люблю родной русский язык, но из-за постоянного использования английского, могут быть проблемы в тексте. Если Вы заметите ошибку, то я буду очень признателен, если вы мне напишите о ней в личном сообщении.