RabbitMQ pika.exceptions.ConnectionClosed答案

【问题标题】：RabbitMQ pika.exceptions.ConnectionClosedRabbitMQ pika.exceptions.ConnectionClosed
【发布时间】：2016-09-16 04:59:22
【问题描述】：

我尝试使用 RabbitMQ 发送和接收消息。我没有计算机科学背景，我使用的术语可能不是很准确。

我尝试复制教程文件：提交我的html表单时，我的python脚本（cgi）消息正在提交到队列

connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))
        channel = connection.channel()
        channel.queue_declare(queue='task_queue', durable=True)
        message = PN
        channel.basic_publish(exchange='',
                              routing_key='task_queue',
                              body=message,
                              properties=pika.BasicProperties(
                                 delivery_mode = 2, # make message persistent
                              ))
        connection.close()

我的接收器正在运行：

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 Project %r" % body)
    #ch.basic_ack(delivery_tag = method.delivery_tag) 
    if not (os.path.isfile(js_path)):
        print (' [*] ERROR files missing ')
        #ch.basic_ack(delivery_tag = method.delivery_tag)
        return
    p= subprocess.Popen(run a subprocess here)
    p.wait()

    print (' [*] Temporary Files removed')
    print(" [*] Waiting for messages. To exit press CTRL+C")

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

它管理大部分时间，但随机崩溃并出现以下错误：

Traceback（最近一次调用最后一次）：文件“Receive5.py”，第 139 行，在 channel.start_sumption() 文件 "C:\Python27\lib\site-packages\pika\adapters\blocking_connection.py", 第 1681 行，开始消费 self.connection.process_data_events(time_limit=None) 文件 "C:\Python27\lib\site-packages\pika\adapters\blocking_connection.py", 第 647 行，在 process_data_events 中 self._flush_output(common_terminator) 文件 "C:\Python27\lib\site-packages\pika\adapters\blocking_connection.py", 第 426 行，在 _flush_output 中引发 exceptions.ConnectionClosed() pika.exceptions.ConnectionClosed

【问题讨论】：

您应该检查日志，由于某种原因，消费者的连接已关闭
看起来是一个“心跳”问题..¨=错误报告==== 20-May-2016::08:36:28 === 关闭 AMQP 连接...错过心跳客户端，超时：60s =INFO REPORT==== 20-May-2016::08:40:22 === 接受 AMQP 连接 ... =INFO REPORT==== 20-May-2016::08:44 :51 === 接受 AMQP 连接 ... =INFO REPORT==== 20-May-2016::08:44:51 === 关闭 AMQP 连接 ... =ERROR REPORT==== 20-May- 2016::08:46:22 === 关闭 AMQP 连接...错过来自客户端的心跳，超时：60 秒 =INFO REPORT==== 20-May-2016::09:38:51 === 接受 AMQP 连接...
通过将心跳设置为 0 解决的问题 ...
已经快两年了，我认为这个问题没有得到更好的处理或记录:(

标签： python rabbitmq pika

【解决方案1】：

这是因为你让主线程等待，因为这个 pika 无法处理传入的消息；在这种情况下，它在子进程完成之前无法响应心跳。这会导致 RabbitMQ 认为客户端已死并强制断开连接。

如果您希望它与心跳一起使用（推荐），您需要定期致电connection.process_data_events。这可以通过添加一个检查线程是否完成的循环来完成，并且每 30 秒左右调用一次process_data_events，直到线程完成。

【讨论】：

只是好奇，是 RabbitMQ 教程缺少这部分还是我对教程的使用不当？ (rabbitmq.com/tutorials/tutorial-two-python.html)
我会说这在很大程度上是无证的。为了避免这个问题，我创建了自己的 AMQP 库。

【解决方案2】：

看看添加这个https://github.com/mosquito/aio-pika

它是一个 asynchio 包装器，如果你了解 asynchr 背后的概念，它非常易于使用:)

【讨论】：

【解决方案3】：

它似乎没有在 127.0.0.1:5672 连接到 RabbitMQ -- 你确定 RabbitMQ 正在运行并在 127.0.0.1:5672 上侦听吗？

适用于 linux (debian) 用户

您可以通过输入此命令检查是否安装了 RabbitMQ。

sudo service rabbitmq-server status

如果您收到回复，则表示它已安装。检查它是否正在运行？

启动rabbitMQ服务类型

sudo service rabbitmq-server start

使用这个命令重启rabbitMq

sudo service rabbitmq-server restart

没有响应意味着您没有安装 Rabbitmq。通过键入安装它以下命令。

sudo apt-get update

sudo apt-get -y upgrade

sudo apt-get install rabbitmq-server

然后通过键入上面的启动命令来启动服务器。

【讨论】：

downwoting 因为答案完全不重要

【解决方案4】：

这里是关于如何避免因心跳而断开连接的 pika 文档。

https://pika.readthedocs.io/en/stable/examples/heartbeat_and_blocked_timeouts.html

在pika 0.11.2之前的版本中，虽然我们可以在pika.ConnectionParameters里面加一个参数：heartbeat_interval=600，但是如果服务端的心跳值短60s就无济于事了。只有在版本至少为0.11.2时才能工作

【讨论】：