线上有个 Python 定时任务跑得越来越慢。

一开始没人觉得是代码问题，毕竟数据量也没涨多少，CPU 也没打满。

但是监控里能看到一个很奇怪的现象：

任务执行时间从几分钟慢慢涨到了半小时。

看日志：

2026-09-18 02:00:01 task start
2026-09-18 02:21:43 fetch_user finished
2026-09-18 02:29:17 build_report finished
2026-09-18 02:31:02 task finished

这种问题我一般不会先去重构业务代码。

先找慢在哪里。

Python 这种脚本型语言，很多性能问题不是算法多复杂，而是一些地方反复执行了没必要的东西：

一个函数被调用几万次，但每次都重复计算；

一个接口慢，不知道到底慢在哪一步；

批处理脚本里面偷偷创建了大量对象；

数据库查询耗时，没有统一统计；

某段代码偶发卡顿，日志又没留下。

这类问题，其实用几个装饰器就能快速把现场信息拿出来。

不是说加个装饰器性能直接起飞。

装饰器更多是帮你定位瓶颈，然后针对性优化。

我平时排查 Python 性能问题，会先放这几个。

1. 统计函数耗时：先找到真正的慢点

最常用的就是耗时统计。

不要看到接口慢，就直接优化 SQL。

有时候 SQL 只占 200ms，真正慢的是 Python 自己处理数据。

比如下面这个：

import time
from functools import wraps


def cost_time(func):
    @wraps(func)
    def wrapper(*args, **kwargs):
        start = time.perf_counter()

        result = func(*args, **kwargs)

        cost = time.perf_counter() - start

        print(
            f"[slow-check] {func.__name__} cost={cost:.4f}s"
        )

        return result

    return wrapper


@cost_time
def rebuild_daily_report(records):
    result = []

    for item in records:
        result.append({
            "id": item["id"],
            "amount": item["amount"] * 1.05
        })

    return result

执行：

[slow-check] rebuild_daily_report cost=8.4312s

看到这个结果，下一步才有方向。

比如：

8秒。

那继续拆。

是查询数据慢？

是循环处理慢？

还是序列化慢？

不要一上来就优化。

很多代码优化失败，就是因为连瓶颈在哪里都不知道。

2. 统计调用次数：小函数可能才是罪魁祸首

有些函数看起来很简单。

但是放在循环里面调用几十万次，问题就来了。

比如：

def normalize_phone(phone):
    return phone.replace("-", "")

单看没问题。

但是：

for user in users:
    normalize_phone(user.phone)

users 有 500 万条。

这个函数就执行了 500 万次。

可以先加个调用统计：

from functools import wraps


def count_call(func):
    total = 0

    @wraps(func)
    def wrapper(*args, **kwargs):
        nonlocal total

        total += 1

        if total % 10000 == 0:
            print(
                f"{func.__name__} called {total} times"
            )

        return func(*args, **kwargs)

    return wrapper


@count_call
def normalize_phone(phone):
    return phone.replace("-", "")

跑一次：

normalize_phone called 10000 times
normalize_phone called 20000 times
normalize_phone called 30000 times
...
normalize_phone called 5000000 times

这个时候你就知道：

优化方向不是函数内部。

而是调用方式。

比如：

提前批量处理。

或者缓存结果。

3. 给重复计算加缓存：别让 CPU 白干活

这个场景线上特别常见。

比如根据规则生成价格：

def calculate_discount(user_level, amount):
    # 很复杂的规则计算
    ...

同样的用户等级，同样的金额，一天调用几百万次。

每次重新算。

CPU 就在重复劳动。

简单场景可以直接用：

from functools import lru_cache


@lru_cache(maxsize=2048)
def calculate_discount(level, amount):
    print("real calculate")

    return amount * 0.85

第一次：

real calculate

第二次：

直接返回缓存。

不过这里有个坑。

缓存不是免费的。

如果参数无限变化：

calculate_discount(
    user_id,
    timestamp
)

这种缓存基本没意义。

甚至可能把内存撑爆。

线上我一般会看：

参数是不是稳定；

命中率怎么样；

生命周期多长。

别看到缓存两个字就往上加。

4. 统计异常：别让失败请求悄悄消失

很多脚本任务最麻烦的是：

跑失败了。

但是不知道为什么。

尤其是批处理：

for item in data:
    process(item)

中间某条数据异常，整个任务挂掉。

可以统一记录：

import traceback
from functools import wraps


def catch_error(func):

    @wraps(func)
    def wrapper(*args, **kwargs):

        try:
            return func(*args, **kwargs)

        except Exception as e:

            print(
                f"""
function={func.__name__}
error={e}
trace=
{traceback.format_exc()}
"""
            )

            raise

    return wrapper

使用：

@catch_error
def import_customer(row):
    save_customer(row)

日志：

function=import_customer

error=duplicate key

trace:
File "worker.py", line 88
save_customer(row)

至少知道死在哪里。

生产环境排查，最怕的不是报错。

是没有线索。

5. 统计资源占用：别只看执行时间

有些代码执行不慢。

但是吃内存。

比如导入 Excel：

rows = load_all_rows()

for row in rows:
    handle(row)

10万数据没问题。

1000万数据直接爆。

可以简单包一下：

import os
import psutil
from functools import wraps


def memory_watch(func):

    @wraps(func)
    def wrapper(*args, **kwargs):

        process = psutil.Process(
            os.getpid()
        )

        before = process.memory_info().rss

        result = func(*args, **kwargs)

        after = process.memory_info().rss

        print(
            f"{func.__name__} memory +"
            f"{(after-before)/1024/1024:.2f}MB"
        )

        return result

    return wrapper

运行：

load_excel memory +732.56MB

看到这个数字，基本不用猜了。

问题不是 CPU。

是一次性把数据全部塞进内存。

改成：

for chunk in read_excel_stream():
    process(chunk)

通常效果比调参数明显。

这几个装饰器，我一般怎么组合？

实际项目里不会全部开。

线上排查时：

先开耗时：

@cost_time
def sync_order():
    ...

发现某一步慢。

再加调用统计：

@count_call
def query_user():
    ...

发现调用次数异常。

最后针对问题改：

重复计算 → 缓存

循环调用 → 批处理

大对象 → 流式处理

外部接口慢 → 并发或者异步

性能优化不是把所有代码改一遍。

大部分时候，是找到那个偷偷消耗资源的小地方。

Python 最大的问题不是跑不快。

而是不知道它为什么慢。

先把现场数据拿出来，再动刀。

这比凭感觉重构靠谱多了。

 更多相关技术内容咨询欢迎前往并持续关注好学星城论坛了解详情。

想高效系统的学习Python编程语言，推荐大家关注一个微信公众号：Python编程学习圈。每天分享行业资讯、技术干货供大家阅读，关注即可免费领取整套Python入门到进阶的学习资料以及教程，感兴趣的小伙伴赶紧行动起来吧。