用非阻塞方式使用Kombu

Question

我正在寻找一种方法将kombu用作Tornado-SockJS和Django应用服务器之间的消息队列适配器。我已经尝试了以下代码：

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执行时被阻止了（正如预期的那样）。

有没有什么办法可以防止这种情况发生？

我知道Pika库适配器适用于Tornado，但我想使用kombu，因为它已经在项目中使用，并且具有灵活的传输方式。

更新：

将_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)

- Dmitry Demidenko

Answer 1

最终我找到了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)

对于其他后端，您可以使用在问题中发布的解决方案。

注意：不适用于librabbitmq。

- Dmitry Demidenko

---

注意：在撰写本文时，最新的py-amqp和kombu尚未正式支持异步读取。请参见此处。但是，有一些线索可以用于异步消费。 - Realistic

Answer 2

我有类似的需求，需要在Kombu/RabbitMQ和ZeroMQ之间以非阻塞的方式切换。解决方案是使用Gevent对socket库进行猴子补丁，使得Kombu也变为非阻塞模式。我的“主”线程运行了Kombu drain_events回调函数，在另一个Gevent线程中，我有一个循环从ZeroMQ socket接收消息。效果很好。

这也不适用于librabbitmq，因为它在C中执行自己的socket操作，不受Gevent影响。

- aaa90210