用 Python tracemalloc 排查内存增长：比较批次快照并定位保留数据

批处理每跑一轮，Python 进程占用的内存就多一点。只看到系统监控曲线上升，仍不知道是哪一行创建的数据一直留着。本文用标准库 tracemalloc 对比批次前后的快照，找出仍被引用的数据的分配位置，再用不保留数据的对照组验证变化。

面向初学者到中级读者，示例不需要第三方包，也不会耗尽服务器内存。如果任务是停止前进而不是内存持续增长，可以先看 faulthandler 线程堆栈排查。

1. 方法：比较相邻批次，不只看总量

单次总量难以区分初始化和持续积累。先记录基线，再完成一批工作，比较两次快照；连续几轮都在相同分配行增长，就去检查谁保存了那些对象。

Python 官方 tracemalloc 文档说明了快照比较、按行统计以及当前值与峰值接口。下面报告把源码范围内的净增长与进程中全部被跟踪分配的当前值、峰值分开显示。

本例每轮创建 4 个 64 KiB 的 bytearray：retain 把它们加入跨批次列表；discard 只计算长度，函数返回后不保留这些数据。两种模式处理量相同，每轮都是 262144 字节，最多允许 20 轮，保留的有效载荷最多 5 MiB，另有解释器和诊断开销。

2. 准备环境

使用 CPython 3.10 或更新版本，新建练习目录：

python3 --version
mkdir tracemalloc-demo
cd tracemalloc-demo

把下一节完整代码保存为该目录中的 memory_demo.py。后面的命令都在这个目录运行；重跑会覆盖本例的两个 JSONL 报告。

示例已在 Linux、CPython 3.11.4 上测试。代码是独立诊断练习进程，结束时会关闭跟踪；不要把这段入口整体复制进已有诊断配置的服务。

3. 完整实现

"""Compare bounded retain/discard batches with CPython tracemalloc."""
import argparse
import gc
import json
from pathlib import Path
import tracemalloc

CHUNK_BYTES = 64 * 1024
BLOCKS_PER_ROUND = 4


def run_batch(cache, retain):
    processed = 0
    for _ in range(BLOCKS_PER_ROUND):
        chunk = bytearray(CHUNK_BYTES)
        processed += len(chunk)
        if retain:
            cache.append(chunk)
    return processed


def main():
    parser = argparse.ArgumentParser(description=__doc__, allow_abbrev=False)
    parser.add_argument("--mode", choices=("retain", "discard"), default="retain")
    parser.add_argument("--rounds", type=int, default=3)
    args = parser.parse_args()
    if not 1 <= args.rounds <= 20:
        parser.error("--rounds must be between 1 and 20")

    if not tracemalloc.is_tracing():
        tracemalloc.start(1)
    source = str(Path(__file__).resolve())
    filters = (tracemalloc.Filter(True, source),)
    cache = []
    gc.collect()
    before = tracemalloc.take_snapshot().filter_traces(filters)
    try:
        for number in range(1, args.rounds + 1):
            tracemalloc.reset_peak()
            processed = run_batch(cache, args.mode == "retain")
            gc.collect()
            current, peak = tracemalloc.get_traced_memory()
            after = tracemalloc.take_snapshot().filter_traces(filters)
            changes = after.compare_to(before, "lineno")
            positive = [item for item in changes if item.size_diff > 0]
            positive.sort(key=lambda item: item.size_diff, reverse=True)
            report = {
                "round": number,
                "mode": args.mode,
                "processed_bytes": processed,
                "retained_blocks": len(cache),
                "source_growth_kib": round(sum(s.size_diff for s in changes) / 1024, 1),
                "traced_current_kib": round(current / 1024, 1),
                "traced_peak_kib": round(peak / 1024, 1),
                "top_growth": [
                    {
                        "file": s.traceback[0].filename,
                        "line": s.traceback[0].lineno,
                        "size_diff_kib": round(s.size_diff / 1024, 1),
                        "count_diff": s.count_diff,
                    }
                    for s in positive[:3]
                ],
            }
            print(json.dumps(report, ensure_ascii=True), flush=True)
            before = after
    finally:
        cache.clear()
        tracemalloc.stop()


if __name__ == "__main__":
    main()

cache 刻意在多轮之间存活。每轮测量前调用 gc.collect() 是为了减少这次对照实验中的回收时机差异，仍然被列表引用的数据不会因此消失；这不是生产脚本的修复方法。

过滤器仅纳入本文件直接分配的数据，便于在练习中找出 bytearray 所在行。接入实际项目时，需要把过滤范围改为相关业务文件；第三方库内部的分配可能被这个过滤器排除。

4. 执行两组实验并读懂输出

查看参数，并分别运行两个独立进程：

