Python 资源泄露排查asyncio 场景下连接未关闭导致的诡异 Bug 定位一、深度引言与场景痛点大家好我是赵咕咕。凌晨三点手机收到 PagerDuty 报警我们线上 RAG 服务的内存占用从正常的 800MB 一路飙到 4GB然后 OOM Killer 把进程杀了。重启后正常然后4小时后又热门。重启解决了症状但病根还在。我开始排查。日志里没有任何异常请求量稳定业务逻辑也没问题。CPU 曲线正常就是内存出现了可疑的阶梯式上涨——每处理一批请求内存往上跳一截GC 回收不了。这就是典型的资源泄露。而在 asyncio 世界里资源泄露的排查比同步代码要难好几个数量级。二、底层机制与原理深度剖析2.1 asyncio 资源泄露为什么难排查同步代码里的资源泄露通常比较直观你打开一个文件没关你的with open(...) as f:写成了f open(...)。Python 对象最终会被 GC 回收连接也会随之关闭。但 asyncio 不一样协程的引用可能挂在事件循环上。一个未 await 的协程可能被 event loop 持有引用导致相关对象永远不被回收。连接池的引用计数不直观。像aiohttp.ClientSession、asyncpg.Pool这种连接池内部有复杂的生命周期管理。信号量可能泄漏。asyncio.Semaphore如果 acquire 了没 release轻则死锁重则资源堆积。Task 泄漏。创建了 Task 但没有 await 或 cancel它就一直跑着持有的资源也不会释放。2.2 这次 Bug 的真相排查过程用了一个下午。最终定位到的罪魁祸首是这样的flowchart TD A[收到请求] -- B[创建临时 ClientSession] B -- C[发起 HTTP 请求到向量服务] C -- D{请求成功?} D --|是| E[解析响应] D --|否| F[抛出异常] E -- G[session 未显式关闭] F -- H[异常处理中 session 也未关闭] G -- I[底层 TCP 连接悬挂] H -- I I -- J[连接数持续增长] J -- K[内存泄漏累积] K -- L[OOM Kill] style F fill:#f96,stroke:#333 style G fill:#f96,stroke:#333 style H fill:#f96,stroke:#333问题出在一段被简化过的代码里# ❌ 问题代码 async def search_documents(query: str): session aiohttp.ClientSession() async with session.post(url, json{query: query}) as resp: result await resp.json() return result # session 没有关闭 async def main(): tasks [search_documents(q) for q in queries] # 100个请求 results await asyncio.gather(*tasks)ClientSession创建后没有调用await session.close()。在async with session.post(...)里只有resp这个响应对象被自动关闭了但底层连接被放回了连接池——而连接池本身没有被正确关闭。每次请求都创建新的ClientSession每次都没关。100个请求就漏了100个 session。在高并发场景下几分钟就积累了几千个悬挂连接。三、生产级代码实现3.1 正确用法复用 ClientSessionimport asyncio import aiohttp import logging from contextlib import asynccontextmanager from typing import AsyncIterator logger logging.getLogger(__name__) class SearchClient: 封装向量搜索的客户端确保连接正确管理。 def __init__(self, base_url: str, timeout: int 30): self._base_url base_url self._timeout aiohttp.ClientTimeout(totaltimeout) self._session: aiohttp.ClientSession | None None async def start(self) - None: 初始化 session在应用启动时调用一次。 self._session aiohttp.ClientSession( timeoutself._timeout, connectoraiohttp.TCPConnector(limit50, limit_per_host20), ) async def stop(self) - None: 关闭 session在应用关闭时调用。 if self._session is not None: await self._session.close() self._session None async def search(self, query: str) - dict: if self._session is None: raise RuntimeError(Client 未启动请先调用 start()) url f{self._base_url}/api/search try: async with self._session.post( url, json{query: query} ) as resp: resp.raise_for_status() return await resp.json() except aiohttp.ClientError as e: logger.error(向量搜索请求失败: %s, e) raise except asyncio.TimeoutError: logger.error(向量搜索超时: %s, query[:50]) raise asynccontextmanager async def managed_search_client(base_url: str) - AsyncIterator[SearchClient]: 通过上下文管理器确保资源正确释放。 client SearchClient(base_url) await client.start() try: yield client finally: await client.stop()3.2 带资源泄漏检测的用法async def safe_batch_search(queries: list[str], client: SearchClient) - list[dict]: 带超时和异常保护的批量搜索。 semaphore asyncio.Semaphore(10) # 控制并发数 async def _search_one(q: str) - dict: async with semaphore: return await client.search(q) tasks [] for q in queries: task asyncio.create_task(_search_one(q)) tasks.append(task) try: results await asyncio.gather(*tasks, return_exceptionsTrue) except Exception: for t in tasks: if not t.done(): t.cancel() raise # 如果 gather 本身出问题罕见取消所有 task # 处理单个任务异常 output [] for i, r in enumerate(results): if isinstance(r, Exception): logger.error(查询 %s 失败: %s, queries[i][:50], r) output.append({error: str(r), query: queries[i]}) else: output.append(r) return output关键点asyncio.gather(..., return_exceptionsTrue)处理单个任务失败Semaphore控制并发避免连接池打满异常时cancel()清理未完成的 task。四、边界分析与架构权衡4.1 排查资源泄露的工具箱排查 asyncio 资源泄露我常用的三板斧用gc模块看对象引用import gc import aiohttp sessions [obj for obj in gc.get_objects() if isinstance(obj, aiohttp.ClientSession)] print(f悬挂的 ClientSession 数量: {len(sessions)})用tracemalloc追踪内存分配import tracemalloc tracemalloc.start() # ... 运行一段时间 ... snapshot tracemalloc.take_snapshot() for stat in snapshot.statistics(lineno)[:10]: print(stat)用asyncio.all_tasks()检查未完成的协程pending [t for t in asyncio.all_tasks() if not t.done()] print(fPending tasks: {len(pending)})4.2 设计原则全局共享 ClientSession。应用级别生命周期内只创建一次、只关闭一次。显式关闭优于依赖 GC。await session.close()写清楚别指望 Python 帮你打扫。连接池限制。TCPConnector(limit...)设置一个合理的上限防止连接数打满系统 fd。监控先行。上线前就对连接数、内存做监控不要等 OOM 了才发现。4.3 为什么不直接用with自动管理你可能会问为什么不给ClientSession加__aenter__/__aexit__然后用async with答案是可以这么做但不适合高频调用场景。每次请求都创建/销毁 session 至少要经历 TCP 握手和 TLS 协商延迟增加 50-100ms。在高并发下socket 的 TIME_WAIT 状态也会耗尽。所以结论是用全局单例 连接池复用生命周期与进程一致。五、总结asyncio 的资源泄露排查起来确实比同步代码痛苦——引用挂在 event loop 上、连接池生命周期隐晦、协程挂起后 GC 无法回收。但这些坑本质上都是生命周期管理的问题。记住三个原则创建了就负责关闭——ClientSession、数据库连接池、文件句柄都一样。复用优于反复创建——一个全局 session 比每次 new 一个更省资源、更快、更安全。监控资源上限——连接数、内存、fd 数设好告警阈值。那天调试结束后我把那段每次创建 session的代码改了线上内存稳定在了 600MB 以下再也没有凌晨三点的报警。从此以后我写的每个 asyncio 客户端都有start()和stop()方法——这是一个工程师的自我修养。下一篇预告RAG 反馈回路设计让你的检索越用越聪明。