RabbitMQ pika报StreamLostError连接被重置问题排查求助

• Publisher端死循环中使用time.sleep()阻塞了pika的心跳处理逻辑：pika的BlockingConnection依赖线程内定期调用连接相关方法完成心跳交互，标准库的time.sleep()会完全阻塞线程，导致pika无法向RabbitMQ broker发送心跳包，超时后broker会主动断开连接，触发Connection reset by peer报错。
• Publisher的send_message方法每次发送都新建channel，属于不必要的资源开销。
• Consumer端主动关闭心跳（parameters.heartbeat = 0）属于不规范配置，会导致broker无法检测死连接。

1. 修改Publisher端循环逻辑
   将main函数中的time.sleep(0.2)替换为你已经实现的basic_message_sender.pika_sleep(0.2)，pika.BlockingConnection.sleep()方法在等待期间会自动处理心跳事件，不会导致连接断连。

# 修改前
while True:
    time.sleep(0.2)
    check_stuff()

# 修改后
while True:
    basic_message_sender.pika_sleep(0.2)
    check_stuff()

2. 优化Publisher发送逻辑
   修改send_message方法，复用初始化时创建的channel，不需要每次发送都新建channel：

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

3. 调整心跳配置

• 给Publisher端的连接参数增加心跳配置：在BasicPikaClient的__init__方法中新增parameters.heartbeat = 30，和broker协商30秒心跳周期。
• 删除Consumer端的parameters.heartbeat = 0配置，同样设置为30秒即可，你的消费逻辑耗时不足1秒，完全不会阻塞心跳处理，不需要关闭心跳。

4. 可选兜底优化
   可以给Publisher的发送逻辑增加断连重连的异常捕获，避免偶发网络波动导致程序直接崩溃：

def send_message(self, exchange, routing_key, body):
    try:
        self.channel.basic_publish(exchange=exchange,
                              routing_key=routing_key,
                              body=body)
        print(f"Sent message. Exchange: {exchange}, Routing Key: {routing_key}, Body: {body}")
    except pika.exceptions.StreamLostError:
        # 断连后重新初始化连接和channel
        self.__init__(self.rabbitmq_broker_id, self.rabbitmq_user, self.rabbitmq_password, self.region)
        self.channel.basic_publish(exchange=exchange,
                              routing_key=routing_key,
                              body=body)

（注：需要在BasicPikaClient的__init__中把几个连接参数存为实例属性，方便重连时调用）

内容的提问来源于stack exchange，提问作者Frank