4. 全局变量与缓存

内存泄漏这事儿，在Python里其实挺隐蔽的。

不像C语言你得自己malloc/free，Python有GC（垃圾回收）兜底。但GC不是万能的。我见过太多项目，跑着跑着内存飙到几个G，最后OOM被系统kill掉。查下来，十有八九是全局变量或者缓存惹的祸。

这一节，我们就专门聊聊这个。

4.1 全局列表/字典：最隐蔽的“内存黑洞”

先看一个我实际遇到的案例。

有个数据处理服务，核心逻辑是不断从消息队列里拿数据，处理后存到数据库。代码大概长这样：

# 全局缓存，用来暂存处理结果
processed_data = []

def handle_message(msg):
    result = process(msg)
    processed_data.append(result)
    # 这里忘记清理了
    if len(processed_data) > 10000:
        # 本意是批量写入后清空
        batch_write_to_db(processed_data)
        # 但！没有清空列表！
        # processed_data.clear()  # 这行被注释掉了

你猜怎么着？

这个服务跑了大概6个小时，内存从200MB涨到了8GB。运维报警了，我上去一看，processed_data这个列表里存了将近500万条记录。每条记录是一个字典，平均2KB，算下来就是10GB。

嗯，这就是典型的“只增不减”。

⚠️ 避坑指南：

我曾经在一个金融交易系统里也犯过类似的错。全局字典用来缓存用户会话，但用户登出后没有删除对应的key。结果呢？内存里躺着几十万个僵尸会话，每个会话还带着一堆交易快照。最后系统每4小时重启一次，直到我加上了session.pop(user_id)才解决。

正确的做法是什么？

  • 用完即清：批量操作后，记得.clear()或者重新赋值= []
  • 限制大小：用collections.deque(maxlen=N)代替普通列表
  • 定时清理：写个后台线程，定期检查全局字典的大小，超过阈值就清理过期数据

4.2 lru_cache的误用：你以为的“优化”其实是“毒药”

@lru_cache是Python标准库里的好东西。它能把函数的返回值缓存起来，避免重复计算。但好东西用错了地方，就是灾难。

我见过一个同事，把@lru_cache加在了一个处理用户上传文件的函数上：

from functools import lru_cache

@lru_cache(maxsize=None)  # 无限制缓存！
def process_upload(file_path: str) -> dict:
    # 读取文件，解析内容，返回结果
    with open(file_path, 'r') as f:
        data = f.read()
    return parse(data)

这个函数每天被调用几万次，每次传入的文件路径都不同。maxsize=None意味着缓存永不淘汰。结果呢？

  • 缓存里存了几万个文件的内容
  • 每个文件平均5MB
  • 内存直接爆了

为什么会这样？

因为lru_cache默认把参数作为key，返回值作为value。如果参数是文件路径这种唯一值，那每个新路径都会产生一个新缓存条目。而且maxsize=None意味着缓存无限增长。

核心原则：

  • lru_cache适合参数组合有限的场景（比如斐波那契数列、配置查询）
  • 永远不要对参数无限（如文件路径、时间戳、用户ID）的函数使用maxsize=None
  • 一定要设置maxsize，比如@lru_cache(maxsize=128)

4.3 如何正确清理？三种实用方案

说白了，清理缓存就三个思路：定时清、定量清、按需清。

方案一：定时清理（TTL）

用cachetools库的TTLCache，设置过期时间：

from cachetools import TTLCache

# 最多缓存1000条，每条存活300秒
cache = TTLCache(maxsize=1000, ttl=300)

def get_data(key):
    if key in cache:
        return cache[key]
    data = expensive_query(key)
    cache[key] = data
    return data

我个人习惯用这个方案。它最符合直觉：数据放久了就该失效。我在一个API网关项目里就用它缓存第三方接口的响应，TTL设了60秒，既减少了重复请求，又不会让内存爆炸。

方案二：定量清理（LRU）

用functools.lru_cache或者cachetools.LRUCache：

from functools import lru_cache

@lru_cache(maxsize=256)
def get_user_config(user_id: int) -> dict:
    # 从数据库读取用户配置
    return db.query(f"SELECT * FROM config WHERE user_id={user_id}")

这里maxsize=256意味着只保留最近使用的256个用户配置。超过这个数，最久没用的那个会被淘汰。

💡 小技巧：

如果你不确定该设多大，可以先设一个保守值（比如128），然后监控缓存命中率。命中率低于50%就调大，高于95%就调小。我一般用cache_info()方法查看：

print(get_user_config.cache_info())
# CacheInfo(hits=1234, misses=56, maxsize=256, currsize=256)

方案三：按需清理（手动控制）

有些场景下，你需要自己控制清理时机。比如用户登出时清理他的会话缓存：

class SessionCache:
    def __init__(self):
        self._cache = {}
    
    def set(self, user_id, session_data):
        self._cache[user_id] = session_data
    
    def get(self, user_id):
        return self._cache.get(user_id)
    
    def invalidate(self, user_id):
        self._cache.pop(user_id, None)  # 用户登出时调用
    
    def clear_expired(self, max_age_seconds=3600):
        # 定期清理过期会话
        now = time.time()
        expired = [uid for uid, data in self._cache.items() 
                   if now - data['last_access'] > max_age_seconds]
        for uid in expired:
            self._cache.pop(uid)

这个模式我在一个实时协作系统里用过。每个用户连接WebSocket时创建会话，断开时清理。配合一个每5分钟跑一次的定时任务清理僵尸会话，内存一直很稳定。

4.4 一张图总结：缓存清理的核心逻辑

下面这张图，是我自己总结的缓存清理决策流程。你写代码前，先走一遍这个流程，基本不会踩坑。

缓存清理决策流程 需要缓存吗？ 参数是否有限？ 是 用 lru_cache 否 手动管理缓存 设置 maxsize 设置 TTL 过期 手动 invalidate 监控命中率 调整 maxsize 核心原则：缓存必须有上限和淘汰策略 没有清理策略的缓存，就是内存泄漏

4.5 实战建议：写代码前先问自己三个问题

每次你要往全局变量里塞东西，或者用缓存装饰器时，先问自己：

  1. 这个数据会无限增长吗？——如果是，必须设上限
  2. 数据多久失效？——如果永远不失效，那就是内存泄漏
  3. 谁负责清理？——是自动淘汰，还是手动触发？

这三个问题想清楚了，你的缓存代码基本不会出大问题。

最后说一句：

我见过太多“临时方案”变成“永久负债”。那个被注释掉的.clear()，那个忘记设的maxsize，那个以为“反正数据不多”的全局字典……它们最终都会在凌晨3点让你从床上爬起来处理告警。

嗯，别问我怎么知道的。

可可精品资料库

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