RabbitMQ Pika多线程长任务消费如何实现5分钟空闲自动关闭

• 业务要求：消费者连续5分钟未收到新消息时自动关闭，单条消息处理属于长耗时任务，单任务执行时长15-30分钟
• 现有实现：基于Pika多线程消费逻辑实现了长任务正常消费，但整合inactivity_timeout空闲超时参数后，消费部分消息就会触发连接关闭报错，无法达到预期效果

当前问题代码如下：

import time
import pika
import threading
import functools
from typing import Callable

class RabbitMQClient:
    def __init__(self, host: str, port: int, user: str, password: str, heartbeat: int = 60, **kwargs):
        self.__credentials = {"host": host, "port": port, "user": user, "password": password}
        self.heartbeat = heartbeat
        self.__consumer_handler = None
        self.connection = self.__create_connection(**self.__credentials, heartbeat=self.heartbeat)
        self.channel = self.connection.channel()

    def __del__(self):
        try:
            self.connection.close()
        except:
            pass

    # Create new connection
    @staticmethod
    def __create_connection(**kwargs):
        param = pika.ConnectionParameters(kwargs.get("host"), kwargs.get("port"), '/',
                                          pika.PlainCredentials(kwargs.get("user"), kwargs.get("password")),
                                          heartbeat=kwargs.get("heartbeat"))
        return pika.BlockingConnection(param)


     @staticmethod
    def __ack_message(channel, delivery_tag, ack):
        """Note that `channel` must be the same pika channel instance via which
        the message being ACKed was retrieved (AMQP protocol constraint).
        """
        if channel.is_open:
            if ack:
                channel.basic_ack(delivery_tag)
            else:
                channel.basic_nack(delivery_tag, requeue=False)
        else:
            pass

    def __on_message(self, channel, method_frame, properties, body, args):
        def do_work():
            delivery_tag, binding_key = method_frame.delivery_tag, method_frame.routing_key
            # Simulating a long running task here
            time.sleep(1800)
            cb = functools.partial(self.__ack_message, channel, delivery_tag, ack=True)
            connection.add_callback_threadsafe(cb)

        (connection, threads) = args
        t = threading.Thread(target=do_work)
        t.start()
        threads.append(t)  

    def consume_data(self, queue: str, exchange: str):
            channel = self.channel
            channel.exchange_declare(exchange=exchange, exchange_type='direct')

            threads = []
            for method, properties, body in channel.consume(queue, auto_ack=False, inactivity_timeout=900):
                if body:
                    on_message_callback = functools.partial(self.__on_message, args=(self.connection, threads))
                    on_message_callback(channel, method, properties, body)
                else:
                    print('breaking')
                    break

            # Wait for all to complete
            for thread in threads:
                thread.join()

            channel.cancel()

• 心跳保活失效：BlockingConnection的IO事件处理全部在主线程执行，长任务运行期间如果主线程阻塞在消费等待逻辑，没有及时处理IO事件，会导致心跳包无法按时发送，Broker主动断开连接
• 超时退出逻辑不合理：inactivity_timeout触发时仅判断了是否有新消息，没有判断当前是否还有未完成的长任务，直接break退出循环后，工作线程执行完回调ack时连接已经被关闭，触发报错
• 线程资源未清理：已完成的工作线程一直存在于线程列表中，既会造成内存泄漏，也无法准确判断当前是否还有运行中的任务
• 退出顺序错误：现有逻辑先break消费循环，再join线程，最后才调用channel.cancel()，顺序颠倒会导致消费者取消前连接可能已经被回收

核心修复点：

1. 将inactivity_timeout设置为300秒（即5分钟）匹配业务需求
2. 每次消费循环迭代主动调用process_data_events()处理IO事件，保证心跳正常发送
3. 增加运行中任务计数，仅当连续5分钟无新消息且无运行中任务时才触发退出
4. 每次循环自动清理已完成的线程，准确统计活跃任务数
5. 调整退出顺序：先取消消费者、停止接收新消息，再等待所有剩余任务执行完成，最后关闭连接

import time
import pika
import threading
import functools

class RabbitMQClient:
    def __init__(self, host: str, port: int, user: str, password: str, heartbeat: int = 60, **kwargs):
        self.__credentials = {"host": host, "port": port, "user": user, "password": password}
        self.heartbeat = heartbeat
        self.connection = self.__create_connection(**self.__credentials, heartbeat=self.heartbeat)
        self.channel = self.connection.channel()
        # 活跃任务计数
        self.active_task_count = 0

    def __del__(self):
        try:
            if self.connection.is_open:
                self.connection.close()
        except:
            pass

    @staticmethod
    def __create_connection(**kwargs):
        param = pika.ConnectionParameters(
            kwargs.get("host"), 
            kwargs.get("port"), 
            '/',
            pika.PlainCredentials(kwargs.get("user"), kwargs.get("password")),
            heartbeat=kwargs.get("heartbeat"),
            # 阻塞IO超时设置，避免主线程长时间卡死导致心跳断连
            blocked_connection_timeout=300
        )
        return pika.BlockingConnection(param)

    @staticmethod
    def __ack_message(channel, delivery_tag, ack):
        if channel.is_open:
            if ack:
                channel.basic_ack(delivery_tag)
            else:
                channel.basic_nack(delivery_tag, requeue=False)

    def __on_message(self, channel, method_frame, properties, body, args):
        def do_work():
            delivery_tag = method_frame.delivery_tag
            try:
                # 替换为实际业务处理逻辑，此处模拟30分钟长任务
                time.sleep(1800)
                cb = functools.partial(self.__ack_message, channel, delivery_tag, ack=True)
                self.connection.add_callback_threadsafe(cb)
            finally:
                # 任务完成后活跃计数减1
                self.active_task_count -= 1

        (connection,) = args
        t = threading.Thread(target=do_work, daemon=True)
        t.start()
        # 新任务启动，活跃计数加1
        self.active_task_count += 1

    def consume_data(self, queue: str, exchange: str):
        channel = self.channel
        channel.exchange_declare(exchange=exchange, exchange_type='direct', durable=True)
        channel.queue_declare(queue=queue, durable=True)
        channel.queue_bind(queue=queue, exchange=exchange, routing_key=queue)
        # 预取数设置为1，避免消息堆积在本地
        channel.basic_qos(prefetch_count=1)

        threads = []
        try:
            # inactivity_timeout设置为300秒即5分钟空闲超时
            for method, properties, body in channel.consume(queue, auto_ack=False, inactivity_timeout=300):
                # 主动处理IO事件，保证心跳正常
                self.connection.process_data_events()
                # 清理已完成的线程
                threads = [t for t in threads if t.is_alive()]

                if body:
                    # 收到新消息，启动工作线程处理
                    on_message_callback = functools.partial(self.__on_message, args=(self.connection,))
                    t = threading.Thread(target=on_message_callback, args=(channel, method, properties, body), daemon=True)
                    t.start()
                    threads.append(t)
                else:
                    # 空闲超时触发，判断是否还有活跃任务
                    if self.active_task_count == 0:
                        print("连续5分钟无新消息且无运行中任务，退出消费")
                        break
        finally:
            # 先取消消费者，停止接收新消息
            if channel.is_open:
                channel.cancel()
            # 等待所有运行中任务执行完成
            for thread in threads:
                thread.join()
            # 关闭连接
            if self.connection.is_open:
                self.connection.close()

注意：如果实际部署中单节点消费者并发数较高，可根据服务器配置调整basic_qos的预取数值，不要设置过大避免本地内存溢出。

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