跳到主要内容

内存泄漏排查

问题​

Python 应用出现内存持续增长怎么排查？常见的内存泄漏场景有哪些？

答案​

常见泄漏场景​

用 tracemalloc 定位​

debug/trace_memory.py
import tracemalloc

tracemalloc.start()

# ... 业务代码执行 ...

snapshot = tracemalloc.take_snapshot()
top_stats = snapshot.statistics("lineno")

print("Top 10 内存分配:")
for stat in top_stats[:10]:
    print(stat)

# 对比两个快照找增量
snapshot1 = tracemalloc.take_snapshot()
# ... 执行一段时间 ...
snapshot2 = tracemalloc.take_snapshot()

diff = snapshot2.compare_to(snapshot1, "lineno")
print("\n内存增量 Top 10:")
for stat in diff[:10]:
    print(stat)

用 objgraph 分析对象图​

debug/objgraph_demo.py
import objgraph

# 查看增长最多的对象类型
objgraph.show_growth(limit=10)

# 执行操作后再看增长
do_something()
objgraph.show_growth(limit=10)

# 查找特定对象的引用链
leaking_obj = find_suspect()
objgraph.show_backrefs(leaking_obj, max_depth=5, filename="refs.png")

典型泄漏案例​

examples/leak_cases.py
# ❌ 案例 1：全局列表无限增长
_cache = []

def process(data):
    result = heavy_compute(data)
    _cache.append(result)  # 永远不清理
    return result

# ✅ 修复：使用 LRU 缓存或定期清理
from functools import lru_cache

@lru_cache(maxsize=1000)
def process(data):
    return heavy_compute(data)
# ❌ 案例 2：闭包持有大对象
def create_handler():
    large_data = load_huge_dataset()  # 100MB

    def handler(request):
        # 只用了 large_data 的一个属性，但整个对象被闭包持有
        return large_data.name

    return handler

# ✅ 修复：只捕获需要的值
def create_handler():
    large_data = load_huge_dataset()
    name = large_data.name  # 只取需要的

    def handler(request):
        return name

    return handler
# ❌ 案例 3：循环引用 + __del__
class Node:
    def __init__(self):
        self.parent = None
        self.children = []

    def __del__(self):
        print("deleted")  # 有 __del__ 时 GC 可能无法回收循环引用

# ✅ 修复：使用弱引用
import weakref

class Node:
    def __init__(self):
        self._parent = None
        self.children = []

    @property
    def parent(self):
        return self._parent() if self._parent else None

    @parent.setter
    def parent(self, node):
        self._parent = weakref.ref(node) if node else None

生产环境监控​

monitoring/memory_monitor.py
import psutil
import os

def get_memory_usage() -> dict:
    process = psutil.Process(os.getpid())
    mem = process.memory_info()
    return {
        "rss_mb": mem.rss / 1024 / 1024,      # 常驻内存
        "vms_mb": mem.vms / 1024 / 1024,      # 虚拟内存
        "percent": process.memory_percent(),
    }

# 定期监控，超阈值告警
import threading

def memory_watchdog(threshold_mb: float = 500, interval: float = 60):
    def check():
        while True:
            usage = get_memory_usage()
            if usage["rss_mb"] > threshold_mb:
                alert(f"内存超限: {usage['rss_mb']:.0f}MB > {threshold_mb}MB")
            threading.Event().wait(interval)

    threading.Thread(target=check, daemon=True).start()

常见面试问题​

Q1: 排查内存泄漏的步骤？​

答案：

  1. 确认泄漏：监控 RSS 是否持续增长（不是波动）
  2. 定位代码：tracemalloc 对比两个快照找增量
  3. 分析引用：objgraph 查看对象引用链
  4. 修复验证：修复后监控内存曲线是否平稳

Q2: gc.collect() 不能回收的情况？​

答案：

  • C 扩展中分配的内存不受 Python GC 管理
  • 有 __del__ 方法的循环引用对象（Python 3.4+ 已改善）
  • 全局变量引用的对象

Q3: weakref 什么时候用？​

答案：

  • 缓存：WeakValueDictionary 让缓存对象可被 GC
  • 观察者模式：回调引用不阻止对象被回收
  • 循环引用：父子关系中弱引用打破循环

相关链接​