RabbitMQ/Pika连接重置异常：长任务ACK失败及重连问题求助

1. 手动维护心跳，避免阻塞事件循环 #

Pika同步客户端的心跳依赖事件循环处理，长任务会卡住start_consuming()的事件循环，导致心跳包无法及时发送，最终被RabbitMQ断开连接。解决办法是在长任务执行过程中，定期调用connection.process_data_events()处理心跳和AMQP事件。

修改你的generate_summary方法，在长任务中插入心跳处理：

def generate_summary(self, ch, method, properties, body):
    # 解析消息体
    data = self.parse_body(body)
    
    # 将长任务拆分为小片段，插入心跳处理
    summary = ""
    chunks = split_into_small_chunks(data)  # 自定义拆分逻辑，把大任务拆成小步骤
    for chunk in chunks:
        summary += process_chunk(chunk)  # 处理单个小片段
        # 处理心跳事件，time_limit设为0.1秒避免阻塞任务
        self._rabbitmq_connection.process_data_events(time_limit=0.1)
    
    # 尝试确认消息
    try:
        ch.basic_ack(delivery_tag=method.delivery_tag)
    except (pika.exceptions.ChannelWrongStateError, pika.exceptions.AMQPConnectionError):
        # 连接已断开，直接放弃确认，让RabbitMQ自动重新投递消息
        pass

2. 配置死信队列处理未确认消息 #

连接断开后，原通道的delivery_tag已失效，重连后调用basic_ack完全无效。正确做法是通过死信队列（DLQ）让RabbitMQ自动重试未确认消息：

1. 初始化时配置死信队列：

def setup_queues(self):
    # 声明死信交换机与队列
    self._rabbitmq_channel.exchange_declare(exchange='summary_dlx', exchange_type='direct')
    self._rabbitmq_channel.queue_declare(queue='summary_dlq', durable=True)
    self._rabbitmq_channel.queue_bind(exchange='summary_dlx', queue='summary_dlq', routing_key='summary_rk')
    
    # 为业务队列绑定死信规则
    queue_args = {
        'x-dead-letter-exchange': 'summary_dlx',
        'x-dead-letter-routing-key': 'summary_rk',
        'x-message-ttl': 60000  # 消息1分钟后自动重新投递
    }
    self._rabbitmq_channel.queue_declare(queue=self._rabbitmq_queue, durable=True, arguments=queue_args)

2. 修改消费逻辑，失败时直接让消息进入死信队列：

def generate_summary(self, ch, method, properties, body):
    try:
        # 执行长任务
        summary = create_summary(body)
        ch.basic_ack(delivery_tag=method.delivery_tag)
    except Exception as e:
        # 任务失败或连接断开，拒绝消息并转入死信队列
        ch.basic_nack(delivery_tag=method.delivery_tag, requeue=False)

3. 切换到异步Pika客户端（aio-pika） #

异步客户端可以在执行长任务的同时，不阻塞心跳与事件循环，从根源上避免超时问题。示例代码：

import asyncio
import aio_pika

class CreateSummaryAsync:
    def __init__(self, queue_name):
        self.queue_name = queue_name
        self.connection = None
        self.channel = None

    async def connect(self):
        self.connection = await aio_pika.connect_robust(
            "amqp://guest:guest@localhost/"
        )
        self.channel = await self.connection.channel()
        await self.channel.set_qos(prefetch_count=1)
        queue = await self.channel.declare_queue(self.queue_name, durable=True)
        await queue.consume(self.generate_summary)

    async def generate_summary(self, message: aio_pika.IncomingMessage):
        async with message.process():
            # 异步执行长任务，不阻塞心跳
            summary = await async_create_summary(message.body)
            # 上下文管理器自动处理ack，失败会触发重新投递

async def main():
    consumer = CreateSummaryAsync("summary_queue")
    await consumer.connect()
    await asyncio.Future()  # 保持进程运行

if __name__ == "__main__":
    asyncio.run(main())

4. 任务拆分与流水线处理 #

把长耗时的摘要生成拆分为多个阶段，每个阶段作为独立消息发送到RabbitMQ队列：

• 阶段1：解析输入数据，发送到parse_queue
• 阶段2：处理数据片段，发送到process_queue
• 阶段3：合并片段生成最终摘要，发送到summary_queue

每个阶段任务时间缩短，不会超过心跳超时阈值，从根本上避免连接被断开。

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