Создание сервера обмена сообщениями в реальном времени на Go

Эта история - репортаж после моего выступления на goway.io о Centrifugo. Некоторые вещи здесь уже упоминались в моих предыдущих постах. Оригинальные слайды доступны здесь.

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

Меня зовут Александр Емелин, я программист из Москвы. На данный момент я работаю в Avito, одном из крупнейших секретных сайтов в мире. На работе мы помогаем людям продавать и покупать вещи, давая этим вещам вторую жизнь и просто позволяя людям экономить свои деньги.

Но сегодня я не буду говорить о том, чем мы занимаемся на работе, я расскажу о проекте с открытым исходным кодом, который я начал около 6 лет назад. Проект называется Центрифуго. Это сервер обмена сообщениями в реальном времени. Первоначально он был написан на Python (в структуре Tornado), но затем был переведен на язык Go. Буквально неделю назад я выпустил 2-ю версию Centrifugo - так что на самом деле это первое публичное объявление о новом сервере. Все мои замечания в этом выступлении относятся к этой новой версии.

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

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

Мотивация

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

Существует множество решений для обмена сообщениями в реальном времени. Этот слайд содержит некоторые из них.

Если вы создаете свой бэкэнд на асинхронном параллельном языке, таком как Erlang, NodeJS или Go, тогда у вас все в порядке даже без какой-либо структуры, хотя вам нужно будет решить некоторые конкретные проблемы на стороне интерфейса и бэкэнда. Например, масштабируемость, правильное управление подключением и так далее.

Фил Леггеттер с сайта pusher.com создал замечательный источник, описывающий современные технологии реального времени. Проверьте это - множество отличных серверов, библиотек и сервисов. Также хороший обзор есть в блоге deepstreamhub. Если вы начинаете поиск решения в реальном времени - начните с этого.

Мой личный опыт - Python и, если быть более конкретным, фреймворк Django. Как вы знаете, Django - это классическая структура рабочего потока, в которой каждый рабочий запускается в своем собственном процессе и потоке ОС и блокируется на время выполнения запроса. Теперь с постоянными соединениями, такими как Websocket, у вас быстро заканчиваются доступные воркеры, поэтому ваше основное приложение перестает принимать новые запросы. Это то, для чего изначально был создан Centrifugo - иметь дело с постоянными соединениями, тем самым давая бэкэнд возможность обслуживать общие краткосрочные запросы и публиковать новые сообщения для клиентов, используя Centrifugo API, когда это необходимо.

Django не одинок - существует множество подобных фреймворков на Python и других языках. Поскольку Centrifugo работает как отдельный сервис, можно просто интегрировать его, не внося много изменений в код приложения. На самом деле ничто не мешает использовать Centrifugo вместе с приложением, написанным на языке NodeJS или Go, поскольку у него есть некоторые полезные функции, которые я опишу очень скоро.

Обзор дизайна

В двух словах, Centrifugo - это сервер, который позволяет обрабатывать постоянные соединения от клиентов приложений и предоставляет API для публикации сообщений в реальном времени для подключенных клиентов, которые заинтересованы в этом сообщении. Клиенты указывают свой интерес к конкретным сообщениям, подписываясь на каналы (или, другими словами, темы). Таким образом, Centrifugo на самом деле является просто сервером PUB / SUB.

Теперь мы можем взглянуть на упрощенную схему интеграции Centrifugo с серверной частью приложения:

Как видите, в этой схеме задействованы 3 части - серверная часть вашего приложения, Centrifugo и пользователи вашего приложения. Пользователи подключаются к Centrifugo через Websocket или SockJS с веб-токеном JSON для аутентификации, подписываются на каналы и прослушивают сообщения. Как только на вашем сервере появляется новое событие, он публикует его в Centrifugo API, и сообщение отправляется подключенным клиентам. В Centrifugo вы можете использовать простые HTTP-запросы с JSON в теле или GRPC для вызова методов API.

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

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

Давайте посмотрим на основные функции Centrifugo:

Независимый от языка
Отделен от серверной части приложения
JWT-аутентификация с поддержкой истечения срока действия
Простая интеграция с существующим приложением
Горизонтальное масштабирование с помощью Redis
Высокая производительность
Восстановление сообщений после коротких отключений
Информация об активных подписчиках в канале
Кросс-платформенный (Linux, MacOS, Windows)
Лицензия MIT

