1

我正在使用 kombu 通过生产者/消费者模型来管理 RabbitMQ。我启动了我的生产者，它在一个队列中放置了 100 个作业（我只有一个队列和一个交换器）。我想同时启动多个消费者，并让每个消费者一次处理一项工作。不幸的是，消费者相互阻塞（即，当一个消费者从队列中获取工作时，其他消费者只是闲置）。如果我杀死了正在工作的消费者，那么其他消费者之一就会介入并开始工作。有没有办法让所有消费者同时运行，每个消费者处理队列中的不同作业？我的消费者代码如下：

def start_consumer(self, incoming_exchange_name):
    if self.rabbitmq_connection.connected:
        callbacks=[]
        queues=[]

        callbacks.append(self._callback)
        queues.append(self.incoming_queue)

        print 'opening a new *incoming* rabbitmq connection to the %s exchange for the %s queue' % (self.incoming_exchange.name, self.incoming_queue.name)
        self.incoming_exchange(settings.rabbitmq_connection).declare()
        self.incoming_queue(settings.rabbitmq_connection).declare()

        with settings.rabbitmq_connection.Consumer(queues=queues, callbacks=callbacks) as consumer:
            while True:
                try:
                    self.rabbitmq_connection.drain_events()
                except Exception as e:
                    print 'Error -> %s' % e.message
4

2 回答 2

5

您需要将消费者预取设置为 1（https://kombu.readthedocs.org/en/latest/reference/kombu.transport.pyamqp.html#kombu.transport.pyamqp.Connection.Channel.basic_qos），这样每个消费者将只获取 1 条消息，并将其余消息留在队列中并保持状态就绪，因此如果您有 2 个 QOS 设置为 1 的消费者并且您有 100 条消息，您将同时处理 2 个任务。

我已将缺少的部分添加到您的代码中，以设置预取计数

def start_consumer(self, incoming_exchange_name):
if self.rabbitmq_connection.connected:
    callbacks=[]
    queues=[]

    callbacks.append(self._callback)
    queues.append(self.incoming_queue)

    print 'opening a new *incoming* rabbitmq connection to the %s exchange for the %s queue' % (self.incoming_exchange.name, self.incoming_queue.name)
    self.incoming_exchange(settings.rabbitmq_connection).declare()
    self.incoming_queue(settings.rabbitmq_connection).declare()

    channel = self.rabbitmq_connection.channel()
    channel.basic_qos(prefetch_size=0, prefetch_count=1, a_global=False)

    with settings.rabbitmq_connection.Consumer(queues=queues, callbacks=callbacks, channel=channel) as consumer:
        while True:
            try:
                self.rabbitmq_connection.drain_events()
            except Exception as e:
                print 'Error -> %s' % e.message
于 2014-09-19T02:48:00.233 回答
-1

我认为您实际上是在尝试自己重写 Celery：

http://www.celeryproject.org/

除非您纯粹是为了学习而这样做，否则请避免痛苦并使用 Celery。顺便说一句，kombu这RabbitMQ正是 Celery 用作后端的东西（更不用说 Redis 后端可用，这在某些应用程序中为我节省了数小时的精力）。

于 2014-05-30T08:50:30.220 回答