本文目录导读：
我来介绍几种排查Python进程内存泄露的方法和脚本工具。
基础内存监控脚本
import os
import psutil
import sys
import tracemalloc
from datetime import datetime
class MemoryMonitor:
def __init__(self, pid=None):
self.pid = pid or os.getpid()
self.process = psutil.Process(self.pid)
self.memory_snapshots = []
def get_memory_usage(self):
"""获取当前内存使用情况"""
try:
memory_info = self.process.memory_info()
return {
'rss': memory_info.rss / 1024 / 1024, # MB
'vms': memory_info.vms / 1024 / 1024, # MB
'time': datetime.now().isoformat()
}
except (psutil.NoSuchProcess, psutil.AccessDenied):
return None
def monitor_memory_growth(self, interval=1, duration=60):
"""监控内存增长趋势"""
import time
start_time = time.time()
print(f"Starting memory monitor for PID: {self.pid}")
print(f"{'Time':20} {'RSS (MB)':15} {'VMS (MB)':15}")
while time.time() - start_time < duration:
mem_info = self.get_memory_usage()
if mem_info:
print(f"{mem_info['time']:20} {mem_info['rss']:15.2f} {mem_info['vms']:15.2f}")
self.memory_snapshots.append(mem_info)
time.sleep(interval)
self.analyze_trend()
def analyze_trend(self):
"""分析内存增长趋势"""
if len(self.memory_snapshots) < 2:
print("Insufficient data points")
return
first = self.memory_snapshots[0]
last = self.memory_snapshots[-1]
growth = last['rss'] - first['rss']
growth_rate = growth / len(self.memory_snapshots)
print(f"\nAnalysis:")
print(f"Initial RSS: {first['rss']:.2f} MB")
print(f"Final RSS: {last['rss']:.2f} MB")
print(f"Total growth: {growth:.2f} MB")
print(f"Growth rate: {growth_rate:.2f} MB/sample")
if growth > 0:
print("⚠️ Potential memory leak detected!")
else:
print("✅ Memory usage is stable")
使用tracemalloc定位内存泄漏
import tracemalloc
import gc
from collections import Counter
import linecache
class MemoryLeakDetector:
def __init__(self):
tracemalloc.start()
self.snapshot1 = None
self.snapshot2 = None
def take_snapshot(self, label=""):
"""获取内存快照"""
if self.snapshot1 is None:
self.snapshot1 = tracemalloc.take_snapshot()
print(f"First snapshot taken: {label}")
else:
self.snapshot2 = tracemalloc.take_snapshot()
print(f"Second snapshot taken: {label}")
self.compare_snapshots()
self.snapshot1 = self.snapshot2
self.snapshot2 = None
def compare_snapshots(self, top_n=20):
"""比较两个快照，找出内存增长最多的代码"""
if not self.snapshot1 or not self.snapshot2:
return
# 计算差异
stats = self.snapshot2.compare_to(self.snapshot1, 'lineno')
print(f"\nTop {top_n} memory allocations:")
print("-" * 80)
print(f"{'Size (bytes)':15} {'Count':10} {'Location'}")
print("-" * 80)
for stat in stats[:top_n]:
size_diff = stat.size_diff
count_diff = stat.count_diff
if size_diff > 0: # 只显示增长的部分
frame = stat.traceback[0]
filename = frame.filename.replace(os.getcwd() + '/', '')
print(f"{size_diff:>10,} (+{size_diff:>8,}) {count_diff:>5} (+{count_diff:>3}) {filename}:{frame.lineno}")
# 显示代码行
if frame.lineno:
code_line = linecache.getline(frame.filename, frame.lineno).strip()
if code_line:
print(f"{'':>25}→ {code_line}")
def get_top_allocations(self, top_n=10):
"""获取当前内存分配最多的对象"""
snapshot = tracemalloc.take_snapshot()
stats = snapshot.statistics('traceback')
print(f"\nTop {top_n} memory allocations (current):")
print("-" * 80)
for stat in stats[:top_n]:
print(f"Size: {stat.size:,} bytes, Count: {stat.count}")
for frame in stat.traceback[:3]: # 只显示前3层调用栈
line = linecache.getline(frame.filename, frame.lineno).strip()
print(f" {frame.filename}:{frame.lineno}: {line}")
print()
对象引用追踪
import gc
import sys
from collections import defaultdict
class ObjectTracker:
def __init__(self):
self.object_history = []
def track_custom_objects(self, classes_to_track):
"""追踪特定类的对象数量"""
objects = defaultdict(list)
for obj in gc.get_objects():
if obj.__class__.__name__ in classes_to_track:
class_name = obj.__class__.__name__
objects[class_name].append(id(obj))
for class_name, obj_ids in objects.items():
print(f"{class_name}: {len(obj_ids)} instances")
return objects
def find_reference_chain(self, target_obj):
"""查找对象引用链"""
referrers = gc.get_referrers(target_obj)
print(f"Object {type(target_obj).__name__} has {len(referrers)} reference(s):")
for i, referrer in enumerate(referrers[:10]): # 只显示前10个
print(f" {i+1}. {type(referrer).__name__} at {hex(id(referrer))}")
return referrers
def check_circular_references(self):
"""检查循环引用"""
gc.collect()
unreachable = gc.garbage
if unreachable:
print(f"Found {len(unreachable)} unreachable objects (potential circular refs):")
for obj in unreachable[:10]:
print(f" {type(obj).__name__}: {str(obj)[:50]}")
else:
print("No circular references detected")
完整的内存泄漏排查脚本
#!/usr/bin/env python3
"""
完整的内存泄漏排查工具
"""
import argparse
import time
import sys
import gc
import objgraph # 需要安装: pip install objgraph
class MemoryLeakScanner:
def __init__(self, target_pid=None):
self.pid = target_pid or os.getpid()
self.monitor = MemoryMonitor(self.pid)
self.detector = MemoryLeakDetector()
self.tracker = ObjectTracker()
def full_scan(self, duration=300, interval=5):
"""执行完整的内存泄漏扫描"""
print("=" * 60)
print("Memory Leak Scanner")
print("=" * 60)
print(f"\n1. Starting basic memory monitoring for {duration}s...")
self.monitor.monitor_memory_growth(interval, duration)
print(f"\n2. Taking tracemalloc snapshots...")
self.detector.take_snapshot("before operation")
# 执行目标操作（这里需要替换为你的实际业务逻辑）
# self.simulate_operation()
self.detector.take_snapshot("after operation")
print(f"\n3. Checking object references...")
# 追踪特定对象
classes_to_track = ['dict', 'list', 'MyCustomClass']
self.tracker.track_custom_objects(classes_to_track)
print(f"\n4. Current top allocations:")
self.detector.get_top_allocations()
print(f"\n5. Checking for memory leaks...")
self.check_memory_leaks()
def check_memory_leaks(self):
"""检查内存泄漏迹象"""
# 强制垃圾回收
unreachable_count = gc.collect()
print(f"Garbage collected: {unreachable_count} objects")
# 获取所有对象的引用计数
total_objects = len(gc.get_objects())
print(f"Total live objects: {total_objects}")
# 使用objgraph找出常见泄漏模式
try:
# 增长最快的对象类型
growth = objgraph.growth(limit=10)
if growth:
print("\nObject growth patterns:")
for type_name, count, delta in growth:
if delta > 0:
print(f" {type_name}: +{delta} (total: {count})")
except ImportError:
print("Tip: Install objgraph for detailed object analysis: pip install objgraph")
def simulate_operation(self):
"""模拟产生内存泄漏的操作"""
# 这只是示例，实际中替换为你的业务代码
import random
leaky_list = []
for i in range(100):
# 模拟内存泄漏：不断添加对象到全局列表
leaky_list.append({
f"key_{i}": [1] * 1000,
"timestamp": time.time()
})
print(f"Simulated operation added {len(leaky_list)} objects")
return leaky_list # 故意保留引用造成泄漏
def main():
parser = argparse.ArgumentParser(description='Python Memory Leak Detector')
parser.add_argument('--pid', type=int, help='Process ID to monitor')
parser.add_argument('--duration', type=int, default=60, help='Monitoring duration in seconds')
parser.add_argument('--interval', type=int, default=1, help='Sampling interval in seconds')
parser.add_argument('--mode', choices=['monitor', 'track', 'full'], default='full',
help='Scanning mode')
args = parser.parse_args()
scanner = MemoryLeakScanner(args.pid)
if args.mode == 'monitor':
scanner.monitor.monitor_memory_growth(args.interval, args.duration)
elif args.mode == 'track':
scanner.detector.get_top_allocations()
scanner.tracker.check_circular_references()
else:
scanner.full_scan(args.duration, args.interval)
if __name__ == "__main__":
main()
使用方法和最佳实践
安装依赖
pip install psutil tracemalloc objgraph
运行脚本
# 监控当前进程 python memory_leak_scanner.py --duration 300 # 监控特定PID的进程 python memory_leak_scanner.py --pid 12345
排查步骤
-
初步确认内存泄漏
# 定期记录内存使用 while True: memory = process.memory_info().rss / 1024 / 1024 print(f"Memory: {memory:.2f} MB") time.sleep(5) -
隔离嫌疑代码
# 使用tracemalloc精确定位 with tracemalloc.start(): # 测试代码块 pass
-
检查常见泄漏模式
- 全局缓存/列表持续增长
- 未关闭的文件/数据库连接
- 循环引用（虽然GC能处理大部分）
- 第三方库的内部状态
-
使用对象图分析
import objgraph objgraph.show_growth() # 显示增长最快的对象类型
这个工具套件可以帮助你系统地排查Python进程的内存泄漏问题，建议先从整体监控开始,然后逐步深入到具体代码位置。
