python - 如何在 Tornado 应用程序中与 RabbitMQ(鼠兔库)通信

Question

鼠兔库支持 Tornado 适配器，here是关于如何使用异步适配器发布消息的示例。

我想在tornado应用中使用pika，举个例子，我想把tornado请求数据放到RabbitMQ，但是不知道怎么做。

两个问题不知道怎么解决。

1 Pika使用tornado adapter有自己的ioloop，

self._connection = pika.SelectConnection(pika.URLParameters(self._url),  
                                         self.on_connection_open)  
self._connection.ioloop.start()

Tornado 应用程序有自己的 ioloop，

tornado.ioloop.IOLoop.instance().start()

如何结合这两个 ioloop？

2 鼠兔示例一次又一次发布相同的消息，但我想发布请求数据，如何将请求数据传递给发布方法？

Answer 1

在我搜索完全相同的东西时，我发现了这个 blog post of Kevin Jing Qiu .

我进一步研究了 rabbitmq 漏洞，为每个 websocket 提供了自己的一组 channel 和队列。

我的项目的摘录可以在下面找到。绑定(bind)到 RabbitMQ 的 tornado 应用程序由以下部分组成:

1. 将处理网络请求的 Tornado 应用程序。我在这里只看到长生命周期的 websockets，但你也可以用短生命周期的 http 请求来做到这一点。
2. 由 PikaClient 实例持有的(一个)RabbitMQ 连接
3. 在触发 open 方法时定义其 channel 、队列和交换的网络连接。

现在 websocket 连接可以通过 on_message 从 tornado 接收数据(来自浏览器的数据)并将其发送到 RabbitMQ。

websocket 连接将通过 basic_consume 从 RabbitMQ 接收数据。

这还没有完全发挥作用，但您应该明白了。

class PikaClient(object):

    def __init__(self, io_loop):
        logger.info('PikaClient: __init__')
        self.io_loop = io_loop

        self.connected = False
        self.connecting = False
        self.connection = None
        self.channel = None
        self.message_count = 0
    """ 
    Pika-Tornado connection setup
    The setup process is a series of callback methods.
    connect:connect to rabbitmq and build connection to tornado io loop -> 
    on_connected: create a channel to rabbitmq ->
    on_channel_open: declare queue tornado, bind that queue to exchange 
                     chatserver_out and start consuming messages. 
   """

    def connect(self):
        if self.connecting:
            #logger.info('PikaClient: Already connecting to RabbitMQ')
            return

        #logger.info('PikaClient: Connecting to RabbitMQ')
        self.connecting = True

        cred = pika.PlainCredentials('guest', 'guest')
        param = pika.ConnectionParameters(
            host='localhost',
            port=5672,
            virtual_host='/',
            credentials=cred
        )
        self.connection = TornadoConnection(param,
            on_open_callback=self.on_connected,stop_ioloop_on_close=False)
        self.connection.add_on_close_callback(self.on_closed)

    def on_connected(self, connection):
        logger.info('PikaClient: connected to RabbitMQ')
        self.connected = True
        self.connection = connection
        # now you are able to call the pika api to do things
        # this could be exchange setup for websocket connections to 
        # basic_publish to later.
        self.connection.channel(self.on_channel_open)

    def on_channel_open(self, channel):
        logger.info('PikaClient: Channel %s open, Declaring exchange' % channel)
        self.channel = channel

    def on_closed(self, connection):
        logger.info('PikaClient: rabbit connection closed')
        self.io_loop.stop()


class MyWebSocketHandler(websocket.WebSocketHandler):
    def __init__(self):
        self.status = 'not connected yet'

    def open(self, *args, **kwargs):
        self.status = "ws open"
        self.rabbit_connect() # connect this websocket object to rabbitmq

    def rabbit_connect():
        self.application.pc.connection.channel(self.rabbit_channel_in_ok)

    def rabbit_channel_in_ok(self,channel):
        self.channel_in = channel
        self.channel_in.queue_declare(self.rabbit_declare_ok,
                                      exclusive=True,auto_delete=True)


# and so on...


handlers = [ your_definitions_here_like_websockets_or_such ]
settings = { your_settings_here }
application = tornado.web.Application(handlers,**settings)

def main():
    io_loop = tornado.ioloop.IOLoop.instance()
    # PikaClient is our rabbitmq consumer
    pc = PikaClient(io_loop)
    application.pc = pc
    application.pc.connect()
    application.listen(config.tornadoport)
    try:
        io_loop.start()
    except KeyboardInterrupt:
        io_loop.stop()

if __name__ == '__main__':
    main()