Как правильно организовать очереди?

Question

[Даша Циклаури]

Есть некое приложение:
- несколько реплик микросервисов (сервис_1), которые держат WebSocket соединение с клиентом (либа centrifuge)
- брокер rabbitmq
- stateless сервис бизнес логики (сервис_2) для общения с базой
- redis для хранения сквозных данных между репликами сервисов

Юз кейс такой:
1) клиент коннектится на любой инстанс сервиса_1, шлет команду
2) команда сервисом_1 отсылается в брокер
3) сервис_2 читает команду, зная кто и что запрашивает
4) сервис_2 отправляет в брокер команду, которая должна быть направлена в нужный инстанс и далее нужному клиенту

Вопрос: как правильно организовать отправку команды сервис_2 -> сервис_1 ? Проблема в том как узнать какой сервис на каких пользователей подписан? хранить эту связь в redis? Или на каждого пользователя создавать по очереди? Или это все одна очередь и каждое сообщение должно быть вычитано каждым инстансом сервиса_1 а затем отфильтровано?

Comments

[d'Ivan]

Интересная задача, непростая. Думаю, стоит привлечь экспертов по тегам.

Answer 1

[ayazer]

я конечно мимокрокодил с другого стека, но как вариант:

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

в этот момент мы уже получаем равномерное распределение пользователей по серверам И возможность гарантировано по идшнику понять на каком с серверов висит коннекшн. самая очевидная проблема - автоматический скейлинг этого всего в обе стороны. бо прийдется много жонглировать. условиях не идеального коннекта/загруженой сетке тут будут обрывы (но возможно это и допустимый компромисс).

как узнать какой сервис на каких пользователей подписан

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

хранить эту связь в redis

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

Или на каждого пользователя создавать по очереди? Или это все одна очередь и каждое сообщение должно быть вычитано каждым инстансом сервиса_1 а затем отфильтровано?

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

Answer 2

[Сергей]

Если клиент не закрывает соединение с конкретным инстансом (или если закрыл, то ответ не должен получить), то посмотрите в сторону RPC, вот с помощью чего оно работает, вот примеры.
Если результат требует много вычислительных ресурсов и повторять нежелательно, то либо создавать очереди под каждого клиента, либо где-то кэшировать результат (например, в СУБД).

Comments

[Даша Циклаури]

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

очереди под каждого клиента это не считается дорого?

[Сергей]

Опишите тогда поподробней процесс взаимодействия пользователя и требования, например, если ответ готов, когда клиент ушёл, это надо держать в очереди до его возвращения? Клиент может подключиться к абсолютно любому серверу и выбор сервера случайный процесс? Чем больше требований к сохранению состояния, тем сложнее будет система. Как вариант хранить результат в БД (при отсылке удалять), а RPC использовать как реалтайм уведомление о готовности, и 99% времени именно он и будет работать.

[Евгений Самсонов]

Даша Циклаури, Смотрел на днях ролик https://www.youtube.com/watch?v=eP_02zkwM5Y, где рассказывают про реализацию чатов в Авито на WebSocket. Под каждый пользовательский чат отдельная очередь, но используют очереди Redis, а не RabbitMQ, потому что последний жрет ресурсы, если много очередей создавать

[Даша Циклаури]

Сергей,
- пользователь может отключаться по своей воле, тогда при этом событии данные из очереди очищаются и сервисы websocket отписываются
- пользователь может отключаться временно (на короткий срок, разрывы по причине перезагрузки nginx(edge), плохая связь)
- пользователь подключается к случайному websocket сервису (hash балансировку не планируется), то есть даже в момент разрыва он может переподключиться к уже другой ноде
- сохранность данных-некритичный показатель, поскольку все действия - результат бизнес-логики и операций с базой данных (то есть на каждую команду данные уже в базе), восстановление состояния при долгом разрыве происходит через сервис бизнес-логики и считывание из базы данных

[Даша Циклаури]

Евгений Самсонов, спасибо, тогда думаю есть смысл использовать nats

[Даша Циклаури]

Евгений Самсонов, классный видос, спасибо, теперь смотрю в сторону "отказаться от рэбита" и заменить на редис. Мной, кстати, задавался вопрос про Centrifugo в телеге и Саша ответил, что невозможно дисконнекты отлавливать, - это сразу заставляет писать используя либу Centrifuge =/

[Сергей]

Если переподключение пользователя и перезагрузка сервиса (отказ или обновление) влияют на суммарно малый процент от всех запросов, повтор запроса это не дорого, и есть возможность повлиять на клиентский код, то на мой взгляд тут не надо сильно стараться что-то сохранить, а просто повторить запрос с клиента, и тот же RPC (через RabbitMQ/NATS/Redis) вполне зайдёт. Плюсом это избавит от головной боли отладки, когда что-то не так с "состоянием".

[Даша Циклаури]

Сергей, с RPC пока не укладывается в голове как реализовать, сервер не хранит ответ (а только состояние игрока и комнаты).
Это игра: клиент сделал запрос, {списалась сумма, посчиталось текущее текущее состояние и сгенерировался ответ}(транзакционно), отправилось игровое состояние.
Если уходить в RPC, то придётся и на сервере реализовывать логику повторного запроса (но без списания суммы, иначе двойная ставка спишется) и клиент меня. По этой причине (как одной из) несколько лет назад первая версия с RPC была заменена на однонаправленный поток данных сервер-клиент

[Сергей]

Даша Циклаури, то есть через всю эту систему Вы хотите сделать транзакционный канал обновления состояния клиента, интересная затея. По сути надо реализовать TCP уровня приложения. У меня не было подобного опыта, но возникает пара возможных путей:
1) У состояния есть версия (к примеру. таймстамп сервера, или просто увеличивающееся число, наподобии optimistic lock). Клиент его получает и запрос на изменени шлёт с текущей версией. На сервере, если версия отличается, то операция отклоняется и клиента просять обновить состояние). Но такое не сильно будет работать, когда допускается несколько параллельных сессий одного и того же клиента.
2) Клиент на каждый запрос генерит уникальный айди (UUID, например) и, если надо повторить то использует тот же идентификатор. Сервер держит список последних запросов за какой-то интервал времени и, если приходит дубликат, то отклоняет его.

