Rabbitmq 在发送 ack 时断开连接

Question

我的 rabbitmq 消费者处理传入的消息，如下例所示： http://pika.readthedocs.org/en/latest/examples/asynchronous_consumer_example.html

我的代码和示例之间的区别在于，在我的代码中，每条消息都需要大量时间来处理。 （一个多小时）。

当消息完成处理时，执行以下行：

self._channel.basic_ack(delivery_tag)

但是在那之后，我收到一个 on_connection_closed 事件，这会导致我的消费者重新连接。

问题是ack没有传递，已经处理好的消息被发送给另一个消费者，再次处理。

任何帮助将不胜感激。

Answer 1

经过长时间的研究，发现了问题所在。

显然，尽管消费者实现具有异步回调功能，但它们都是从同一个线程调用的。

如果“on_message”方法被长时间执行，它会阻止线程向服务器发送心跳消息，导致它在一段时间后关闭连接。

解决方案是将消息处理移动到单独的线程。然而，有一个问题。为了防止消费者同时处理多个消息，必须在 on_channel_open 方法中添加以下行：

channel.basic_qos(prefetch_count=self._num_of_cores)