Centrifugo используется во многих проектах, в основном в веб-приложениях. Он довольно популярен в сообществах Python и PHP. Некоторые компании, использующие Centrifugo, - это Mail.Ru, Badoo, sports.ru, Spot.im. На Spot.im установлена самая большая установка Centrifugo, о которой я слышал - это 300 тысяч клиентов в сети с 3 миллионами сообщений в минуту.

Транспортировка в реальном времени

Давайте немного поговорим о транспорте в реальном времени, используемом в Centrifugo.

Websocket - самый очевидный выбор в наши дни. У него большое преимущество - он работает везде - в веб-браузерах и в мобильных приложениях. Это протокол с низкими накладными расходами поверх TCP, который работает через те же порты, что и HTTP. Соединение начинается с механизма обновления по общему протоколу HTTP, а затем переключается на сеанс TCP. Я не буду много говорить о деталях протокола Websocket - конечно, я знаком с ним, поскольку он широко распространен в наши дни.

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

Но в наши дни ситуация намного лучше.

Текущая поддержка браузером https://caniuse.com/#search=websocket составляет около 93%. Некоторые клиенты по-прежнему используют браузеры без поддержки веб-сокетов, несколько расширений браузера могут блокировать трафик веб-сокетов к определенным доменам, поэтому в некоторых сценариях нам все еще нужен откат к транспорту HTTP, если мы хотим, чтобы все наши пользователи успешно подключились.

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

Только 2 из этих транспортов основаны на постоянном соединении - xhr-streaming и eventourse. Но поскольку они основаны на HTTP, они не могут быть двунаправленными, поэтому SockJS имитирует двунаправленную связь, используя отдельные кратковременные HTTP-запросы от клиента к серверу.

Наличие резервного HTTP-транспорта может дать одно интересное преимущество в среде веб-браузера. Как мы знаем, мы живем в то время, когда популярность HTTP / 2 растет. Если используется HTTP / 2, постоянные HTTP-соединения будут автоматически мультиплексированы в один реальный TCP-сеанс реализацией HTTP / 2. При открытии новых вкладок приложения Websocket вы устанавливаете новое TCP-соединение с сервером. Это может быть решено с помощью синхронизации через LocalStorage или SharedWorker, но HTTP / 2 просто обеспечивает «из коробки» мультиплексирование для транспортных соединений на основе HTTP.

Сейчас разрабатывается спецификация, которая позволит запускать веб-сокет через поток HTTP / 2-соединения.

Это был краткий обзор мотивации и общих понятий Centrifugo. Теперь давайте более конкретно о деталях реализации - как Centrifugo устроен внутри?

Внутренности

Как было сказано выше, их 2:

Веб-сокет на основе красивой библиотеки веб-сокетов Gorilla
SockJS на основе классной библиотеки Sockjs-Go

Когда я работал над версией 2, я также экспериментировал с двунаправленной потоковой передачей GRPC в качестве транспорта клиент-сервер. Но после некоторых измерений я обнаружил, что двунаправленная потоковая передача GRPC не имеет преимуществ по сравнению с Websocket. Например, если мы отправляем 10 тыс. Клиентов на один узел Centrifugo, то в случае веб-сокета потребление памяти на сервере будет около 500 МБ, а в случае GRPC оно будет в 4 раза больше - около 2 ГБ ОЗУ. И в моих синтетических тестах Websocket был почти в 3 раза более производительным в отношении использования процессора.

Centrifugo имеет собственный протокол, который описан в схеме protobuf - он очень похож на JSON RPC на своем высоком уровне. Также есть некоторое сходство с MQTT с точки зрения бизнес-логики. Доступны два формата сериализации: JSON и Protobuf.

Я уже упоминал, что Centrifugo можно масштабировать до многих узлов. Чтобы функции могли работать на многих узлах, у нас есть объект Engine. На самом деле это интерфейс с довольно большим количеством методов. Двигатель позволяет:

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

На данный момент существует 2 реализации движка - в памяти и движке Redis.

Механизм памяти позволяет запускать один узел Centrifugo, поскольку на данный момент нет механизма для того, чтобы каким-то образом соединить узлы.

Redis Engine позволяет масштабировать Centrifugo на множество узлов, которые будут подключены через Redis PUB / SUB, а также имеет встроенную поддержку сегментирования и высокую доступность через Redis Sentinel. Все общение с Redis осуществляется с помощью библиотеки Redigo (еще раз спасибо, Гэри Берд, ты буквально мой герой).

Модель доставки сообщений