Но в обоих случаях по сути решается проблема ненадёжного канала связи и неважно что стоит меджу сервисом_1 и _2.

[Zhainar]

В целом странно, хоть процесс выглядит асинхронно, но требует синхронности. Какая роль у сервиса 1 кроме проксирования данных от клиента к сервису 2?

[Даша Циклаури]

Zhainar, никакой кроме как держать websocket соединения и масштабироваться

Answer 3

[d'Ivan]

А в чём проблема отправлять из сервис_1 в брокер и идентификатор пользователя и идентификатор сервиса (экземпляра), к которому он подключен? Тогда каждый экземпляр сервис_2 будет знать кому предназначена команда. Не вижу смысла плодить очереди на каждого пользователя (и тем более заниматься их управлением) и даже для каждого экземпляра сервис_1.
Так что я за более простой вариант:

Или это все одна очередь и каждое сообщение должно быть вычитано каждым инстансом сервиса_1 а затем отфильтровано?

Comments

[Даша Циклаури]

идентификатор сервиса (экземпляра), к которому он подключен

возможен такой кейс: команда отправлена в брокер, сервис_1 умер или перезапускается, пользователь отвалился и переподключился уже к новому (поскольку был disconnect на веб-клиенте), бизнес-логика произвела действия и положила данные в очередь. данные не будут вычитаны нужным экземпляром по его идентификатору

[d'Ivan]

Даша Циклаури, надо подумать. Утром виднее будет ;-).

[Евгений Самсонов]

Даша Циклаури, У каждого обработчика своя очередь, которая биндится к общему exchange. Сообщения летят в exchange и разлетаются по всем очередям, там уже фильтруется. Если обработчик упал, то так как никто не читает больше эту очередь, ничего не потеряется.