我正在使用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
最佳答案
您需要将使用者预取设置为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
关于python - Python Kombu-阻止,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/23947119/