连接泄漏检测与自动回收

连接泄漏这事儿，说白了就是借了不还。你在连接池里拿了个连接，用完忘了还回去，连接池就少了一个可用连接。时间一长，池子就空了，新请求来了只能排队等着，严重的直接拖垮整个系统。

我在项目中遇到过好几次这样的故障。印象最深的一次，线上服务每隔两小时就卡死一次，查了半天才发现是某个定时任务里用了连接没释放。从那以后，我就在连接池层加上了泄漏检测和自动回收机制。这一章，我们就来聊聊怎么从代码层面解决这个问题。

连接泄漏的成因与危害

连接泄漏的根因，其实就那么几种：

• 异常路径没处理——业务代码抛了异常，finally 块漏写了 close。
• 连接被长时间占用——比如事务内做了耗时网络请求，连接一直没归还。
• 第三方库的坑——某些 ORM 内部缓存了连接引用，外部无法主动回收。
• 线程池配合失误——连接绑定到线程，线程不退，连接也不回池。

危害大家应该都懂：连接池枯竭、请求堆积、CPU 打满、服务雪崩。更隐蔽的是，泄漏往往不是瞬间发生的，而是慢慢"温水煮青蛙"。等发现的时候，业务已经受了影响。

检测策略：怎么发现泄漏

检测连接泄漏，常用的策略有四种。我整理了一张对比表，方便你看场景选方案：

我个人习惯组合使用代理拦截 + 后台扫描。代理拦截负责精确计数和调用栈记录，后台扫描兜底，防止计数遗漏。你想想看，一个准实时、一个定时兜底，基本能覆盖大部分泄漏场景。

自动回收机制设计

检测到泄漏之后，不能光报警不干活。自动回收的核心逻辑就三步：

1. 标记——后台线程扫描时，把超时未归还的连接标记为"待回收"。
2. 强制关闭——调用连接的真实 close 方法，释放底层 socket 和数据库端资源。
3. 回调通知——通过事件或日志回调，把泄漏堆栈和连接信息记录下来，方便事后排查。

这里有个细节要注意：强制关闭正在执行事务的连接，会导致事务回滚。所以回收前要判断连接是否处于活动事务中。如果是，先记录告警，等事务超时再回收。

实战代码：带泄漏检测的连接池包装器

下面是一个 Python 示例，用代理模式包装连接对象。核心思路是在获取连接时记录时间戳和调用栈，在归还时清除记录。后台线程扫描记录，发现超时未清除的就强制回收。

import threading
import time
import traceback

class ConnectionProxy:
    """连接代理，记录获取时间和调用栈"""
    
    def __init__(self, real_conn, conn_id):
        self._real_conn = real_conn
        self._conn_id = conn_id
        self._acquire_time = time.time()
        self._stack = traceback.format_stack()
        self._closed = False
    
    def close(self):
        if not self._closed:
            self._real_conn.close()
            self._closed = True
    
    def is_alive(self):
        return not self._closed
    
    @property
    def age(self):
        return time.time() - self._acquire_time

class LeakDetector:
    """泄漏检测器，后台扫描并自动回收"""
    
    def __init__(self, pool, timeout=30, check_interval=5):
        self._pool = pool
        self._timeout = timeout
        self._active = {}
        self._lock = threading.Lock()
        self._running = True
        self._thread = threading.Thread(target=self._scan, daemon=True)
        self._thread.start()
    
    def register(self, conn_id, proxy):
        with self._lock:
            self._active[conn_id] = proxy
    
    def unregister(self, conn_id):
        with self._lock:
            self._active.pop(conn_id, None)
    
    def _scan(self):
        while self._running:
            time.sleep(self._check_interval)
            now = time.time()
            to_reclaim = []
            with self._lock:
                for conn_id, proxy in list(self._active.items()):
                    if now - proxy._acquire_time > self._timeout:
                        to_reclaim.append((conn_id, proxy))
            for conn_id, proxy in to_reclaim:
                print(f"[LeakDetector] 回收泄漏连接 {conn_id}，"
                      f"存活 {proxy.age:.1f}s")
                # 强制关闭
                try:
                    proxy.close()
                except Exception as e:
                    print(f"关闭失败: {e}")
                self.unregister(conn_id)
                # 回调：记录泄漏堆栈
                self._on_leak(conn_id, proxy._stack)
    
    def _on_leak(self, conn_id, stack):
        # 实际项目中这里写日志或上报监控
        print(f"[Leak] 连接 {conn_id} 泄漏，调用栈：")
        for line in stack[-6:]:
            print(line.strip())
    
    def stop(self):
        self._running = False
        self._thread.join()

连接泄漏检测与自动回收流程

下面这张图，把从连接获取、使用、正常归还，到泄漏检测、自动回收的完整链路画了出来。你可以对照着代码看，会更清楚：

从图上能清楚看到，正常路径走左侧绿色分支，异常路径走右侧红色分支。泄漏检测器作为"哨兵"，一旦发现连接超时未归还，就触发自动回收，并记录堆栈到告警系统。这个模式我在好几个项目里都用了，效果很稳。

连接泄漏检测不是什么高深技术，但做得好能帮团队省下大量排障时间。你想想看，线上出了故障，能直接定位到哪行代码没释放连接，比对着日志猜快多了。