如何在龙卷风应用程序中传达RabbitMQ(Pika库)

Question

Pika库支持龙卷风适配器,这里有一个关于如何使用异步适配器发布消息的示例.

我想在龙卷风应用中使用鼠兔,只是一个例子,我想把龙卷风请求数据放到RabbitMQ,但不知道该怎么做.

两个问题不知道如何解决.

1 Pika使用龙卷风适配器有自己的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

在我搜索完全相同的东西时,我发现了Kevin Jing Qiu的这篇博文.

我进一步去了rabbitmq洞,为每个websocket提供了他自己的一套频道和队列.

我项目的摘录可以在下面找到.绑定到RabbitMQ的龙卷风应用程序包含以下部分:

1. 将处理Web请求的Tornado应用程序.我在这里只看到长寿的websockets,但你也可以使用短暂的http请求.
2. 由PikaClient实例保存的(一)RabbitMQ连接
3. 触发open方法时定义其通道,队列和交换的Web连接.

现在,websocket连接可以通过on_message从龙卷风(来自浏览器的数据)接收数据并将其发送到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()