Кейс: реактивный подход в высоконагруженном приложении на примере сервиса для начисления кэшбэка

Рассказывает команда SimbirSoft

Эта статья не ставит своей целью описание фреймворков и архитектуры, поскольку они и так достаточно хорошо задокументированы. Скорее, она предназначена для тех, кто начинает работу с микросервисами и Project Reactor, и описывает основные особенности указанных технологий и то, с чем придётся столкнуться и работать.

В отличие от монолитной, микросервисная архитектура основана на выделении небольших независимых служб, каждая из которых реализует отдельную бизнес-функцию. Если в монолите всё связано (и в случае отказа одной функции могут «отвалиться» остальные), то микросервисы позволяют обеспечить гибкость и устойчивость системы. Крупные IT-решения могут содержать в своей архитектуре десятки микросервисов, и с каждым из них может работать отдельная независимая команда.

Пример

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

На первый взгляд, задача выглядела просто.

1. За каждый оплаченный (продлённый) полис начислять пользователю определённый кэшбэк X% через бухгалтерский сервис. Пользователю должна быть доступна информация о поступившем кэшбэке.
2. По достижении определённого суммарного кэшбэка автоматически переводить средства клиенту через бухгалтерский сервис. Пользователю должна быть доступна история выплат.

Основной проект использовал очередь сообщений Kafka в качестве средства обмена информацией между микросервисами, а также в качестве единственного перманентного реплицированного хранилища информации.

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

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

Blocking vs Non-blocking

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

Но, во-первых, не всегда есть возможность приобрести такой сервер. Во-вторых, и это самое главное, что-то всегда может пойти не так. Например возникнут какие-либо задержки бэка или проявится ситуация «retry storm» — когда множество пользователей приложения из-за недоступности функций инициируют повторные попытки запроса со своих устройств. Тогда количество активных соединений и потоков увеличивается. В этом случае кластерные ноды могут попасть в спираль — резервные копии потоков увеличивают нагрузку на сервер и перегружают кластер. Конечно, можно встроить механизмы регулирования, чтобы компенсировать риски и помочь поддержать стабильность во время этих событий, но понятно, что это не панацея. Кроме того, восстановление может быть достаточно долгим и рискованным.

Асинхронные системы работают по-другому. В них обычно одному ядру соответствует один поток, а цикл запроса-ответа обрабатывается через события и коллбэки. Получается, что там, где мы раньше «платили» за запрос целым потоком, теперь просто добавляется ещё одно сообщение.

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

Устойчивость системы достигается за счёт самой природы асинхронного подхода. Во-первых, мы, конечно, можем попробовать обработать исключение локально. А во-вторых, и это главное, в асинхронных системах компоненты не блокируются обработкой исключений: ошибка в одном компоненте не влияет на остальные. Более того, если один компонент не справляется с обработкой сообщения, то сообщение может быть обработано другим компонентом, записанным на этот же адрес.

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

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

Согласно требованиям заказчика, реализовать проект нужно было на Java для удобства его дальнейшей поддержки. Распространённых вариантов высокоуровневых абстракций было не так уж и много, поэтому мы остановились на Project Reactor.

Работа с Project Reactor

Reactor — реализация спецификации Reactive Streams, об этом вопросе уже достаточно подробно рассказывали.

Несмотря на обилие документации по Reactor, всё же отдельно хотелось бы отметить одну из особенностей при работе со стримами Reactor. В Reactor есть 2 основные структуры данных — Flux и Mono. Обе они являются реализациями интерфейса Publisher и представляют собой асинхронный поток элементов, либо единичный асинхронный элемент соответственно.

Стримы в Reactor внешне очень похожи на стандартные стримы Java по коллекциям (java.util.stream.Stream). В Java, конечно, Stream — это не только и не столько механизм работы с коллекциями. Но тут важно помнить, что Flux — это тем более не коллекция.

В Java перед началом стрима по коллекции у нас есть все её элементы, мы знаем её размер и т. п. Flux же лучше рассматривать как некую отложенную коллекцию, количество элементов которой на момент выполнения стрима неизвестно.

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

Схема работы

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

