RabbitMQ消费者连接在90秒空闲后死亡

Question

我有一个RabbitMQ任务队列和一个Pika消费者来消费这些任务(带ack)。问题是连接在90秒空闲后就会死亡，但我的任务通常需要更长的时间。这意味着，当任务仍在计算时，它们将返回到任务队列中，并且永远不会进行打包。

使用RabbitMQ 3.5.3和Pika 0.9.14使用channel.basic_consume()方法。连接heartbeat_interval为30秒。

使用代码:

import pika
from time import sleep
RABBITMQ_URL = "amqp://user:[e-mail]/my_virtual_host?heartbeat_interval=30"
QUEUE_NAME = "my_queue"

def callback(ch, method, properties, body):
    print body
    sleep(91)  # if sleep value < 90 this code works (even 89)
    ch.basic_ack(delivery_tag=method.delivery_tag)

parameters = pika.URLParameters(RABBITMQ_URL)
connection = pika.BlockingConnection(parameters)
channel = connection.channel()
channel.queue_declare(queue=QUEUE_NAME, durable=True)
channel.basic_qos(prefetch_count=1)
channel.basic_consume(callback, queue=QUEUE_NAME)
channel.start_consuming()

回溯:

Traceback (most recent call last):
  File "main.py", line 19, in <module>
    channel.basic_consume(callback, queue=QUEUE_NAME)
  File "/usr/local/lib/python2.7/site-packages/pika/channel.py", line 221, in basic_consume
    {'consumer_tag': consumer_tag})])
  File "/usr/local/lib/python2.7/site-packages/pika/adapters/blocking_connection.py", line 1143, in _rpc
    self.connection.process_data_events()
  File "/usr/local/lib/python2.7/site-packages/pika/adapters/blocking_connection.py", line 240, in process_data_events
    if self._handle_read():
  File "/usr/local/lib/python2.7/site-packages/pika/adapters/blocking_connection.py", line 347, in _handle_read
    if self._read_poller.ready():
  File "/usr/local/lib/python2.7/site-packages/pika/adapters/blocking_connection.py", line 43, in inner
    return f(*args, **kwargs)
  File "/usr/local/lib/python2.7/site-packages/pika/adapters/blocking_connection.py", line 89, in ready
    self.poll_timeout)
select.error: (9, 'Bad file descriptor')

Answer 1

这里的问题是，因为你睡了很长时间，pika无法响应来自RabbitMQ的心跳请求，当这种情况发生时，RabbitMQ将关闭连接。

解决这个问题的唯一方法是禁用心跳或以较小的间隔睡眠并连续运行process_data_events()，以便pika可以处理心跳。

。

def amqp_sleep(connection, time_to_sleep=20):
    remaining = time_to_sleep
    while remaining > 0:
        connection.process_data_events()
        time.sleep(5)
        remaining -= 5

就我个人而言，我会选择一个自动处理后台心跳的库，这样你就不必处理它们了，例如rabbitpy或我自己的amqp-storm。