Разработка системы синхронизации в реальном времени с использованием 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 сервере и клиенте, вы получите полноценную работающую систему синхронизации в реальном времени для вашего приложения.

Автор: krollik

Источник