RabbitMQ pika.exceptions.ConnectionClosed

Question

我在 RabbitMQ 中有一个任务队列，其中有多个生产者 (12) 和一个消费者，用于 webapp 中的繁重任务。当我运行消费者时，它会在出现此错误之前开始将一些消息出列：

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)

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

Answer 1

您的 RabbitMQ 日志显示了一条我认为我们可能会看到的消息：missed&nbsp;heartbeats&nbsp;from&nbsp;client,&nbsp;timeout:&nbsp;60s发生的事情是您long_api_call阻止了 Pika 的 I/O 循环。Pika 是一个非常轻量级的库，不会在后台为您启动线程，因此您必须以这样一种方式编写代码，即不会阻塞 Pika 的 I/O 循环超过心跳间隔。RabbitMQ 认为您的客户端已死或无响应并强行关闭连接。请在此处查看我的答案，其中链接到此示例代码，显示如何在单独的线程中正确执行长时间运行的任务。您仍然可以使用no_ack=True，您只需跳过ack_message呼叫。