有一次半夜快十二点,我在公司楼下啃着个凉掉的肉夹馍,手机里监控一直在滴滴响——一条 Python 定时任务又把机器内存打满,被运维同学一脚给干掉了。那天我就下决心:以后写 Python,时间和内存一定要“算清楚”,不能全靠感觉。
今天就聊聊,怎么用 Python 比较靠谱地监控代码执行时间和内存占用,都是日常能直接用起来的办法,顺手附上完整代码。
最原始那种,你肯定也写过:
import timestart = time.time()# TODO: 你的业务逻辑,比如大循环time.sleep(1.23)end = time.time()print(f"耗时: {end - start:.4f} 秒")能用,但有两个问题:
time.time() 分辨率一般般,短代码测不准start = ...end = ...,一坨一坨的很烦时间这块,推荐你直接用 time.perf_counter(),这是专门用来做性能计数器的,精度更高,受系统时间调整影响也更小:
import timestart = time.perf_counter()# 模拟一段代码time.sleep(0.123)end = time.perf_counter()print(f"耗时: {(end - start) * 1000:.2f} ms")但真正好用的,其实是封装成上下文管理器或者装饰器,这样不用每次写一遍开始/结束。
一个最常用的计时上下文:
import timefrom contextlib import contextmanager@contextmanagerdeftime_block(tag: str = ""): start = time.perf_counter()try:yieldfinally: cost = (time.perf_counter() - start) * 1000 print(f"[TIMER] {tag} 耗时 {cost:.2f} ms")if __name__ == "__main__":with time_block("处理一批订单"): total = 0for i in range(10_0000): total += i ** 2你只要把“可疑耗时”的代码塞进 with time_block("xxx") 里面,一跑就知道哪一块在拖后腿了。
如果你是想监控函数级别的时间,可以上装饰器:
import timefrom functools import wrapsdeftimeit(func): @wraps(func)defwrapper(*args, **kwargs): start = time.perf_counter()try:return func(*args, **kwargs)finally: cost = (time.perf_counter() - start) * 1000 print(f"[FUNC TIMER] {func.__name__} 耗时 {cost:.2f} ms")return wrapper@timeitdefheavy_task(n: int = 10_0000): total = 0for i in range(n): total += i ** 2return totalif __name__ == "__main__": heavy_task(50_0000)这个装饰器以后可以直接贴在业务函数上,哪天线上慢了,先打开日志级别,看一眼就心里有数了。
有时候你不是想知道“这段逻辑大概要跑多久”,而是想比较两种写法谁更快,这时候就轮到 timeit 上场了。
最简单的用法是命令行:
python -m timeit "sum(range(1000))"在代码里也可以直接用:
import timeitdefway1():return [i * 2for i in range(1000)]defway2(): res = []for i in range(1000): res.append(i * 2)return resif __name__ == "__main__": t1 = timeit.timeit(way1, number=10000) t2 = timeit.timeit(way2, number=10000) print(f"way1: {t1:.4f} 秒, way2: {t2:.4f} 秒")这个适合你纠结“写法 A 还是写法 B”纠结半天的时候,直接测,别吵架。
#@@# 再看内存:别等到 OOM 才后悔
时间比较直观,内存一般是“出了事才想起来看”。Python 这边,常用的有三种层次:
咱一个个来。
先上个简单粗暴的:
import osimport psutilimport timedefget_memory_mb() -> float: process = psutil.Process(os.getpid())return process.memory_info().rss / 1024 / 1024# MBif __name__ == "__main__": print(f"启动时内存: {get_memory_mb():.2f} MB") data = []for i in range(10): data.append("x" * 10_0000) # 100 KB * 10 print(f"第 {i+1} 次追加后内存: {get_memory_mb():.2f} MB") time.sleep(0.5)这个非常适合你排查那种“写了个 while True 的守护脚本,跑久了内存慢慢往上涨”的场景。 加几行 print,很快就知道是不是你自己不停往 list / dict 里面塞东西不清理。
tracemalloc 是 Python 标准库自带的内存分配跟踪工具,用起来其实不难:
import tracemallocdefmake_leak(): data = []for i in range(100_000): data.append(str(i) * 10)return dataif __name__ == "__main__": tracemalloc.start() snapshot1 = tracemalloc.take_snapshot() _ = make_leak() snapshot2 = tracemalloc.take_snapshot() top_stats = snapshot2.compare_to(snapshot1, 'lineno') print("内存增长 Top 5:")for stat in top_stats[:5]: print(stat)跑完你会看到类似:
/path/to/demo.py:5: size=12.3 MiB (+12.3 MiB), count=100000, average=128 B哪一行代码导致内存上涨,一眼就看出来了。 线上疑似有“内存泄漏”(其实大多是自己一直堆数据不释放)的,可以用这个手段快速缩小范围。
如果你想细到“这一行多了几 MB,那一行少了几 MB”,可以用 memory_profiler 这个三方库。
安装:
pip install memory_profiler代码里这样用:
from memory_profiler import profile@profiledefprocess_data(): data = [i for i in range(10_0000)] data2 = [str(i) for i in data]del datareturn data2if __name__ == "__main__": process_data()执行时加上:
python -m memory_profiler your_script.py你就能看到类似这样的输出(这里只是示意):
Line # Mem usage Increment Occurrences Line Contents============================================================= 3 10.0 MiB 10.0 MiB 1 @profile 4 def process_data(): 5 20.5 MiB 10.5 MiB 1 data = [i for i in range(10_0000)] 6 32.0 MiB 11.5 MiB 1 data2 = [str(i) for i in data] 7 20.5 MiB -11.5 MiB 1 del data哪一行在飙内存,哪一行释放了多少,一清二楚,非常适合改老项目时“摸黑排查”。
实际工作里大多是这样的场景:你有一个重要函数,既担心它太慢,又担心它太吃内存。那就干脆写个“二合一”的装饰器,套上去就行。
import osimport timeimport psutilfrom functools import wrapsdefprofile_time_memory(tag: str = ""):""" 打印函数耗时 + 内存变化 """defdecorator(func): @wraps(func)defwrapper(*args, **kwargs): process = psutil.Process(os.getpid()) mem_before = process.memory_info().rss / 1024 / 1024 t_start = time.perf_counter() result = func(*args, **kwargs) t_cost = (time.perf_counter() - t_start) * 1000 mem_after = process.memory_info().rss / 1024 / 1024 delta = mem_after - mem_before name = tag or func.__name__ print(f"[PROFILE] {name}: 耗时 {t_cost:.2f} ms, "f"内存变化 {delta:+.2f} MB (前 {mem_before:.2f} -> 后 {mem_after:.2f})" )return resultreturn wrapperreturn decorator@profile_time_memory("批量处理订单")defprocess_orders(n: int = 100_000): orders = [{"id": i, "amount": i % 100} for i in range(n)] total = sum(o["amount"] for o in orders)return totalif __name__ == "__main__": process_orders(200_000)这个东西的好处就是:
我一般是只在“核心几个链路”上加这种东西,比如: 批量导入、报表计算、推荐结果生成之类,一旦慢或炸内存,影响就比较大。
讲两个我自己踩过的:
1)无脑 list,忘了生成器
比如你在读一个巨大的文件:
# 不太友好lines = open("big.log").readlines()for line in lines: handle(line)readlines() 会直接把整个文件一次性读进内存,日志一大直接爆。 更温柔一点的写法是:
with open("big.log", "r", encoding="utf-8") as f:for line in f: # 逐行生成 handle(line)配合上面的 profile_time_memory,你一跑就能看到内存变化非常平稳。
2)缓存 / 全局变量越写越大
有时候你以为自己在“优化性能”,结果变成在“优化 OOM”。
_cache = {}defget_user(user_id):if user_id in _cache:return _cache[user_id]# 这里假设是查库 user = {"id": user_id, "name": f"user-{user_id}"} _cache[user_id] = userreturn user如果你线上有几十万、上百万用户,且从来不淘汰,这个 _cache 会长得很可怕。 这类场景就很适合:
functools.lru_cache(maxsize=1024)很多时候,不是 Python 内存管理差,而是我们“太会存东西,还舍不得删”。
说白了,监控时间和内存这件事,不是为了写那种花里胡哨的“性能优化故事”,而是让你在开发阶段就心里有底:这段代码大概会慢到什么程度,会不会悄悄把机器吃光。
你完全可以现在新建一个 perf_utils.py,把上面这几个东西直接复制进去:
time_block 上下文timeit 装饰器profile_time_memory 二合一装饰器get_memory_mb() 小函数以后哪个模块有点“看着就不太放心”,套一下,看一眼日志,再决定要不要动刀子,成本很低。
行了,先这样,我去给自己的几个老脚本也把监控补一补,免得哪天半夜又被电话吵醒…
-END-
我为大家打造了一份RPA教程,完全免费:songshuhezi.com/rpa.html
虎哥作为一名老码农,整理了全网最全《python高级架构师资料合集》,总量高达650GB