Как создавать высокопроизводительные очереди сообщений с различной архитектурой

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

Рассказываем об основных сложностях развития высоконагруженных ИТ-систем и о способах их преодоления с помощью очередей сообщений на примере Tarantool Queue Enterprise.

Материал подготовлен по мотивам вебинара «Как создавать высокопроизводительные очереди сообщений с различной архитектурой». Вы можете посмотреть его здесь.

Актуальные бизнес-задачи и сопутствующие им сложности 

Во всех отраслях бизнеса развитие ИТ становится важным фактором масштабирования и успешности компаний. Поэтому бизнес все активнее внедряет сложные системы, строит решения для многопоточной обработки, делает акцент на работу с данными в real-time режиме или близком к нему. 

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

• сбор данных из источников в нужной последовательности;

• обеспечение консистентности данных в контуре;

• управление жизненным циклом объектов;

• настройка пайплайнов обработки;

• гарантированная доставка до получателя;

• обеспечение приоритизации выполнения задач;

• обеспечение надежной буферизации данных;

• выполнение бизнес-логики на потоке транзакций в рамках SLA;

• масштабирование входящей нагрузки.

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

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

• выравнивание нагрузки;

• приоритизацию обмена сообщениями между системами;

• обеспечение надежности доставки

• и не только.

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

Решение от Tarantool для асинхронной передачи данных

Tarantool на рынке более 15 лет — и сегодня предлагает широкий продуктовый портфель инструментов для хранения и обработки данных. Среди них:

• Tarantool Enterprise Edition; 

• Tarantool CDC;

• Tarantool DB;

• Tarantool Graph DB;

• Tarantool Column Store.

Наш опыт позволяет гибко реагировать на запросы клиентов и не только предоставлять им готовые продукты, но и разрабатывать новые решения в ответ на частые запросы бизнеса.

Например, именно так в экосистеме Tarantool появился Tarantool Queue Enterprise — высокопроизводительный инструмент для построения масштабируемых очередей задач и сообщений. Фактически он стал ответом на запрос клиента, которому были нужны компоненты для построения биржевой системы электронных торгов, способной безотказно, корректно и надежно работать с заявками, сделками и операциями сотен или даже тысяч клиентов.

В итоге уже сейчас с помощью решений экосистемы Tarantool можно закрыть весь типовой пайплайн работы с данными на всех этапах. 

Подробнее о Tarantool Queue Enterprise

Как уже упоминали раньше, Tarantool Queue Enterprise (TQE) — высокопроизводительный инструмент для построения масштабируемых очередей задач и сообщений.

Tarantool — промежуточное ПО для хранения и обработки данных. В основе — in-memory распределенная база данных, сервер приложений и средства масштабирования.

То есть Tarantool Queue Enterprise доступны преимущества работы в оперативной памяти, что позволяет обрабатывать до миллиона запросов в секунду. При этом инструмент является собственной, независимой разработкой и внесён в реестр российского ПО.

TQE поддерживает две модели взаимодействия очередь задач и очередь сообщений.

При работе в качестве классической очереди задач TQE фактически является аналогом RabbitMQ и предоставляет пользователям возможность:

• работы с моделью Put/Take (сквозная доставка до получателя);

• работы с Pull-моделью (получатель забирает данные из очереди);

• управления приоритетами обработки;

• отложенной отправки и получения.

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

• Pub/Sub — модель подписки на сообщения;

• Push-модель — очередь отправляет данные подписчикам;

• гибкая параметризация очистки очереди;

• кластерные идентификаторы сообщений;

• управление чтением данных через API (offsets).

Характеристики и преимущества TQE

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

• Производительность. In-memory обработка данных обеспечивает высокую скорость выполнения операций (на чтение и запись).

• Гибкость. Поддержка различных моделей взаимодействия (Publish/Subscribe, Put/Take) позволяет свободно адаптировать инструмент под любые архитектуры и бизнес-потребности.

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

• Хороший TTM. Коробочное решение требует минимальных ресурсов на установку и настройку, а также упрощает эксплуатацию. Это помогает сократить цикл выпуска фич и продуктов.

• Простота поддержки. Благодаря развитой и зрелой экосистеме Tarantool с лёгкой интеграцией между решениями, TQE позволяет сократить «зоопарк технологий», особенно если данные уже хранятся в Tarantool.

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

Решаемые задачи 

Очереди сообщений классически помогают решить несколько распространенных проблем. 

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

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

• Упрощают интеграцию новых сервисов. Обычно для расширения приложения или другой ИТ-системы надо глубоко перерабатывать его архитектуру: предусматривать новые каналы передачи данных, настраивать зависимости и не только. Интеграция очереди исключает эти проблемы — очередь становится «центральным» компонентом для обмена данными, поэтому для добавления новых сервисов достаточно подключить к ней нового читателя. 

• Сглаживают пиковые нагрузки. Очередь дает возможность обрабатывать сообщения в асинхронном режиме. Поэтому отправитель и получатель могут работать в своем темпе и обрабатывать сообщения по мере возможностей. Отказ получателя из-за перегрузок исключен.

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

TQE реализует весь стандартный и даже расширенный функционал очередей, поэтому инструмент можно использовать в качестве шины для:

• надежной обработки миллионов транзакций в реальном времени;

• персонализации предложений и маркетинга в реальном времени;

• сглаживания неравномерных нагрузок;

• связи множества микросервисов в одной системе;

• сбора данных со множества IoT-устройств;

• импортозамещения других pub/sub- и put/take-очередей в имеющихся системах.

Архитектура Tarantool Queue Enterprise

