在一个小项目中，我用到了 RabbitMQ 进行任务的异步处理。RabbitMQ 服务搭建在一个 VPS 服务上，客户端使用了 Python 的 aio_pika 库（一个支持异步的 RabbitMQ 客户端库）进行消息的发送和接收。

这个小项目只用到了两个队列，有 6 个消费者在消费，两个生产者发布消息，所以同一时间最多有 8 条连接。但随着时间推移，我发现 rabbitmq 服务连接数量越来越多，有时候竟然达到了 200 多条。事出反常必有妖，一开始我还以为是库本身的问题，心想难不成是 aio_pika 库的 bug，这么巧就撞上了？但仔细分析了一下代码，才发现原来是连接管理有问题。

项目中的使用方式

问题主要源于生产者和消费者在连接建立的异常处理路径中未能正确关闭连接。具体实现如下:

class BaseProducer:
    async def connect(self) -> None:
        try:
            if not self._is_connected:
                self.connection = await aio_pika.connect_robust(...)
                self.channel = await self.connection.channel()
                self.queue = await self.channel.declare_queue(...)
                self._is_connected = True
        except Exception as e:
            self._is_connected = False
            logger.error(f"Connection error: {str(e)}")
            # problem 1: no connection.close() called here
            # crucial_step: if self.connection and not self.connection.is_closed:
            #    await self.connection.close() # 显式关闭部分建立的连接

class BaseConsumer:
    async def connect(self) -> None:
        try:
            if not self._is_connected:
                ... # 省略了与 BaseProducer 类似的连接代码
        except Exception as e:
            self._is_connected = False
            logger.error(f"Connection error: {str(e)}")
            # problem 2: no connection.close() called here
            # crucial_step: if self.connection and not self.connection.is_closed:
            #    await self.connection.close()

    async def process_queue(self) -> typing.AsyncGenerator[bytes, None]:
        try:
            ...
        except Exception as e: # [*] 此处假设是连接级或通道级异常导致需要重置连接状态
            self._is_connected = False
            logger.error(f"Processing error leading to connection issue: {str(e)}")
            # problem 3: no connection.close() called here (if connection object exists and is open)
            # crucial_step: if self.connection and not self.connection.is_closed:
            #    await self.connection.close()

在上述 BaseProducer 的实现中，连接管理的逻辑大致如下：

通过 _is_connected 标志位判断是否已存在有效连接。
若未连接，则尝试创建新连接、通道并声明队列。
若连接过程中（包括创建连接、通道、声明队列等步骤）发生任何异常，则捕获异常，并将 _is_connected 设置为 False。

这个流程图清晰地揭示了问题的核心：当生产者在 connect 方法中遭遇异常时（例如网络抖动、RabbitMQ 服务暂时不可用等），代码仅仅是将 _is_connected 标志位重置为 False。然而，已经建立的（或者说尝试建立但未完全成功的）connection 对象并没有被显式关闭。这意味着，下一次调用 connect 方法时，由于 _is_connected 为 False，程序会尝试建立一条全新的连接，而之前那条未能成功使用或中途出错的连接，就成了“游离连接” (orphaned connection)，依然占据着 RabbitMQ 服务器的连接资源。日积月累，连接数自然水涨船高，这不仅浪费客户端资源，还可能导致服务端达到连接数上限或因管理大量连接而性能下降。

消费者的实现中也存在着完全相同的问题。

追根溯源：为什么连接不会自动关闭？

那么，为什么会发生这种连接泄漏呢？为什么在应用层面出现异常后，这些网络连接不会自动“消失”？

要理解这个问题，我们需要区分应用层逻辑和网络连接本身的生命周期。简单来说，应用层抛出异常，并不意味着底层的网络连接（如 TCP 连接）就失效了或应该自动关闭。这背后有几个关键原因：

