Pika处理RabbitMQ连接丢失：消费者模式下的异常处理方案

这事儿我做RabbitMQ消费者的时候踩过好几次坑，Pika的BlockingConnection默认不会自动处理连接断开的情况，一旦ConnectionClosed或者ChannelClosed异常抛出，消费者直接就停了。下面分享两种靠谱的解决思路，你可以根据场景选：

1. 利用Pika内置参数快速实现基础重连 #

Pika的ConnectionParameters自带了重连相关配置，能帮你搞定基础的自动重连逻辑，不用自己写太多循环：

import pika
from pika.exceptions import ConnectionClosed, ChannelClosed

QUEUE_NAME = "your_durable_queue"

def callback(ch, method, properties, body):
    # 这里写你的消息处理逻辑，记得一定要ACK！
    print(f"处理消息: {body.decode()}")
    ch.basic_ack(delivery_tag=method.delivery_tag)

def setup_consumer():
    # 配置连接参数，开启自动重连相关选项
    credentials = pika.PlainCredentials('rabbit_user', 'rabbit_pass')
    conn_params = pika.ConnectionParameters(
        host='localhost',
        port=5672,
        credentials=credentials,
        heartbeat=300,  # 心跳间隔，防止RabbitMQ因闲置断开连接
        connection_attempts=10,  # 最大重连尝试次数
        retry_delay=3,  # 每次重连的间隔秒数
        blocked_connection_timeout=120  # 连接阻塞时的超时时间
    )

    try:
        connection = pika.BlockingConnection(conn_params)
        channel = connection.channel()
        
        # 重连后必须重新声明队列（虽然是durable的，但新通道需要重新关联）
        channel.queue_declare(queue=QUEUE_NAME, durable=True)
        channel.basic_qos(prefetch_count=1)
        channel.basic_consume(queue=QUEUE_NAME, on_message_callback=callback)
        
        print("消费者启动成功，开始监听消息...")
        channel.start_consuming()
    except (ConnectionClosed, ChannelClosed) as e:
        print(f"连接中断: {str(e)}，正在尝试重连...")
        setup_consumer()  # 递归调用实现重连
    except Exception as e:
        print(f"发生无法恢复的错误: {str(e)}")

if __name__ == "__main__":
    setup_consumer()

这种方式适合简单场景，不用自己写复杂循环，但要注意递归重连的栈溢出问题（不过一般重连次数有限，影响不大）。

2. 手动实现带退避策略的重连循环（更灵活可控） #

如果你的场景需要精细控制（比如指数退避、重连失败告警、自定义重试逻辑），手动写循环是更好的选择：

import pika
import time
from pika.exceptions import ConnectionClosed, ChannelClosed, AMQPConnectionError

QUEUE_NAME = "your_durable_queue"

def callback(ch, method, properties, body):
    print(f"收到消息: {body.decode()}")
    # 业务逻辑要保证幂等，避免重连后重复处理消息
    ch.basic_ack(delivery_tag=method.delivery_tag)

def start_consumer():
    retry_count = 0
    max_retries = 10
    base_delay = 2  # 初始重连间隔
    
    while True:
        try:
            credentials = pika.PlainCredentials('rabbit_user', 'rabbit_pass')
            conn_params = pika.ConnectionParameters(
                host='localhost',
                credentials=credentials,
                heartbeat=300
            )
            
            connection = pika.BlockingConnection(conn_params)
            channel = connection.channel()
            
            channel.queue_declare(queue=QUEUE_NAME, durable=True)
            channel.basic_qos(prefetch_count=1)
            channel.basic_consume(queue=QUEUE_NAME, on_message_callback=callback)
            
            print("消费者已连接，开始消费消息")
            retry_count = 0  # 连接成功后重置重试计数
            channel.start_consuming()
            
        except (ConnectionClosed, ChannelClosed, AMQPConnectionError) as e:
            if retry_count >= max_retries:
                print(f"已达到最大重连次数({max_retries})，停止重试")
                break
            retry_count += 1
            # 指数退避：每次重连间隔翻倍，避免给RabbitMQ造成压力
            delay = base_delay * (2 ** (retry_count - 1))
            print(f"连接失败({retry_count}/{max_retries}): {str(e)}，{delay}秒后重试...")
            time.sleep(delay)
        except KeyboardInterrupt:
            print("用户中断，退出消费者")
            break
        except Exception as e:
            print(f"未知错误: {str(e)}，退出消费")
            break

if __name__ == "__main__":
    start_consumer()

这种方式的优势很明显：

• 可以控制最大重连次数，避免无限重试
• 指数退避策略能减少对RabbitMQ服务器的冲击
• 可以在重连失败时加入告警逻辑（比如发邮件、短信通知）

必须注意的几个细节 #

• 重连后要重新配置所有资源：每次重连都会创建新的连接和通道，之前的队列声明、QoS设置都会失效，必须重新执行。
• 保证消息处理的幂等性：连接断开后，未ACK的消息会被RabbitMQ重新投递，你的业务逻辑要能处理重复消息，比如用消息ID做去重。
• 合理设置心跳：RabbitMQ会主动断开长时间无交互的连接，设置heartbeat参数（比如300秒）能让Pika定期发送心跳包，维持连接。

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