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

Question

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()

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

Answer 1

有一个heartbeats的概念。它基本上是服务器可以确保客户端仍然连接的一种方式。

当你这么做的时候

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

阻止代码的时间n秒数。这意味着，如果服务器希望发送heartbeat帧来检查您的客户端是否还活着，那么它将无法返回响应，因为您的代码被阻塞，并且不知道heartbeat是否到达。

 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.