在 python 中使用 rabbitmq 时出现“pika.exceptions.ConnectionClosed”错误

Question

我在 :http://www.rabbitmq.com/tutorials/tutorial-two-python.html 中使用“hello world”教程。 worker.py 是这样的

import pika
import time


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

channel.queue_declare(queue='task_queue', durable=True)
print ' [*] Waiting for messages. To exit press CTRL+C'

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

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

channel.start_consuming()

我已经在我的工作中使用了这个代码来实现。一切顺利，直到队列中出现一个点，在打印[x] Done后引发异常@

Traceback (most recent call last):
  File "hullworker2.py", line 242, in <module>
    channel.basic_consume(callback,queue='test_queue2')
  File "/usr/local/lib/python2.7/dist-packages/pika/channel.py", line 211, in basic_consume
    {'consumer_tag': consumer_tag})])
  File "/usr/local/lib/python2.7/dist-packages/pika/adapters/blocking_connection.py", line 904, in _rpc
    self.connection.process_data_events()
  File "/usr/local/lib/python2.7/dist-packages/pika/adapters/blocking_connection.py", line 88, in process_data_events
    if self._handle_read():
  File "/usr/local/lib/python2.7/dist-packages/pika/adapters/blocking_connection.py", line 184, in _handle_read
    super(BlockingConnection, self)._handle_read()
  File "/usr/local/lib/python2.7/dist-packages/pika/adapters/base_connection.py", line 300, in _handle_read
    return self._handle_error(error)
  File "/usr/local/lib/python2.7/dist-packages/pika/adapters/base_connection.py", line 264, in _handle_error
    self._handle_disconnect()
  File "/usr/local/lib/python2.7/dist-packages/pika/adapters/blocking_connection.py", line 181, in _handle_disconnect
    self._on_connection_closed(None, True)
  File "/usr/local/lib/python2.7/dist-packages/pika/adapters/blocking_connection.py", line 232, in _on_connection_closed
    self._channels[channel]._on_close(method_frame)
  File "/usr/local/lib/python2.7/dist-packages/pika/adapters/blocking_connection.py", line 817, in _on_close
    self._send_method(spec.Channel.CloseOk(), None, False)
  File "/usr/local/lib/python2.7/dist-packages/pika/adapters/blocking_connection.py", line 920, in _send_method
    self.connection.send_method(self.channel_number, method_frame, content)
  File "/usr/local/lib/python2.7/dist-packages/pika/adapters/blocking_connection.py", line 120, in send_method
    self._send_method(channel_number, method_frame, content)
  File "/usr/local/lib/python2.7/dist-packages/pika/connection.py", line 1331, in _send_method
    self._send_frame(frame.Method(channel_number, method_frame))
  File "/usr/local/lib/python2.7/dist-packages/pika/adapters/blocking_connection.py", line 245, in _send_frame
    super(BlockingConnection, self)._send_frame(frame_value)
  File "/usr/local/lib/python2.7/dist-packages/pika/connection.py", line 1312, in _send_frame
    raise exceptions.ConnectionClosed
pika.exceptions.ConnectionClosed

我不明白连接是如何在进程之间自动关闭的。对于队列中的 100 条消息，进程运行良好，然后突然出现此错误。 任何帮助表示赞赏。

Answer 1

有heartbeats 的概念。这基本上是服务器如何确保客户端仍然连接的一种方式。

当你这样做时

time.sleep( body.count('.') )

您将代码阻止了N 秒数。这意味着如果服务器想发送一个heartbeat 帧来检查您的客户端是否还活着，那么它不会得到响应，因为您的代码被阻塞并且不知道心跳是否到达。

您应该使用connection.sleep()，而不是使用time.sleep()，这也会使代码“休眠”N 秒数，但它也会与服务器通信并做出响应。

 sleep(duration)[source]

    A safer way to sleep than calling time.sleep() directly which will keep the adapter from ignoring frames sent from RabbitMQ. The connection will “sleep” or block the number of seconds specified in duration in small intervals.