Python技术迷

监控 Python 内存使用情况和代码执行时间

有一次半夜快十二点,我在公司楼下啃着个凉掉的肉夹馍,手机里监控一直在滴滴响——一条 Python 定时任务又把机器内存打满,被运维同学一脚给干掉了。那天我就下决心:以后写 Python,时间和内存一定要“算清楚”,不能全靠感觉。

今天就聊聊,怎么用 Python 比较靠谱地监控代码执行时间和内存占用,都是日常能直接用起来的办法,顺手附上完整代码。

先把“时间”量出来:从 print 到专业一点的计时器

最原始那种,你肯定也写过:

import time

start = time.time()
# TODO: 你的业务逻辑,比如大循环
time.sleep(1.23)
end = time.time()

print(f"耗时: {end - start:.4f} 秒")

能用,但有两个问题:

  1. time.time() 分辨率一般般,短代码测不准
  2. 到处写 start = ...end = ...,一坨一坨的很烦

时间这块,推荐你直接用 time.perf_counter(),这是专门用来做性能计数器的,精度更高,受系统时间调整影响也更小:

import time

start = time.perf_counter()
# 模拟一段代码
time.sleep(0.123)
end = time.perf_counter()

print(f"耗时: {(end - start) * 1000:.2f} ms")

但真正好用的,其实是封装成上下文管理器或者装饰器,这样不用每次写一遍开始/结束。

一个最常用的计时上下文:

import time
from contextlib import contextmanager

@contextmanager
deftime_block(tag: str = ""):
    start = time.perf_counter()
try:
yield
finally:
        cost = (time.perf_counter() - start) * 1000
        print(f"[TIMER] {tag} 耗时 {cost:.2f} ms")


if __name__ == "__main__":
with time_block("处理一批订单"):
        total = 0
for i in range(10_0000):
            total += i ** 2

你只要把“可疑耗时”的代码塞进 with time_block("xxx") 里面,一跑就知道哪一块在拖后腿了。

如果你是想监控函数级别的时间,可以上装饰器:

import time
from functools import wraps

deftimeit(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


@timeit
defheavy_task(n: int = 10_0000):
    total = 0
for i in range(n):
        total += i ** 2
return total


if __name__ == "__main__":
    heavy_task(50_0000)

这个装饰器以后可以直接贴在业务函数上,哪天线上慢了,先打开日志级别,看一眼就心里有数了。

想更严谨一点:用 timeit 做基准测试

有时候你不是想知道“这段逻辑大概要跑多久”,而是想比较两种写法谁更快,这时候就轮到 timeit 上场了。

最简单的用法是命令行:

python -m timeit "sum(range(1000))"

在代码里也可以直接用:

import timeit

defway1():
return [i * 2for i in range(1000)]

defway2():
    res = []
for i in range(1000):
        res.append(i * 2)
return res

if __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 这边,常用的有三种层次:

  1. 看整体进程占了多少内存
  2. 看某一段代码运行前后,内存变化多少
  3. 精细到逐行代码,哪一行在猛涨

咱一个个来。

粗粒度:用 psutil 看进程内存

先上个简单粗暴的:

import os
import psutil
import time

defget_memory_mb() -> float:
    process = psutil.Process(os.getpid())
return process.memory_info().rss / 1024 / 1024# MB


if __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 看“谁”在占内存

tracemalloc 是 Python 标准库自带的内存分配跟踪工具,用起来其实不难:

import tracemalloc

defmake_leak():
    data = []
for i in range(100_000):
        data.append(str(i) * 10)
return data

if __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

哪一行代码导致内存上涨,一眼就看出来了。 线上疑似有“内存泄漏”(其实大多是自己一直堆数据不释放)的,可以用这个手段快速缩小范围。

逐行看:memory_profiler 精准定位

如果你想细到“这一行多了几 MB,那一行少了几 MB”,可以用 memory_profiler 这个三方库。

安装:

pip install memory_profiler

代码里这样用:

from memory_profiler import profile

@profile
defprocess_data():
    data = [i for i in range(10_0000)]
    data2 = [str(i) for i in data]
del data
return data2

if __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 os
import time
import psutil
from functools import wraps

defprofile_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 result
return wrapper
return 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 total


if __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] = user
return user

如果你线上有几十万、上百万用户,且从来不淘汰,这个 _cache 会长得很可怕。 这类场景就很适合:

  • 用 functools.lru_cache(maxsize=1024)
  • 或者自己加个“最多缓存 N 条”的逻辑
  • 实在不行,至少用上面那个内存监控装饰器,观察一下增长趋势

很多时候,不是 Python 内存管理差,而是我们“太会存东西,还舍不得删”。

说白了,监控时间和内存这件事,不是为了写那种花里胡哨的“性能优化故事”,而是让你在开发阶段就心里有底:这段代码大概会慢到什么程度,会不会悄悄把机器吃光。

你完全可以现在新建一个 perf_utils.py,把上面这几个东西直接复制进去:

  • time_block 上下文
  • timeit 装饰器
  • profile_time_memory 二合一装饰器
  • 一个 get_memory_mb() 小函数

以后哪个模块有点“看着就不太放心”,套一下,看一眼日志,再决定要不要动刀子,成本很低。

行了,先这样,我去给自己的几个老脚本也把监控补一补,免得哪天半夜又被电话吵醒…