适用场景

一个常驻的 Python API、消费者或定时任务进程，刚启动时只占用几百 MB，运行数小时后内存持续上涨。请求结束后内存没有明显回落，重启可以暂时恢复，但过一段时间又出现同样问题。监控最终触发容器 OOMKilled，或节点开始频繁交换内存。

本文针对“对象仍被 Python 引用”的增长场景，演示如何用标准库 tracemalloc 对两个时间点做快照差分，定位到无界字典缓存，再用有容量上限的 LRU 缓存和可观测指标完成治理。整套方法不依赖在线调试器，适合先在预发复现，也可以经过权限与性能评估后用于生产短时取证。

现象描述

先区分三个容易混淆的指标：

  • RSS 是操作系统看到的进程驻留内存，包含 Python 堆、扩展模块、线程栈和内存分配器保留的页面。
  • tracemalloc 只跟踪由 Python 内存分配器管理的分配，不能完整解释 NumPy、图像库或数据库驱动在原生代码中的内存。
  • 垃圾回收只能回收不可达对象。只要对象仍被全局字典、单例、回调或任务引用，手动执行 gc.collect() 也不会释放它。

常见现场表现如下：

  1. RSS 与业务请求量共同上升，低峰期仍不下降。
  2. GC 次数正常，强制回收后占用变化很小。
  3. 堆快照中某个业务文件的对象数量持续增加。
  4. 重启后恢复，说明更像进程内状态增长，而不是节点级资源问题。

不要一看到 RSS 不下降就判断为泄漏。CPython 的内存分配器可能保留已经释放的内存供后续复用，因此应重点观察“相同负载下是否持续创新高”和“快照中的存活对象是否持续增加”。

可能原因

1. 无界缓存

下面的代码按用户和查询条件缓存结果，但没有容量、过期时间或淘汰策略。只要键持续变化，字典就只增不减：

"""演示无界缓存导致的存活对象增长。"""

result_cache: dict[str, bytes] = {}


def query_report(cache_key: str) -> bytes:
    """返回报表内容，并错误地永久缓存每一个结果。"""
    cached = result_cache.get(cache_key)
    if cached is not None:
        return cached

    payload = (cache_key.encode("utf-8") + b":") * 4096
    result_cache[cache_key] = payload
    return payload

2. 后台任务或回调长期持有请求对象

把请求、响应或大列表捕获进闭包，再将闭包注册到一个不会清空的回调列表，也会让整条对象引用链存活。异步服务中，永不结束的任务同样可能保存创建时的局部变量。

3. 队列没有背压

生产速度长期高于消费速度时，内存队列会表现为“缓慢泄漏”。这种情况要同时检查队列深度和处理延迟，不能只看对象类型。

4. 原生扩展或碎片化

如果 RSS 持续增长，而 tracemalloc 的已跟踪内存基本不变，应转向原生扩展、线程栈、内存映射或分配器碎片。此时继续调大 tracemalloc 深度通常没有帮助。

排查思路

第一步：确认增长趋势，而不是单点数值

Linux 上可先记录进程 RSS 和虚拟内存：

PID="$(pgrep -n -f 'python.*app')"
ps -o pid,etimes,rss,vsz,cmd -p "$PID"
grep -E 'VmRSS|VmSize|VmSwap|Threads' "/proc/$PID/status"

rss 和 /proc/.../status 中的单位通常是 KiB。采样时还应同步记录请求量、队列深度和缓存条目数，避免把正常的流量扩容误判为泄漏。pgrep 可能匹配多个进程，生产操作前必须核对 PID 与命令行。

第二步：在应用启动早期启用 tracemalloc

tracemalloc 只能追踪启用之后的 Python 分配。最简单的方法是在启动命令中开启，并保留 25 层调用栈：

PYTHONTRACEMALLOC=25 python -m app

也可以在程序入口、导入业务模块之前调用：

import tracemalloc

