RabbitMQ pika.exceptions.ConnectionClosed (-1, "error(104, 'Connection reset by peer')")

Question

我在 RabbitMQ 中有一个任务队列，其中有多个生产者 (12) 和一个消费者，用于在 Web 应用程序中执行繁重的任务。当我运行消费者时，它开始使一些消息出队，然后崩溃并出现此错误：

Traceback (most recent call last):
File "jobs.py", line 42, in <module> jobs[job](config)
File "/home/ec2-user/project/queue.py", line 100, in init_queue
channel.start_consuming()
File "/usr/lib/python2.7/site-packages/pika/adapters/blocking_connection.py", line 1822, in start_consuming
self.connection.process_data_events(time_limit=None)
File "/usr/lib/python2.7/site-packages/pika/adapters/blocking_connection.py", line 749, in process_data_events
self._flush_output(common_terminator)
File "/usr/lib/python2.7/site-packages/pika/adapters/blocking_connection.py", line 477, in _flush_output
result.reason_text)
pika.exceptions.ConnectionClosed: (-1, "error(104, 'Connection reset by peer')")

生产者代码是：

message = {'image_url': image_url, 'image_name': image_name, 'notes': notes}

connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()

channel.queue_declare(queue='tasks_queue')
channel.basic_publish(exchange='', routing_key=queue_name, body=json.dumps(message))

connection.close()

以及唯一的消费者代码（那个是冲突的）：

def callback(self, ch, method, properties, body):
    """Callback when receive a message."""
    message = json.loads(body)
    try:
        image = _get_image(message['image_url'])
    except:
        sys.stderr.write('Error getting image in note %s' % note['id'])
   # Crop image with PIL. Not so expensive
   box_path = _crop(image, message['image_name'], box)

   # API call. Long time function
   result = long_api_call(box_path)

   if result is None:
       sys.stderr.write('Error in note %s' % note['id'])
       return
   # update the db
   db.update_record(result)


connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()
channel.queue_declare(queue='tasks_queue')
channel.basic_qos(prefetch_count=1)
channel.basic_consume(callback_obj.callback, queue='tasks_queue', no_ack=True)
channel.start_consuming()

如您所见，消息有 3 个昂贵的函数。一项裁剪任务、一次 API 调用和一次数据库更新。没有 API 调用，查询消费者运行很顺利。

提前致谢

Answer 1

您的 RabbitMQ 日志显示一条消息，我认为我们可能会看到：

missed heartbeats from client, timeout: 60s

你的 long_api_call 阻塞了 Pika 的 I/O 循环。 Pika 是一个非常 轻量级 的库，不会在后台为您启动线程，因此您必须以这样的方式进行编码，以免阻塞 Pika 的 I/O 循环的时间长于心跳间隔. RabbitMQ认为你的客户端挂了或者没有响应，强行关闭连接。

请参阅 my answer here which links to this example code 展示如何在单独的线程中正确执行长运行任务。您仍然可以使用 no_ack=True，您将跳过 ack_message 调用。

_{注意： RabbitMQ 团队监控 the rabbitmq-users mailing list 并且有时只在 Whosebug 上回答问题。}

Answer 2

Starting with RabbitMQ 3.5.5, the broker’s default heartbeat timeout decreased from 580 seconds to 60 seconds.

参见pika: Ensuring well-behaved connection with heartbeat and blocked-connection timeouts。

最简单的解决方法是增加心跳超时：

rabbit_url = host + "?heartbeat=360"
conn = pika.BlockingConnection(pika.URLParameters(rabbit_url))

# or

params = pika.ConnectionParameters(host, heartbeat=360)
conn = pika.BlockingConnection(params)

RabbitMQ pika.exceptions.ConnectionClosed (-1, "error(104, 'Connection reset by peer')")

RabbitMQ pika.exceptions.ConnectionClosed (-1, "error(104, 'Connection reset by peer')")

python

rabbitmq

python-2.7

pika

python-pika