Как мы уже упоминали, Tarantool Queue Enterprise построен на базе Tarantool, который используется в качестве движка хранения данных. Поэтому он унаследовал все его преимущества в виде работы в in-memory, поддержки шардирования, синхронной и асинхронной репликации.

При этом архитектура решения зависит от выбранного способа взаимодействия с очередью.

Архитектура SQ

Архитектура SQ реализует тип взаимодействия put/take через протокол iproto. В ней реализована поддержка разных полезных фич вроде возможности назначения приоритетов и отложенных сообщений. 

Упрощенно принцип работы следующий:

1. Продюсер отправляет сообщение.

2. Сообщение записывается на один из шардов.

3. TQE SQ принимает сообщение и с помощью роутера передает его консьюмеру.

4. После прочтения консьюмером подтверждение передается в обратной последовательности.

5. После подтверждения обработанное сообщение удаляется из очереди.

   

Архитектура MQ 

Архитектура MQ реализует тип взаимодействия pub/sub через протоколы gRPC, гарантирует сохранение порядка доставки и дает возможность фильтрации сообщений при подписке. 

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

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

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

Как устроено сообщение в TQE MQ

Немного подробнее остановимся на TQE MQ и нюансах ее работы.

Базовой единицей TQE MQ является сообщение, которое передается от отправителя (продюсера) к читателю (консьюмеру). 

У сообщения есть ряд обязательных параметров:

• uint64 id — идентификатор сообщения;

• string queue — название очереди;

• bytes payload — полезная нагрузка.

Также есть опциональные параметры:

• *string routing_key — ключ маршрутизации, который указывается, если нужно положить сообщение в конкретную очередь;

• *string sharding_key — ключ шардирования, который указывается, если важно, чтобы сообщение попало на определенный шард;

• *string deduplication_key — уникальный ключ, использующийся очередью для определения дубликатов;

• *map metadata — таблица в формате «ключ — значение», которая используется для служебных целей.

API у очереди MQ очень простое — в большинстве случаев достаточно одного запроса. Например:

• для публикации:

Publish (message): id (в случае одного сообщения);

PublishBatch (message[]): ids[] (при публикации сразу нескольких сообщений).

• для подписки:

Subscribe (queue, cursor, filters): notifications[].

Алгоритмы публикации и подписки

Теперь немного о том, по какому принципу выстроена работа в TQE MQ. 

Так, публикация осуществляется по следующему алгоритму:

1. Publish отправляет запрос с параметрами сообщения.

2. Модуль API пересылает сообщение в доступный роутер.

3. Роутер определяет bucket id и решает, на какой шард отправить сообщение.

4. Шард присваивает сообщению идентификатор и сохраняет в базу данных.

5. Сообщение отправляется на реплику.

6. Модуль API сообщает клиенту идентификатор сообщения.

В случае подписки алгоритм отличается, но также состоит из нескольких шагов.

1. Клиент отправляет subscribe запрос в модуль API.

2. Устанавливается gRPC Stream.

3. Модуль API циклически опрашивает шарды на предмет сообщений.

4. Шарды формируют ответ или ожидают добавления новых сообщений.

5. Модуль API отправляет сообщения, полученные от шардов, клиенту вместе с композитным курсором.

Варианты применения Tarantool Queue Enterprise

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

TQE SQ как очередь задач

Использование Tarantool Queue Enterprise в качестве очереди задач — типовой сценарий. 

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

Таким образом, TQE SQ обеспечивает распределение нагрузки на получателей, гарантирует доставку и позволяет обрабатывать задачи в многопоточном режиме.

TQE MQ как брокер сообщений в бирже

Один из базовых сценариев использования Tarantool Queue Enterprise MQ — в качестве брокера сообщений в бирже, которая обрабатывает заявки от множества клиентов.

Основные требования к TQE в данном случае выглядят так:

• Клиенты отправляют заявку в сервис обработки:

  • запрос на покупку; 

  • запрос на продажу. 

• Клиент может управлять параметрами выполнения своей заявки:

  • аннулировать;

  • отложить;

  • исполнить в заданный временной интервал.

• Заявки попадают в брокер сообщений TQE, который:

  • буферизует заявки; 

  • проверяет по заданному алгоритму; 

  • упорядочивает по времени отправки заявки;

  • последовательно передает заявки в целевой сервис обработки.

• Сервис обработки заявок формирует сделки на основании подобранной пары заявок на покупку и продажу.

• Брокер сообщений TQE уведомляет клиента обо всех этапах обработки заявки от отправки до совершения сделки.

• Клиент может получить информацию по своим заявкам и сделкам на всем протяжении жизненного цикла, подписавшись на каналы своего брокера.

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

TQE MQ как буфер механизма CDC

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

В Tarantool Change Data Capture задачи такого буфера выполняет именно TQE MQ. Причем выполняет стабильно и прогнозируемо.

TQE SQ/MQ для фоновых операций

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

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

Вместо выводов

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

Понимая это, мы развиваем Tarantool Queue Enterprise, который изначально ориентирован на системы с высокими нагрузками, а также может легко адаптироваться под разные сценарии и задачи использования. Уже сейчас наш продукт сочетает быстродействие, надежность и масштабируемость. При этом мы продолжаем его совершенствование: в перспективе добавим GUI для управления, функционал архивации/восстановления данных очереди на диск, возможность масштабирования через k8s, а также поддержим Kafka API и конвейеры потоковой обработки сообщений с возможностью вычисления/обогащения данных on-flight (аналог Kafka Streams).

Узнавайте о новых релизах, вебинарах и выходящих статьях в Telegram-канале Tarantool News.

Задать вопросы команде разработчиков про использование Tarantool можно в официальном канале сообщества.

О принципах и примерах работы продуктов Tarantool читайте в блоге на сайте.