Python脚本如何排查进程内存泄露问题

wen python案例 2026-06-27 243

本文目录导读：

基础内存监控脚本
使用tracemalloc定位内存泄漏
对象引用追踪
完整的内存泄漏排查脚本
使用方法和最佳实践

我来介绍几种排查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进程的内存泄漏问题，建议先从整体监控开始,然后逐步深入到具体代码位置。

本文地址： https://www.bmjup.com/post/9731.html