非阻塞方式的Kombu

我正在尋找一種將kombu用作tornado-sockjs和Django應用程序服務器之間的MQ適配器的方法。我做了這樣的事情：非阻塞方式的Kombu

class BrokerClient(ConsumerMixin): 
    clients = [] 

    def __init__(self): 
     self.connection = BrokerConnection(settings.BROKER_URL) 
     self.io_loop = ioloop.IOLoop.instance() 
     self.queue = sockjs_queue 
     self._handle_loop() 

    @staticmethod 
    def instance(): 
     if not hasattr(BrokerClient, '_instance'): 
      BrokerClient._instance = BrokerClient() 
     return BrokerClient._instance 

    def add_client(self, client): 
     self.clients.append(client) 

    def remove_client(self, client): 
     self.clients.remove(client) 

    def _handle_loop(self): 
     try: 
      if self.restart_limit.can_consume(1): 
       for _ in self.consume(limit=5): 
        pass 
     except self.connection.connection_errors: 
      print ('Connection to broker lost. ' 
      'Trying to re-establish the connection...') 
     self.io_loop.add_timeout(datetime.timedelta(0.0001), self._handle_loop) 

    def get_consumers(self, Consumer, channel): 
     return [Consumer([self.queue, ], callbacks=[self.process_task])] 

    def process_task(self, body, message): 
     for client in self.clients: 
      if hasattr(body, 'users') and client.user.pk in body.users: 
       client.send(body) 
     message.ack()

但是，在_handle_loop執行時（如預期的）龍捲風被阻止。

有什麼辦法可以防止這種情況發生？

我知道Tornado的Pika庫適配器，但我想使用kombu，因爲它已用於項目並具有靈活的傳輸。

UPDATE：

更改_handle_loop到發生器功能

def drain_events(self, callback): 
    with self.Consumer() as (connection, channel, consumers): 
     with self.extra_context(connection, channel): 
      try: 
       connection.drain_events(timeout=1) 
      except: 
       pass 
    callback(None) 


@tornado.gen.engine 
def _handle_loop(self): 
    response = yield tornado.gen.Task(self.drain_events) 
    self.io_loop.add_timeout(datetime.timedelta(0.0001), self._handle_loop)

來源

2013-04-01 Dmitry Demidenko

最後我發現RabbitMQ的後端正確的解決方案：

class BrokerClient(object): 
    clients = [] 

    @staticmethod 
    def instance(): 
     if not hasattr(BrokerClient, '_instance'): 
      BrokerClient._instance = BrokerClient() 
     return BrokerClient._instance 

    def __init__(self): 
     self.connection = BrokerConnection(settings.BROKER_URL) 
     self.consumer = Consumer(self.connection.channel(), [queue, ], callbacks=[self.process_task]) 
     self.consumer.consume() 
     io_loop = tornado.ioloop.IOLoop.instance() 
     for sock, handler in self.connection.eventmap.items(): 
      def drain_nowait(fd, events): 
       handler() 
      io_loop.add_handler(sock.fileno(), drain_nowait, l.READ | l.ERROR) 

    def process_task(self, body, message): 
     #something 
     message.ack() 

    def add_client(self, client): 
     self.clients.append(client) 

    def remove_client(self, client): 
     self.clients.remove(client)

對於其他後端，您可以使用發佈的解決方案問題

注：不適用於天秤座bbitmq

來源

2013-04-11 17:37:06

注意：在撰寫本文時，最新的py-amqp和kombu並不正式支持異步讀取。 [見此]（https://github.com/celery/py-amqp/issues/25）。但是，對於異步消費，有[breadcrumbs]（https://github.com/celery/kombu/blob/master/examples/experimental/async_consume.py） – Realistic

我有類似的需求，在Kombu/RabbitMQ和ZeroMQ之間以非阻塞方式切換。解決方案是使用Gevent對套接字庫進行修補，使Kombu變得非阻塞。我的「主」線程運行Kombu drain_events回調，在另一個gevent線程中，我有一個從ZeroMQ套接字接收消息的循環。工作得很好。

這也不會與librabbitmq一起工作，因爲它在C中沒有受到Gevent的影響。

來源

2014-10-01 00:45:58 aaa90210

非阻塞方式的Kombu

回答

相關問題