关注点分离：网络连接（通常是 TCP/IP 层面）的建立和维护，与应用程序如何使用这些连接，在设计上是相对独立的。网络层无法、也不应该去猜测应用层是否“真的”不再需要这个连接了。一个应用级的错误（比如消息格式不正确导致处理失败）可能并不代表网络通道本身有问题，它或许还能被用于其他目的。
资源控制权：显式地打开和关闭资源（如连接、文件句柄等）是编程中的黄金法则。将资源释放的最终控制权交给开发者，可以更精确地管理其生命周期，避免因过早关闭仍在使用的连接而引发错误，或是忘记关闭不再需要的连接导致像本文这样的泄漏悲剧。
连接复用与长连接：在在许多高性能或需要持续通信的场景下，连接通常是被设计为可复用的长连接，而不是“用完即弃”。AMQP 协议本身也鼓励客户端维持长连接，并通过通道（channels）在单一连接上实现多路复用，以提高效率和减少网络开销。如果每次应用层稍有差池就自动断开连接，那么连接复用的效率将大打折扣。

因此，应用在尝试建立连接或在已连接状态下，即便处理业务逻辑时遇到非致命错误，只要底层的 AMQP 连接和 TCP 连接本身仍然健康（例如，TCP 连接未被中断，AMQP 心跳正常），它通常会保持开放状态。这就要求开发者在代码中显式地管理连接的生命周期，确保在不再需要或连接确实无法恢复时进行关闭。

修复

因此正确的做法是，确保在任何不再需要连接，或者连接发生无法恢复的错误时，都能显式地关闭它。对于连接的创建和管理，主要有两种推荐的策略：

1. 使用上下文管理器 (async with) 自动关闭短时连接

这种方式非常适合那些生命周期较短、单次或低频使用的连接。例如，生产者发布消息的场景，如果消息发布的频率不高，那么每次发布时建立连接，发布完成后立即通过 async with 语句的特性自动关闭连接，是一种简洁且安全的方式。这样可以最大限度地减少不必要的常驻连接，节约资源。

async def publish(self, message_list: list[bytes]) -> None:
    try:
        async with await aio_pika.connect_robust(...) as connection:
            async with connection.channel() as channel:
                queue = await channel.declare_queue(self.config.queue_name)
                for message in message_list:
                    await channel.default_exchange.publish(
                        aio_pika.Message(body=message),
                        routing_key=queue.name
                    )
    except Exception as e:
        logger.error(f"Publish error: {str(e)}")
        # 此处无需显式关闭连接，因为 async with 会自动关闭连接

2. 使用连接池 (aio_pika.pool 或自定义) 进行连接复用

对于需要频繁创建和释放连接的高并发应用（例如，大量的生产者或消费者实例），连接池是更优的选择。连接池可以预先创建一定数量的连接，并在需要时从中获取，用完后归还。这显著减少了连接建立和关闭的开销，提高了性能，并且可以有效限制应用的总连接数，防止资源耗尽。

aio_pika 自身提供了连接池和通道池的简单实现 (aio_pika.pool.Pool)。

connection_pool = Pool(get_connection, max_size=2)
channel_pool = Pool(get_channel, max_size=2)

async with connection_pool, channel_pool:
    async for message in self.queue.iterator():
        async with message.process() as processed_message:
            yield processed_message.body

使用连接池时，需要注意池的 max_size 参数设置，以及 acquire 和 release（通常通过 async with 自动完成）的正确使用。

小结与反思

无论是采用 async with 的简洁之道，还是连接池的集约化管理，核心都在于对资源的生命周期有清晰的认知和负责任的控制。资源管理无小事，尤其是在异步编程和与外部服务频繁交互的场景下。今天看似微不足道的一个疏忽，比如忘记在某个异常分支中优雅地关闭连接，都可能在系统长时间运行后，悄无声息地累积成压垮骆驼的最后一根稻草，不仅消耗客户端资源，更可能给 RabbitMQ 服务器带来沉重负担，例如内存耗尽、CPU 占用过高，甚至影响管理插件的性能，最终引发雪崩式的资源耗尽问题。

这次“踩坑”也提醒我们，除了关注业务逻辑的实现，更要理解所用库和协议（如 AMQP）的底层机制和最佳实践。例如，优先使用长连接、合理利用通道、配置心跳检测等，这些都是保证消息系统稳定高效运行的重要方面。

RabbitMQ 连接泄漏“踩坑记”：异常处理中忘记 close 的代价

项目中的使用方式

追根溯源：为什么连接不会自动关闭？

修复

1. 使用上下文管理器 (async with) 自动关闭短时连接

2. 使用连接池 (aio_pika.pool 或自定义) 进行连接复用

小结与反思

参考

Comments

理想拖

Previous Article

Next Article