我还在寻找答案，同时我重新连接到我的兔子服务器：

while True:
    try:
​
        connection = pika.BlockingConnection(params)
        channel = connection.channel() # start a channel
        channel.queue_declare(queue=amqp_q, durable=True) # Declare a queue
        ...
​
    except pika.exceptions.ConnectionClosed:
        print('connection closed... and restarted')

我有同样的问题与MySQL服务器托管...我来了解，它发生了，如果我们打开连接很长一段时间或未经修改的很长一段时间..如果您的程序打开数据库或任何东西，直到整个程序运行使它在这样一种方式，它打开数据库写入一切，关闭和重复
我不知道rabbitmq到底是什么，但我认为你写的标题错误可能是因为这个原因

我遇到了同样的错误（使用纯PIKA库），并试图通过Amazon MQ连接到RabbitMQ代理。
正确设置ssl配置后，问题解决。
请在此处查看完整的博客文章：https://docs.aws.amazon.com/amazon-mq/latest/developer-guide/amazon-mq-rabbitmq-pika.html

我使用的核心代码片段：

定义Pika客户端：

import ssl
import pika

class BasicPikaClient:

    def __init__(self, rabbitmq_broker_id, rabbitmq_user, rabbitmq_password, region):

        # SSL Context for TLS configuration of Amazon MQ for RabbitMQ
        ssl_context = ssl.SSLContext(ssl.PROTOCOL_TLSv1_2)
        ssl_context.set_ciphers('ECDHE+AESGCM:!ECDSA')

        url = f"amqps://{rabbitmq_user}:{rabbitmq_password}@{rabbitmq_broker_id}.mq.{region}.amazonaws.com:5671"
        parameters = pika.URLParameters(url)
        parameters.ssl_options = pika.SSLOptions(context=ssl_context)

        self.connection = pika.BlockingConnection(parameters)
        self.channel = self.connection.channel()

生产商：

from basicClient import BasicPikaClient

class BasicMessageSender(BasicPikaClient):

    def declare_queue(self, queue_name, durable):
        print(f"Trying to declare queue({queue_name})...")
        self.channel.queue_declare(queue=queue_name, durable=durable)

    def send_message(self, exchange, routing_key, body):
        channel = self.connection.channel()
        channel.basic_publish(exchange=exchange,
                              routing_key=routing_key,
                              body=body)
        print(f"Sent message. Exchange: {exchange}, Routing Key: {routing_key}, Body: {body}")

    def close(self):
        self.channel.close()
        self.connection.close()

调用生成器：

# Initialize Basic Message Sender which creates a connection

# and channel for sending messages.

basic_message_sender = BasicMessageSender(
    credentials["broker_id"],
    credentials["username"],
    credentials['password'],
    credentials['region']
)

# Declare a queue

basic_message_sender.declare_queue("q_name", durable=True)

# Send a message to the queue.

basic_message_sender.send_message(exchange="", routing_key="q_name", body=b'Hello World 2!')

# Close connections.

basic_message_sender.close()

定义消费者：

class BasicMessageReceiver(BasicPikaClient):

    def get_message(self, queue):
        method_frame, header_frame, body = self.channel.basic_get(queue)
        if method_frame:
            print(method_frame, header_frame, body)
            self.channel.basic_ack(method_frame.delivery_tag)
            return method_frame, header_frame, body
        else:
            print('No message returned')

    def close(self):
        self.channel.close()
        self.connection.close()

呼叫消费者：

# Create Basic Message Receiver which creates a connection

# and channel for consuming messages.

basic_message_receiver = BasicMessageReceiver(
    credentials["broker_id"],
    credentials["username"],
    credentials['password'],
    credentials['region']
)

# Consume the message that was sent.

basic_message_receiver.get_message("q_name")

# Close connections.

basic_message_receiver.close()

我希望以上有帮助。谢谢