竜巻アプリケーションでRabbitMQ（Pikaライブラリ）を通信する方法

Question

Pikaライブラリサポート竜巻アダプタ、hereは、非同期アダプタを使用してメッセージを発行する方法の例です。

私は竜巻アプリケーションでpikaを使用したいと思います。ただの例ですが、私は竜巻リクエストデータをRabbitMQに入れたいが、どうやってそれを行うのか分からない。

2つの質問は解決する方法がわかりません。

1ピカ使用竜巻アダプタは、独自のioloopを持ち、

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

トルネード・アプリケーションは、これらの2 ioloopを結合する方法

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

、独自のioloopを持っていますか？

2 Pikaの例では、同じメッセージを何度も繰り返し公開していますが、要求データを公開して、要求データを公開メソッドに渡す方法はありますか？

Answer 1

全く同じことを私の検索では私はこれを見つけたblog post of Kevin Jing Qiu。

私は、すべてのwebsocketに独自のチャネルとキューのセットを与えるためにrabbitmqの穴をもう少し行きました。

私のプロジェクトの抜粋は以下の通りです。 RabbitMQに結合された竜巻アプリケーションは、次の部分で構成されています。

1. Webリクエストを処理するTornadoアプリケーション。私はここに長寿命のWebソケットしか見ることはできませんが、短期間のHTTPリクエストでもそうすることができます。
2. （1）PikaClientインスタンスによるRabbitMQ接続の保持
3. オープンメソッドがトリガーされるときのチャネル、キューおよびエクスチェンジを定義する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()