4. Python内存剖析：用 tracemalloc 和 memory_profiler 定位内存泄漏

内存泄漏这事儿，在嵌入式边缘设备上特别要命。你想想看，设备跑着跑着，内存越吃越多，最后直接挂掉——这在产线上可是重大事故。我当年调试一个摄像头识别模块，就吃过这个亏。今天咱们就聊聊怎么用两个Python神器，把内存泄漏揪出来。

4.1 为什么边缘设备上内存泄漏更致命？

在服务器上，内存泄漏了可以重启。但在边缘设备上，很多场景不允许重启。比如工业相机、车载终端，一跑就是几个月甚至几年。内存一点点被蚕食，最终系统崩溃。

我见过最典型的场景：一个图像处理流水线，每帧都创建新的numpy数组，但忘记释放旧的。刚开始跑得好好的，8小时后内存占用从50MB飙升到800MB。嗯，这就是典型的“慢慢死”问题。

⚠️ 注意： 边缘设备的内存通常只有几百MB到几GB。泄漏几十MB，在服务器上不算事，在边缘设备上可能就是灾难。

4.2 tracemalloc：Python自带的追踪利器

tracemalloc是Python 3.4以后内置的模块，不需要额外安装。它能在代码层面追踪每个内存分配的位置。我个人习惯在调试阶段就把它打开。

4.2.1 基础用法：快照对比

核心思路很简单：拍两张内存快照，对比差异，找出增长最多的代码路径。

import tracemalloc

# 启动追踪
tracemalloc.start()

# 模拟一段可能有泄漏的代码
def process_image():
    # 假设这里有泄漏
    data = [bytearray(1024 * 1024) for _ in range(10)]
    # 忘记释放了...
    return "done"

# 拍第一张快照
snapshot1 = tracemalloc.take_snapshot()

process_image()

# 拍第二张快照
snapshot2 = tracemalloc.take_snapshot()

# 对比差异
stats = snapshot2.compare_to(snapshot1, 'lineno')
print("内存增长最多的前5个位置：")
for stat in stats[:5]:
    print(stat)

输出结果会告诉你：哪个文件的哪一行，多分配了多少内存。我曾经用这个方法，在一个图像处理库中找到了一个循环里反复创建临时列表的问题——每次循环多分配2MB，1000次循环就是2GB。

4.2.2 按文件分组统计

有时候泄漏点分散在多个函数里。这时候按文件分组更清晰。

# 按文件名分组统计
stats = snapshot2.statistics('filename')

print("各文件内存占用排名：")
for stat in stats[:10]:
    print(f"{stat.count} blocks, {stat.size / 1024:.1f} KiB - {stat.traceback[0].filename}")
💡 小技巧： 我建议在关键业务函数前后都加上快照。比如“接收图片前”和“处理完图片后”。如果处理完后内存没回到基线，那基本就是泄漏了。

4.3 memory_profiler：逐行分析内存

tracemalloc能告诉你“哪里分配了内存”，但memory_profiler能告诉你“每行代码消耗了多少内存”。这两个工具配合使用，效果最好。

4.3.1 安装与装饰器用法

# 安装
# pip install memory_profiler

from memory_profiler import profile

@profile
def my_function():
    a = [i for i in range(100000)]
    b = [i * 2 for i in range(100000)]
    c = a + b
    del a
    return c

my_function()

运行后会输出每行代码的内存增量。你会看到哪一行分配了最多内存，哪一行释放了内存。我记得有一次排查一个Web服务的内存泄漏，就是用这个工具发现了一个字典在循环里不断膨胀——每次迭代都往里面加新键，但从不删除旧键。

4.3.2 逐行监控内存变化

memory_profiler还有一个很实用的功能：监控内存随时间的变化。

from memory_profiler import memory_usage
import time

def leaky_function():
    container = []
    for i in range(100):
        container.append(bytearray(1024 * 100))  # 每次100KB
        time.sleep(0.1)
    return container

# 监控内存变化，每0.5秒采样一次
mem_usage = memory_usage((leaky_function,), interval=0.5, timeout=None)
print(f"内存峰值: {max(mem_usage):.1f} MiB")
print(f"内存基线: {mem_usage[0]:.1f} MiB")
print(f"内存增长: {max(mem_usage) - mem_usage[0]:.1f} MiB")

这个输出能直观地看到内存曲线。如果曲线一直向上不回落，那就是泄漏。我曾经在一个MQTT客户端里看到内存曲线像楼梯一样逐级上升——每收到一条消息就涨一点，永远不会降下来。

4.4 实战：定位一个典型的内存泄漏

咱们来模拟一个真实场景。假设你有一个边缘设备上的视频帧处理函数：

import tracemalloc
import gc

# 模拟泄漏场景
class FrameProcessor:
    def __init__(self):
        self.cache = {}  # 这个字典会不断增长
    
    def process(self, frame_id, data):
        # 处理完应该清理，但这里忘了
        self.cache[frame_id] = data.copy()
        # 做一些处理...
        result = data[:100]
        return result

# 启动追踪
tracemalloc.start()
gc.collect()  # 先清理一次
snapshot_before = tracemalloc.take_snapshot()

# 模拟运行
processor = FrameProcessor()
for i in range(1000):
    data = bytearray(1024 * 10)  # 10KB每帧
    processor.process(i, data)

snapshot_after = tracemalloc.take_snapshot()

# 分析差异
stats = snapshot_after.compare_to(snapshot_before, 'lineno')
print("泄漏分析结果：")
for stat in stats[:5]:
    print(f"大小: {stat.size_diff / 1024:.1f} KiB, 位置: {stat.traceback}")

运行结果会明确指向 self.cache[frame_id] = data.copy() 这一行。嗯，问题找到了——cache字典只增不减。

🔑 关键思路： 定位内存泄漏的核心就是“对比”。对比前后的快照，对比不同时间点的内存曲线。没有对比，你就是在黑暗中摸索。

4.5 避坑指南与最佳实践

这些年我踩过不少坑，总结几条实用的：

  • 别信直觉，信数据：我曾经凭感觉认为某个函数有泄漏，结果用tracemalloc一查，发现是另一个完全没想到的地方。所以，先跑工具，再下结论。
  • 注意循环引用：Python的垃圾回收虽然会自动处理循环引用，但如果有__del__方法，就可能造成泄漏。用gc.get_objects()可以检查。
  • 第三方库也要查：有一次我排查了三天，最后发现是某个图像处理库的C扩展有内存泄漏。tracemalloc能追踪到C扩展的调用点。
  • 生产环境慎用：tracemalloc会带来约10%的性能开销。我建议只在测试环境或调试阶段开启。
⚠️ 重要提醒： 在嵌入式设备上，内存泄漏的排查成本很高。因为设备资源有限，你可能没法直接跑这些工具。我的做法是：在开发机上模拟数据流，用tracemalloc和memory_profiler反复测试，确认无误后再部署到设备上。

4.6 总结

说白了，内存泄漏排查就三步：

  1. 用tracemalloc拍快照，对比差异，定位泄漏的代码位置
  2. 用memory_profiler逐行分析，确认具体哪一行消耗了不该消耗的内存
  3. 修复后，再用同样的方法验证内存是否回到基线

这两个工具配合使用，能覆盖90%以上的内存泄漏场景。剩下的10%，可能需要结合gc模块和objgraph来排查更复杂的循环引用问题。不过那是另一个话题了，咱们下节课再聊。

← 上一章 📖 返回目录 下一章 →