RabbitMQ破坏了管道错误或丢失了消息

Question

使用pika库的BlockingConnection连接到RabbitMQ，我偶尔会在发布消息时出错：

致命套接字错误：错误（32，'破管'）

这是一个非常简单的子流程，它从内存中的队列中获取一些信息，并将一个小的JSON消息发送到AMQP。当系统几分钟没有发送任何消息时，似乎只出现错误。

建立：

connection = pika.BlockingConnection(parameters)
channel = self.connection.channel()
channel.exchange_declare(
    exchange='xyz',
    exchange_type='fanout',
    passive=False,
    durable=True,
    auto_delete=False
)

入队代码捕获任何连接错误并重试：

def _enqueue(self, message_id, data):
    try:
        published = self.channel.basic_publish(
            self.amqp_exchange,
            self.amqp_routing_key,
            json.dumps(data),
            pika.BasicProperties(
                content_type="application/json",
                delivery_mode=2,
                message_id=message_id
            )
        )

        # Confirm delivery or retry
        if published:
            self.retry_count = 0
        else:
            raise EnqueueException("Message publish not confirmed.")

    except (EnqueueException, pika.exceptions.AMQPChannelError, pika.exceptions.AMQPConnectionError,
            pika.exceptions.ChannelClosed, pika.exceptions.ConnectionClosed, pika.exceptions.UnexpectedFrameError,
            pika.exceptions.UnroutableError, socket.timeout) as e:
        self.retry_count += 1
        if self.retry_count < 5:
            logging.warning("Reconnecting and resending")
            if self.connection.is_open:
                self.connection.close()
            self.connect()
            self._enqueue(message_id, data)
        else:
            raise e

这有时适用于第二次尝试。它经常挂起一段时间或者在最终抛出异常（possibly related bug report）之前抛弃消息。因为它只发生在系统安静几分钟时，我猜它是由于连接超时。但AMQP有一个心跳系统，据报道pika使用它（related bug report）。

为什么我会收到此错误或丢失邮件，为什么连接在不使用时不会保持打开状态？

Answer 1

从另一个bug report：

由于BlockingConnection不在后台处理心跳，并且heartbeat_interval无法覆盖服务器建议的心跳间隔（这也是一个错误），我建议默认情况下应禁用心跳（依赖于TCP keep-alive）。

如果处理消耗块中的任务需要更长的时间，然后服务器建议心跳间隔，则服务器将关闭连接，并且客户端在完成处理时将无法确认消息。

未发布的update可能有助于解决这个问题。

所以我实施了一个解决方法。每30秒我通过队列发布一个心跳消息。这样可以保持连接处于打开状态，并且可以向客户确认我的应用程序已启动并正在运行。

Answer 2

Broken Pipe错误意味着服务器在客户端关闭连接时尝试将某些内容写入套接字。

正如我所看到的，你有一些共享的“self.connection”可以在并行线程之前/之后关闭？

您还可以将日志级别设置为DEBUG并查看客户端日志以确定客户端关闭连接的时刻。