Мы видим, что у приложения есть 3 основных процесса:

1. Получение информации извне (полисы, пользователи, оплата) и начисление кэшбэка на её основе. Это будет первый сервис — «Калькулятор».
2. Общение с пользователем. Приложение обращается к хранилищу, чтобы по определённому пользователю найти нужный кэшбэк. Это будет второй сервис — «Хранилище».
3. Общение с бухгалтерским сервисом. Начисление кэшбэка и выплаты должны быть проведены через этот сервис. Это будет третий сервис — «Бухгалтер».

Итак, схема довольно простая:

«Калькулятор» вычисляет кэшбэк по каждому полису/пользователю/факту оплаты и отправляет сообщение в отдельную очередь. Сервисы «Хранилище» и «Бухгалтер» читают сообщения из этой очереди. «Хранилище» сохраняет кэшбэк и показывает его пользователю, а в случае достижения минимального порога выплаты — инициирует зачисление средств на карту пользователя. «Бухгалтер» вызывает внешний бухгалтерский сервис для физического начисления бонусов.

Организация локального хранилища

Особое значение имеет очерёдность этих процессов. Мы видим, что наш «Калькулятор» работает на основе сообщений от внешних сервисов. Возможны ситуации, когда «Калькулятор» на основе одного входящего сообщения не сможет принять решение об отправке. Например, ему нужно проверить 2 внешних топика: полисы и оплату. В этом случае необходимо внутреннее хранилище, которое мы формируем на основе всех внешних сообщений.

Сравнивая стандартные SQL варианты (PostgreSQL, MySQL) и NoSQL подход, мы в этом проекте в качестве локального хранилища, решили отдать предпочтение MongoDB по нескольким причинам:

1. Для mongoDB есть готовый фреймворк по работе с Project Reactor — reactive mongo.
2. Малое количество таблиц и связей между ними.
3. Простота использования, нет необходимости следить за соответствием моделей таблицам БД.

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

Репроцессинг

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

Если это невозможно по тем или иным причинам, можно попробовать изолировать в отдельном слое логику работы с состоянием. Иначе говоря, поставить над логикой с состоянием дополнительный уровень абстракции. В этом случае в приложении есть сегмент statefull, но он изолирован, другие части не связаны с состоянием.

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

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

Прочие особенности

Ещё одна особенность Project Reactor при работе с фронтом заключается в том, что в большинстве случаев нам недостаточно просто получить какое-либо значение. Чаще нам нужно получить значение и затем отслеживать его изменения. Этот вопрос достаточно просто решить с помощью reactive mongo. У хранилища из библиотеки reactive mongo есть методы получения и отслеживания, которые вернут не только требуемое значение, но и все его последующие изменения, если таковые будут.

Также обратим внимание на сервис «Бухгалтер». Предположим, что этот сервис работает с внешним API по REST или, как в нашем случае, по SOAP. Здесь также действуют требования по репроцессингу, и нужна отдельная очередь истории. Но также возможны и дополнительные требования по устойчивости системы в целом.

Например, что будет, если внешний API ответит 500 ошибкой? В нашем случае мы можем воспользоваться стандартным механизмом Reactor .retryBackoff() — он попробует отправить сообщение ещё несколько раз, увеличивая задержку между повторными сообщениями. Можно также настроить стрим на отлавливание определённых ошибок и реагировать только на них. Подробнее можно посмотреть тут.

Тестирование

Конечно, на создании рабочих модулей проект не заканчивается. Нам нужно проверить его работоспособность, в частности с помощью юнит-тестов. Для модулей на Project Reactor в юнит-тестах используют StepVerifier — это внутренний компонент, который позволяет правильно протестировать функциональность. Документация по StepVerifier легко доступна и полноценна.

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

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

Вывод

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

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

Использование Reactor позволяет достаточно просто реализовать асинхронную работу, сделать решение более наглядным и понятным для дальнейшей поддержки. При этом работа с Project Reactor потребует особого внимания при написании кода, а именно при выстраивании стримов Flux и Mono, также необходимо всегда сверяться с документацией и проводить промежуточные тесты.

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