Разработка системы синхронизации в реальном времени с использованием SockJS, Django, Tornado и ZeroMQ

Не так давно, разрабатывая очередной программный продукт, наша команда разработчиков столкнулись с задачей реализации полноценной системы синхронизации пользовательских данных в реальном времени, путем рассылки (PUSH метод) изменений сервером. В самом приложении объем данных был не велик, но они могли просматриваться несколькими пользователями одновременно. Поэтому нам был необходим легковесный и достаточно производительный подход к синхронизации данных в рамках Веб-приложения. После того как были рассмотрены различные пути к решению этой задачи, мы остановили свой выбор на достаточно популярном эмуляторе WebSocket’ов – SockJS, который использует различные алгоритмы обмена данными между клиентом и сервером, в зависимости от браузера, которым пользуется клиент. В рамках данной статьи я не буду заострять внимание на том, почему был сделан именно такой выбор (по этому поводу написано немало статей, в том числе и на хабрахабре), а просто скажу, что мы ещё ни разу об этом не пожалели.

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

В этой статье я хотел бы рассказать о том, как мы решили эту задачу.

Итак, исходные данные:

• портал (написан на Django) с API
• веб клиент (с sockjs-client библиотекой для организации синхронизации)
• SockJS сервер (в виде tornado-sockjs, что являлось одной из причин, по которой мы сделали свой выбор в сторону этого продукта)

Общую задачу можно разделить на две подзадачи:

• Организация получения уведомлений конечным пользователем об изменениях в системе (тривиальная задача)
• Создание сообщений/уведомлений о наличии изменений в системе используя средства Django (напомним, что мы не можем использовать стандартный подход и создавать такие сообщения, используя стандартный JS клиент, т.к. в таком случае нам не удастся обработать изменения, сделанные при помощи набора APIs).

Изучив доступную информацию, стало ясно, что без “посредника” эту задачу решить нельзя. В качестве посредника была выбрана система обмена сообщениями, а именно ZeroMQ. Среди основных достоинств именно этой системы стоит отметить её простоту, легкость, а также поддержку языка Python (что для нас является весьма важным условием).

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

Изменения данных в системе, сделанные конечным пользователем с использованием веб интерфейса или APIs, в конечном счете попадают на наш Django сервер, где через ZeroMQ передаются в SockJS сервер, который, в свою очередь, уведомляет пользователей о факте изменения данных отображаемого объекта.

Давайте перейдем к коду

Веб клиент

Для получения уведомлений нам необходимо подключить библиотеку sockjs-client, устанавливать соединение с сервером и подписаться на нужные события. Сделать это можно примерно следующим образом:

<script src="/js/sockjs-0.3.4.min.js"></script> <script> var SyncServer = new SockJS("http://example.com/echo"); SyncServer.onmessage = function(e) {     // ваш обработчик сообщений }; </script> 

ZeroMQ сервер

Для реализации этой части нам понадобится сам ZeroMQ и библиотека Python для работы с ним — pyzmq. После этого необходимо настроить и запустить сервер в режиме прокси (данный режим называется FORWARDER). Сделать это можно буквально несколькими строками кода:

import zmq  def main():     try:         context = zmq.Context(1)         # Socket facing clients         frontend = context.socket(zmq.SUB)         frontend.bind("tcp://127.0.0.1:XXXX")          frontend.setsockopt(zmq.SUBSCRIBE, "")          # Socket facing services         backend = context.socket(zmq.PUB)         backend.bind("tcp://127.0.0.1:YYYY")          zmq.device(zmq.FORWARDER, frontend, backend)     except Exception, e:         # Handle exception         pass     finally:         frontend.close()         backend.close()         context.term()  if __name__ == '__main__':     main()  

Запустив подобный скрипт, вы получите proxy сервер, ожидающий сообщения на порту XXXX и отправляющих их на порт YYYY.

SockJS сервер

В качестве SockJS сервера был выбран стандартный SockJS-tornado с некоторыми небольшими изменениями, позволяющими ему принимать сообщения от ZeroMQ сервера (эти изменения вы можете найти в __init__ методе класса SockJSMyRouter)

from sockjs.tornado import SockJSConnection, SockJSRouter from tornado import ioloop as tornado_ioloop, web import zmq from zmq.eventloop import ioloop from zmq.eventloop.zmqstream import ZMQStream  ioloop.install() io_loop = tornado_ioloop.IOLoop.instance()  class SyncConnection(SockJSConnection):     _connected = set()     stream = None      def __init__(self, session):         super(SyncConnection, self).__init__(session)         self.stream.on_recv(self.on_server_message)      def on_open(self, request):         self._connected.add(self)      def on_message(self, data):         pass      def on_server_message(self, data):         message = "your message"         self.broadcast(self._connected, {'message': message})      def on_close(self):         self._connected.remove(self)   class SockJSMyRouter(SockJSRouter):      def __init__(self, *args, **kw):         super(SockJSMyRouter, self).__init__(*args, **kw)         socket = context.socket(zmq.SUB)         socket.setsockopt(zmq.SUBSCRIBE, "")         socket.connect("tcp://127.0.0.1:YYYY")         self._connection.stream = ZMQStream(socket, self.io_loop)  if __name__ == '__main__':     context = zmq.Context()      EchoRouter = SockJSMyRouter(SyncConnection, '/echo')      app = web.Application(EchoRouter.urls)     app.listen("ZZZZ")     io_loop.start() 

Django сервер

В рамках Django сервера нам необходимо добавить лишь несколько строк, ответственных за создание уведомлений и отправку их на наш ZeroMQ сервер. Сделать это можно следующим образом:

import zmq  context = zmq.Context() socket = context.socket(zmq.PUB) socket.connect("XXX") socket.send("your message") socket.close() 

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