Опыт 09/06/2022

Как интегрировать на проекте WebSocket и не сгореть: пошаговая инструкция

Java Developer в NIX

Привет! Меня зовут Роман Дашковский, я Java Developer в NIX и спикер IT-конференции NIX MultiConf. В этой статье я расскажу, с какими трудностями можно столкнуться при интегрировании на проекте WebSocket и как их преодолеть.

Представим простой ToDo-менеджер. Его клиент (т.е. окно браузера) посылает запросы на сервер и получает ответ. Для того, чтобы другой клиент увидел любые обновления, он должен запросить и получить эти данные.

Но что делать, когда клиент требует от пользователей наблюдать все апдейты «на лету»?

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

Что такое веб-сокет

Так что же такое сокеты, зачем их использовать и как вообще работают веб-приложения? Все это я объясню дальше на простом примере.

Веб-сокет — это протокол, который в отличие от HTTP позволяет организовать постоянную двунаправленную связь между клиентом и сервером.

Жизненный цикл соединения состоит из трех этапов:

Инициализация соединения — происходит handshake-запрос по HTTP, после чего соединение обновляется в WebSocket.
Пересылка данных — отправка сообщения может происходить от клиента на сервер и наоборот.

Разрыв соединения — инициатором разрыва может быть как клиент (например, закрытие вкладки), так и сервер (программно).

Подключение к SpringBoot-приложению

Давайте рассмотрим, как это все подключить к классическому SpringBoot-приложению.

Предварительная подготовка

Любые апдейты начинаются с чего? Правильно — с добавления необходимых зависимостей:

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

Так как мы будем работать с данными JSON-формата, добавим JSON от гуглов.
В нескольких местах я буду использовать некоторые утилиты, упрощающие процесс парсинга URL. Добавляем и третью зависимость от Apache.

Предварительная подготовка завершена. Перейдем к конфигурации Spring.

Конфигурация Spring

Обычно значительная часть подобной «магии» происходит в классах, отмеченных соответствующей аннотацией Configuration. Важно добавить еще одну — @EnableWebSockets.

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

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

Обработчик событий

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

Хранилищем соединений будет обычная карта, где ключом будет юзернейм пользователя, а значением — список всех его открытых сессий (вкладок браузера).

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

UI-часть

Как это выглядит с точки зрения UI-части? Как уже отмечалось, при инициализации сокета мы не можем добавлять ни хедеры, ни тело запроса. Но всегда есть возможность передать параметры через URL, что мы и сделаем.

Дальше добавляем все необходимые слушатели событий — и вот как это выглядит схематически:

В действительности же один клиент посылает HTTP-запрос на обновление данных. После чего в контроллере мы вызываем метод notifyAll или notifyUser по одной из предыдущих схем. Здесь появляется немного магии React. Как видим, данные обновляются сразу в двух окнах.

Все клево, все работает. Но если бы все было так просто, я не писал бы эту статью…

В enterprise-проектах обычно важный фактор — производительность. Но часто одни и те же микросервисы могут существовать в нескольких экземплярах, доступ к которым осуществляется непосредственно через Load Balancer. И в зависимости от загрузки CPU, занятости оперативной памяти или других параметров могут создаваться новые инстансы или удаляться существующие. Но как текущая реализация будет работать в этом случае?

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

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

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

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

Пригодится здесь Message Broker, но я выбрал ActiveMQ по простой и банальной причине — он один из самых популярных. Так что продолжаем кодить…

Как обычно, сначала обновляем зависимости. Я решил заюзать такую крутую штуку, как Apache Camel. Это открытый кроссплатформенный Java-фреймворк, позволяющий производить интеграцию приложений в простой и понятной форме. Camel очень упрощает флоу обработки данных. В особенности это комфортно, если нужно выстроить цепь из нескольких обработчиков. Также он очень легко интегрируется со Spring.

Не нарушаем традиции и дальше переходим к Spring-конфигам. Здесь нужно добавить два бина:

ActiveMqConnectionFactory, которому мы скармливаем пропсы из application.properties;
ActiveMQComponent — компонент Camel, который упростит процесс отправки и получения сообщений с брокера.

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

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

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

Но просто логировать сообщение недостаточно. Дальше его нужно отправить всем клиентам.

Для этого создаем обработчик:

парсим тело сообщения из обычной String в JSON-объект и затем к нашему Java-объекту;
считываем хидер, который мы засетили раньше;
пересылаем дальше уже существующими методами;
добавляем его в наш Data Flow в одну строку.

Наконец-то переходим к тесту. Есть клиенты, подключенные к двум разным инстанциям.

Выполняем апдейт на одном и наблюдаем обновления сразу обоих:

Теперь все работает. Хотел бы я так сказать, но…

Добавляем еще эндпоинт

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

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

Как фиксить проблемы с производительностью

В противном случае это может привести к проблемам с производительностью и формированием так называемого Bottlneck. Итак, как это пофиксить?

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

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

класснейм процессора для обработки сообщений;
параметры в формате карты.

Далее создаем процессор, который будет извлекать обновленные данные из БД, 3rd Party Service или из другого места и передавать их дальше.

После этого обновляем наш Data Flow в роутере. Если найден подходящий процессор, на него и идет обработка. Затем происходит отправка сообщения клиентам по WebSocket.

Вот упрощенная схема, как это работает:

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

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

Надеюсь, теперь при необходимости реализовать сокеты вы заранее продумаете все детали, будете знать, с какими трудностями можете столкнуться и как их преодолеть. А если хотите подробнее рассмотреть приведенный в этой статье код, переходите по ссылкам на репозитории с фронтом и бэкендом:

Если вы нашли ошибку, пожалуйста, выделите фрагмент текста и нажмите Ctrl+Enter.