如何在龙卷风应用程序中通信 RabbitMQ（Pika 库）

Question

Pika 库支持 tornado 适配器，here 是一个关于如何使用异步适配器发布消息的示例。

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

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

1 Pika使用tornado适配器有自己的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 Pika 示例一次又一次发布相同的消息，但我想发布请求数据，如何将请求数据传递给发布方法？

Answer 1

终于明白了！ 对于最新的 pika 组件，以前的解决方案已经过时了！

1.我的鼠兔版本是1.0.1。 警告 ： TornadoConnection 类已更改为最新推送请求的包。

from pika.adapters import tornado_connection

2.有一个例子：log()和config()应该忽略（删除）

import tornado.web

from handlers.notify import NotifyHandler
from function.function import config
from utils.utils import log

import pika
from pika.adapters import tornado_connection

HANDLERS = [(r'/notify', NotifyHandler)]

class PikaClient():
    def __init__(self, io_loop):
        self.io_loop = io_loop
        self.connected = False
        self.connecting = False
        self.connection = None
        self.channel = None
        self.message_count = 9


    def connect(self):
        if self.connecting:
            return
        self.connecting = True
        cred = pika.PlainCredentials('guest', 'guest')
        param = pika.ConnectionParameters(host="10.xxx.xxx.75", credentials=cred)
        self.connection = tornado_connection.TornadoConnection(param, custom_ioloop = self.io_loop, on_open_callback = self.on_connected)
        self.connection.add_on_open_error_callback(self.err)
        self.connection.add_on_close_callback(self.on_closed)


    def err(self, conn):
        log('socket error', conn)
        pass


    def on_connected(self, conn):
        log('connected')
        self.connected = True
        self.connection = conn
        self.connection.channel(channel_number = 1, on_open_callback = self.on_channel_open)


    def on_message(self, channel, method, properties, body):
        log(body)
        print('body : ', body)
        pass


    def on_channel_open(self, channel):
        self.channel = channel 
        channel.basic_consume(on_message_callback = self.on_message, queue='hello', auto_ack=True)
        return


    def on_closed(self, conn, c):
        log('pika close!')
        self.io_loop.stop()
        pass


def main():
    port = 3002
    is_debug = config('sys', 'debug')
    print('DEBUG', is_debug)
    app = tornado.web.Application(
            HANDLERS,
            debug = is_debug,
            )
    io_loop = tornado.ioloop.IOLoop.instance()
    app.pc = PikaClient(io_loop)
    app.pc.connect()
    http_server = tornado.httpserver.HTTPServer(app)
    app.listen(port)
    io_loop.start()
    print('listen {}'.format(port))


if __name__ == '__main__':
    main()

Answer 2

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

我在 rabbitmq 的漏洞中走得更远，为每个 websocket 提供了自己的一组通道和队列。

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

1. 将处理 Web 请求的 Tornado 应用程序。我在这里只看到长期存在的 websocket，但你也可以使用短期的 http 请求来实现。
2. PikaClient 实例持有一个（一个）RabbitMQ 连接
3. 在触发 open 方法时定义其通道、队列和交换的 Web 连接。

现在 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()