Rabbitmq 与 python 的作品出乎意料答案

【问题标题】：Rabbitmq with python works unexpectedRabbitmq 与 python 的作品出乎意料
【发布时间】：2016-03-27 14:42:48
【问题描述】：

我正在和 rabbitmq 一起做一些练习。

但rabbmitmq的行为与官网教程不同。

worker 和task_sender 使用以下代码连接rabbitmq。

connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()
channel.queue_declare(queue='task_queue')

和task_sender通过调用发送任务

for i in range(10):
    message = "job%s %d %s" % (str(random.randint(1,10)), i , '.'*i)
    channel.basic_publish(exchange='',
                      routing_key='task_queue',
                      body=message,
                      properties=pika.BasicProperties(
                         delivery_mode = 2, # make message persistent
                      ))

和worker通过调用获取任务并等待一段时间。

def callback(ch, method, properties, body):
    print(" [x] Received %r" % body)
    time.sleep(body.count(b'.'))
    print(" [x] Job Done!")
    ch.basic_ack(delivery_tag = method.delivery_tag)


channel.basic_consume(callback,
                      queue='task_queue',
                      no_ack=True)

print(' [*] Waiting for messages. To exit press CTRL+C')
channel.start_consuming()

首先我运行task_sender.py 发送十个作业，并且进展顺利。

但是当我在不同的 shell 中启动两个 worker.py 时，似乎只有一个工作人员在获取任务，而另一个工作人员什么也不做。

更重要的是，当正在工作的worker 完成队列中的所有作业时，我再次运行task_sender.py 以发送新任务，workers 不再获取作业。

看来rabbitmq正在阻塞，我该如何解决？

这是我的Rabbitmq status

欢迎任何帮助，在此先感谢。

【问题讨论】：

标签： python rabbitmq message-queue

【解决方案1】：

在您的工作人员中，尝试将 prefetch_count 设置为合理的值并将 no_ack 设置为 false：

...
channel.basic_qos(prefetch_count=1)
channel.basic_consume(callback,
                  queue='task_queue',
                  no_ack=False)
...

【讨论】：

它有效，我会检查文档以获得更多详细信息，非常感谢。

【解决方案2】：

我不确定 pika，但我在使用 oslo.messaging 和 rabbitmq 作为后端时遇到了同样的问题。发生的事情是，通知消息的生产者发送了它，第一个捕获消息的侦听器（消费者）消费了它，而其他侦听器从未得到它。

对于 oslo，有一个功能“扇出”，它将通知消息发送到所有服务器，而不是“第一个赢家，其他输家”的风格。

我认为 pika 可能也有类似的东西。

Oslo notification server documentation

尝试在此处搜索关键字“fanout”。然后你可以检查 pika 文档是否有类似的东西。希望这会有所帮助。

【讨论】：