tracemalloc.start(25)

调用栈越深，定位信息越完整，但额外内存和 CPU 开销也更高。生产环境应先在压测环境评估，并限制取证持续时间、访问权限和快照数量。

第三步：比较两个时间点，而不是只看一次 Top

下面的诊断器保留一份基线快照，稍后输出增量最大的代码位置。接口只返回聚合后的文件名、行号和字节数，不暴露请求体或密钥：

"""提供受控的 Python 堆快照差分能力。"""

import linecache
import threading
import tracemalloc
from dataclasses import dataclass


@dataclass(frozen=True, slots=True)
class AllocationGrowth:
    """描述一个代码位置在两个快照之间的分配增长。"""

    filename: str
    lineno: int
    size_diff: int
    count_diff: int
    source: str


class HeapGrowthProbe:
    """串行维护基线，并返回增长最大的 Python 分配位置。"""

    def __init__(self, frame_depth: int = 25) -> None:
        if frame_depth < 1 or frame_depth > 50:
            raise ValueError("frame_depth 必须在 1 到 50 之间")
        if not tracemalloc.is_tracing():
            tracemalloc.start(frame_depth)
        self._baseline: tracemalloc.Snapshot | None = None
        self._lock = threading.Lock()

    def mark_baseline(self) -> None:
        """在业务稳定后保存基线快照。"""
        with self._lock:
            self._baseline = tracemalloc.take_snapshot()

    def compare(self, limit: int = 10) -> list[AllocationGrowth]:
        """返回相对基线增长最大的分配位置。"""
        if limit < 1 or limit > 100:
            raise ValueError("limit 必须在 1 到 100 之间")

        with self._lock:
            if self._baseline is None:
                raise RuntimeError("请先调用 mark_baseline() 建立基线")
            current = tracemalloc.take_snapshot()
            differences = current.compare_to(self._baseline, "lineno")

        growth: list[AllocationGrowth] = []
        for statistic in differences:
            if statistic.size_diff <= 0:
                continue
            frame = statistic.traceback[0]
            growth.append(
                AllocationGrowth(
                    filename=frame.filename,
                    lineno=frame.lineno,
                    size_diff=statistic.size_diff,
                    count_diff=statistic.count_diff,
                    source=linecache.getline(frame.filename, frame.lineno).strip(),
                )
            )
            if len(growth) == limit:
                break
        return growth

推荐的取证时序是：

  1. 应用预热完成后调用 mark_baseline()。
  2. 保持一段可重复的稳定负载，等待内存出现明确增长。
  3. 调用 compare()，记录前 10 个正增长位置。
  4. 再运行同样负载并重复比较，确认同一位置是否持续增长。

size_diff 表示相对基线增加的字节数，count_diff 表示对象块数量变化。如果两者都持续为正，并指向写入缓存或队列的业务行，证据比单次 Top 排名更可靠。

定位示例

用不断变化的查询键压测前面的无界缓存：

probe = HeapGrowthProbe()
probe.mark_baseline()

for index in range(10_000):
    query_report(f"tenant-{index}:monthly-report")

for item in probe.compare(limit=5):
    print(
        item.filename,
        item.lineno,
        item.size_diff,
        item.count_diff,
        item.source,
    )

如果排名靠前的位置指向 result_cache[cache_key] = payload，并且缓存条目数与唯一键数量近似同步增长，就应继续回答两个问题：这些键是否真的需要跨请求保存，以及缓存允许占用多少内存。不要直接把“定期 clear()”当成最终修复，因为清空瞬间可能造成下游请求洪峰。

修复方案：建立有界缓存

对于进程内、可重新计算、允许按最近使用情况淘汰的数据，可以使用标准库 functools.lru_cache：

"""使用有容量上限的 LRU 缓存保存报表结果。"""

from functools import lru_cache


def load_report(cache_key: str) -> bytes:
    """从可信数据源加载报表；真实实现应设置 I/O 超时。"""
    return (cache_key.encode("utf-8") + b":") * 4096