python3 memory_demo.py --help
python3 memory_demo.py --mode retain --rounds 3 > retain.jsonl
python3 memory_demo.py --mode discard --rounds 3 > discard.jsonl
cat retain.jsonl
cat discard.jsonl

每个报告有三行 JSON。关注以下字段：

字段	含义与预期
`retained_blocks`	`retain` 依次为 4、8、12；`discard` 始终为 0。
`source_growth_kib`	本文件相对上一轮快照的净增长；`retain` 每轮约 256 KiB，`discard` 接近 0。
`top_growth`	正增长最大的三处源码位置；`retain` 首项应指向 `chunk = bytearray(CHUNK_BYTES)`。
`traced_current_kib`	测量时全部被跟踪分配的当前量，包含部分诊断开销。
`traced_peak_kib`	本轮重设峰值后达到的最高跟踪量，含轮次开始时仍存在的数据。

实际行号、对象开销和小幅波动依解释器而变，不要把某个精确字节数写成业务阈值。source_growth_kib 是有符号净值；top_growth 只展示正增长，二者不一定相加相等。报告中的峰值在获取第二次快照之前读取，不覆盖后续格式化报告的整个过程。

自动验证实验结果，使用宽松范围，避免依赖具体对象头大小：

python3 - <<'PY_CHECK'
import json
from pathlib import Path

def read(name):
    return [json.loads(line) for line in Path(name).read_text().splitlines()]

kept = read("retain.jsonl")
discarded = read("discard.jsonl")
assert len(kept) == len(discarded) == 3
assert [r["retained_blocks"] for r in kept] == [4, 8, 12]
assert all(r["processed_bytes"] == 262144 for r in kept + discarded)
assert all(r["retained_blocks"] == 0 for r in discarded)
assert all(r["source_growth_kib"] > 240 for r in kept)
assert all(abs(r["source_growth_kib"]) < 32 for r in discarded)
assert all(r["top_growth"][0]["size_diff_kib"] > 240 for r in kept)
print("PASS: retained growth found; discarded payload does not accumulate")
PY_CHECK

预期输出 PASS: retained growth found; discarded payload does not accumulate。这个断言验证了小型受控实验，不是通用的内存泄漏判定器。正式脚本应按相同输入规模、相同工作阶段比较多轮，并先完成模块加载和必要预热。

5. 接入真实任务并修正保留方式

在真实任务中，把 run_batch() 换成一批实际处理逻辑，保留批次前后测量。初步确认分配行后，检查跨批次列表、无限增长的字典缓存、尚未消费的任务结果；只有在业务不再需要时，才释放对应引用。

若问题来自一次性读取全部 CSV，可参考 Python 流式处理大型 CSV，边读边汇总。若来自一次提交大量 Future，可参考限制并行批处理的待执行任务，让提交量和结果回收都有上限。调整后用相同输入重跑，并核对业务结果一致。

需要观察启动期间的分配时，可提前启用跟踪：

python3 -X tracemalloc=5 memory_demo.py --mode retain --rounds 3

CPython 命令行文档记录了 -X tracemalloc=NFRAME。本例发现已在跟踪时不会重新启动；保存更多调用帧便于后续分析，但当前输出仍按分配行汇总。启动开关本身不会自动打印报告。

6. 排查常见误读

现象	具体处理
快照看不到预期数据	确认跟踪在分配前启动，并检查文件过滤范围。
峰值高，批次结束净增长接近 0	先看批次内临时大对象，考虑分块或减少同时存活的数据。
连续多轮在同一行增长	检查对应对象是否被缓存或结果列表持续保存；增长也可能符合业务需求。
系统监控上涨，跟踪值变化小	扩大文件过滤范围，并调查原生扩展或外部资源；两个指标不能直接等同。
诊断使任务明显变慢	缩小输入和轮数，减少保存的调用帧，避免每条记录都创建快照。

跟踪值不是整个进程的 RSS。Python 分配器可能保留已释放的内存供复用，原生扩展也可能在未被跟踪的分配器中申请内存；参见 PEP 454 的设计说明。因此不能用“RSS 没有马上下降”单独证明引用仍然存在，也不能用此实验排除所有原生层内存问题。

参数错误会明确失败，例如：

python3 memory_demo.py --rounds 0
printf 'exit=%s\n' "$?"

预期最后一行为 exit=2。示例不会尝试自动调大系统限制或终止其他进程。

7. 小结

先用相同工作量比较相邻批次，再沿着持续增长的分配行检查对象保留位置。通过流式处理、限制缓存或及时消费结果减少不必要引用后，重复测量并核对业务输出。把快照净增长、批次峰值和进程 RSS 分开看，才能判断下一步该优化哪里。