我有一个应用程序,在每个websocket连接(在tornado的open回调中)创建一个
然而,我的
我进一步尝试并意识到,在
我还尝试将
是否有一种方法可以将
zmq.SUB
套接字到现有的zmq.FORWARDER
设备。这个想法是通过回调从zmq接收数据,然后将其通过websocket连接传递给前端客户端。
https://gist.github.com/abhinavsingh/6378134
ws.py
import zmq
from zmq.eventloop import ioloop
from zmq.eventloop.zmqstream import ZMQStream
ioloop.install()
from tornado.websocket import WebSocketHandler
from tornado.web import Application
from tornado.ioloop import IOLoop
ioloop = IOLoop.instance()
class ZMQPubSub(object):
def __init__(self, callback):
self.callback = callback
def connect(self):
self.context = zmq.Context()
self.socket = self.context.socket(zmq.SUB)
self.socket.connect('tcp://127.0.0.1:5560')
self.stream = ZMQStream(self.socket)
self.stream.on_recv(self.callback)
def subscribe(self, channel_id):
self.socket.setsockopt(zmq.SUBSCRIBE, channel_id)
class MyWebSocket(WebSocketHandler):
def open(self):
self.pubsub = ZMQPubSub(self.on_data)
self.pubsub.connect()
self.pubsub.subscribe("session_id")
print 'ws opened'
def on_message(self, message):
print message
def on_close(self):
print 'ws closed'
def on_data(self, data):
print data
def main():
application = Application([(r'/channel', MyWebSocket)])
application.listen(10001)
print 'starting ws on port 10001'
ioloop.start()
if __name__ == '__main__':
main()
forwarder.py
import zmq
def main():
try:
context = zmq.Context(1)
frontend = context.socket(zmq.SUB)
frontend.bind('tcp://*:5559')
frontend.setsockopt(zmq.SUBSCRIBE, '')
backend = context.socket(zmq.PUB)
backend.bind('tcp://*:5560')
print 'starting zmq forwarder'
zmq.device(zmq.FORWARDER, frontend, backend)
except KeyboardInterrupt:
pass
except Exception as e:
logger.exception(e)
finally:
frontend.close()
backend.close()
context.term()
if __name__ == '__main__':
main()
publish.py
import zmq
if __name__ == '__main__':
context = zmq.Context()
socket = context.socket(zmq.PUB)
socket.connect('tcp://127.0.0.1:5559')
socket.send('session_id helloworld')
print 'sent data for channel session_id'
然而,我的
ZMQPubSub
类似乎根本没有接收到任何数据。我进一步尝试并意识到,在
ZMQPubSub
注册 on_recv
回调之后,需要调用 ioloop.IOLoop.instance().start()
。但是,这将阻塞执行。我还尝试将
main.ioloop
实例传递给 ZMQStream
构造函数,但也没有帮助。是否有一种方法可以将
ZMQStream
绑定到现有的 main.ioloop
实例,而不会阻止在 MyWebSocket.open
中的流程?
tornado==3.1
pyzmq==13.1.0
请将这两个版本的库安装到您的项目中。 - Abhinav Singh