@lru_cache(maxsize=512)
def query_report(cache_key: str) -> bytes:
    """返回报表，并把进程内缓存限制为最多 512 个键。"""
    if len(cache_key) > 256:
        raise ValueError("cache_key 长度不能超过 256")
    return load_report(cache_key)

maxsize=512 限制的是条目数，不是字节数。若不同条目的体积差异很大，应使用支持权重、TTL 和并发控制的缓存实现，或者把数据放入有内存上限的外部缓存。容量应从以下公式估算，而不是照抄示例值：

缓存预算 ≈ 可用于缓存的内存 × 安全系数
最大条目数 ≈ 缓存预算 ÷ 单条目的高分位体积

还要考虑并发 miss。多个请求同时查询同一个新键时，lru_cache 可以保护内部数据结构，但不保证底层加载函数只执行一次。高成本查询需要额外的 single-flight 或按键锁，并为等待设置超时。

验证修复

用单元测试确认容量边界、命中行为和输入限制：

"""验证报表缓存的容量和输入边界。"""

import unittest
from unittest.mock import patch

from report_cache import query_report


class ReportCacheTest(unittest.TestCase):
    """验证缓存不会随唯一键数量无限增长。"""

    def setUp(self) -> None:
        query_report.cache_clear()

    def test_cache_size_is_bounded(self) -> None:
        for index in range(2_000):
            query_report(f"tenant-{index}")

        info = query_report.cache_info()
        self.assertEqual(info.maxsize, 512)
        self.assertLessEqual(info.currsize, 512)

    def test_repeated_key_hits_cache(self) -> None:
        with patch("report_cache.load_report", return_value=b"report") as loader:
            self.assertEqual(query_report("tenant-a"), b"report")
            self.assertEqual(query_report("tenant-a"), b"report")

        loader.assert_called_once_with("tenant-a")
        self.assertEqual(query_report.cache_info().hits, 1)

    def test_rejects_oversized_key(self) -> None:
        with self.assertRaisesRegex(ValueError, "长度不能超过"):
            query_report("x" * 257)


if __name__ == "__main__":
    unittest.main()

运行：

python -m unittest -v test_report_cache.py

上线前还应进行长时间稳定负载验证：缓存 currsize 到达上限后应进入平台期，hits 应持续增加，RSS 不应在相同吞吐下不断创新高。若 Python 跟踪内存已稳定但 RSS 仍持续上涨，需要转向原生分配诊断，而不是继续修改缓存。

预防措施

  1. 所有进程内缓存必须声明容量、淘汰策略、所有者和失效条件，禁止裸全局字典承担永久缓存职责。
  2. 为缓存暴露条目数、命中率、淘汰数和估算字节数；只看命中率无法发现容量失控。
  3. 为队列配置最大长度和背压策略，告警同时覆盖队列深度、最老消息年龄与消费速率。
  4. 把稳定负载下的内存斜率纳入回归测试，避免只验证短时峰值。
  5. 堆快照可能包含文件路径和业务对象痕迹，应限制诊断入口权限、输出范围与保留时间，不对公网开放。
  6. 不要把定时重启当作治理。它只能降低故障频率，还会掩盖容量模型和引用生命周期问题。

总结

定位 Python 常驻服务的内存增长，关键不是频繁调用垃圾回收，而是建立证据链：先确认 RSS 的持续趋势，再比较 tracemalloc 的两个快照，找到仍在增长的 Python 分配位置，并结合缓存条目数、队列深度等业务指标验证引用来源。

对于无界缓存，修复目标是把资源边界写进设计：限制容量，明确淘汰策略，校验输入，并用测试和监控证明内存会进入平台期。如果 tracemalloc 与 RSS 的走势不一致，也应及时停止在 Python 堆中盲查，转向原生扩展和分配器层面的诊断。