В простом случае модель доставки - не более одного раза. Нет никаких гарантий, что сообщение будет доставлено. Очевидно, что поскольку задействованы Centrifugo на основе модели PUB / SUB и механизм Redis PUB / SUB - другой гарантии невозможно представить. Но на самом деле этого достаточно для большинства приложений, потому что Centrifugo выполняет роль транспорта сообщений - все состояние приложения находится в основной базе данных приложения, поэтому может быть восстановлено при необходимости.

Но Centrifugo предоставляет интересный механизм восстановления сообщений. Centrifugo сохраняет настраиваемый буфер сообщений в кеше и может автоматически восстанавливать состояние клиента после короткого отключения, отправляя ему пропущенные сообщения после повторной подписки клиента. Когда этот механизм для каждой публикации в канале имеет возрастающий порядковый номер, клиенты могут запомнить последний полученный порядковый номер и, таким образом, использовать его для восстановления состояния при повторном подключении. В этом случае Cenrtifugo отправляет пропущенные публикации клиенту, синхронизируя этот процесс с PUB / SUB, чтобы сообщения приходили клиенту в правильном порядке. Если Centrifugo не уверен, что все публикации были восстановлены, он сообщает об этом клиенту, используя специальный флаг в ответ.

Оптимизация

В исходном коде Centrifugo используется несколько оптимизаций. Давайте посмотрим на некоторые из них.

Для работы с буферами протокола мы используем библиотеку gogoprotobuf. Он использует генерацию кода и производит очень оптимизированный код для маршалинга и демаршалинга протобуфов. Этот код в 3–6 раз более производительный, чем стандартная библиотека Protobuf, которая основана на отражении и выделяет больше.

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

Если мы посмотрим на график пламени реального производственного экземпляра Centrifugo v1, мы увидим, насколько велико пламя, связанное с системными вызовами записи.

Дизайн протокола помогает объединять сообщения. В случае формата JSON несколько сообщений могут быть объединены вместе с использованием формата потоковой передачи JSON, в котором каждое отдельное сообщение протокола разделено символом новой строки. А в случае Protobuf мы используем формат с разделителями по длине, где каждое отдельное сообщение имеет префикс длины сообщения varint. Таким образом, мы можем просто записывать разные сообщения во временный буфер, а затем записывать их в соединение за один системный вызов. Конечно, мы также можем повторно использовать эти временные буферы с помощью sync.Pool, и мы это делаем.

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

См. Также Игровая площадка Go.

Представьте, что у нас есть исходный канал, из которого мы получаем элементы для обработки. Мы не хотим обрабатывать товары по отдельности, а будем обрабатывать их партиями. Для этого мы ждем поступления первого элемента из канала, затем пытаемся собрать столько элементов из буфера канала, сколько хотим, без блокировок и тайм-аутов. А потом сразу обработать кусочек собранных нами предметов. Например, создайте из них конвейер Redis и отправьте в Redis за один вызов записи соединения.

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

И мы стараемся объединять сообщения на стороне клиента - в наших клиентах - чтобы уменьшить количество читаемых системных вызовов.

Библиотека центрифуг

Как вы видите из моего выступления, Centrifugo - это автономный сервер с собственной механикой, встроенной в Go. Вопрос, который мне задавали несколько раз, можно ли повторно использовать функциональность Centrifugo из кода Go? Ответ был такой:

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

Во время работы над Centrifugo v2 одной из моих целей было выделить отдельную библиотеку, которая будет ядром Centrifugo v2, но все еще может быть повторно использована сообществом Go.

Результат моей работы - Библиотека центрифуг. Вам все еще нужно понимать, что Centrifuge как библиотека очень специфична с точки зрения того, что она имеет механику, унаследованную от сервера Centrifugo.

Библиотека Centrifuge предоставляет несколько вещей помимо функциональности сервера Centrifugo:

Собственная аутентификация с промежуточным ПО
Тесная интеграция с бизнес-логикой
Двунаправленный обмен сообщениями
Встроенные вызовы RPC
Больше свободы в управлении разрешениями на каналы

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

Давайте увеличим самую интересную часть:

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

Клиентские библиотеки

На данный момент у нас есть несколько клиентов по центрифугам и центрифугам:

Важно то, что клиенты могут работать как с сервером Centrifugo, так и с настраиваемым сервером, построенным на базе библиотеки Centrifuge. Давайте посмотрим на некоторые шаблоны, которые можно реализовать с помощью клиента Javascript.

Первый - PUB / SUB: