Python 性能优化应该从哪里下手才有效?
先确认:慢在哪里,别先改代码
Python 性能优化最容易走偏的地方,是一上来就把 for 循环改成列表推导式,或者把所有地方都加缓存。真正有效的顺序应该反过来:先测量,再定位瓶颈,最后只改最值得改的那一小段。
一个实用判断是:如果你说不清某段代码慢了多少、占总耗时多少、内存峰值在哪里,那现在还不是优化的时候。先把基准数据拿到手。
用 timeit 做小片段基准测试
timeit 适合比较一小段代码的耗时,比如列表查找和集合查找、字符串拼接和 join。它会重复执行代码,减少单次测量的抖动。
pythonimport timeit setup = "data = list(range(10000)); target = 9999" list_time = timeit.timeit("target in data", setup=setup, number=10000) setup = "data = set(range(10000)); target = 9999" set_time = timeit.timeit("target in data", setup=setup, number=10000) print(list_time, set_time)
注意不要把 @timeit.timeit 当装饰器用,timeit.timeit() 返回的是耗时数字,不是函数包装器。函数级别的简单计时可以自己写装饰器,或者用 time.perf_counter():
pythonimport time from functools import wraps def timed(func): @wraps(func) def wrapper(*args, **kwargs): start = time.perf_counter() try: return func(*args, **kwargs) finally: cost = time.perf_counter() - start print(f"{func.__name__}: {cost:.4f}s") return wrapper
用 cProfile 看整段程序的热点
cProfile 适合回答“到底哪个函数占了最多时间”。它给出的 cumtime 很有用,表示函数及其子调用的累计耗时。
pythonimport cProfile import pstats def main(): data = range(1_000_000) sum(x * x for x in data) cProfile.run("main()", "profile.stats") stats = pstats.Stats("profile.stats") stats.strip_dirs().sort_stats("cumtime").print_stats(20)
如果是线上服务,别只在本地跑一个玩具输入。可以在压测环境里收集真实请求路径,必要时用 py-spy 这类采样工具,避免为了分析性能反而把服务拖慢。
内存和逐行分析也要看
有些慢不是 CPU 慢,而是内存涨得太快、频繁触发 GC、或者一次性加载了太多数据。可以用 memory_profiler 看函数执行期间的内存变化:
python# pip install memory-profiler from memory_profiler import profile @profile def load_data(): data = [i for i in range(1_000_000)] return sum(data) load_data()
line_profiler 更适合分析“这个函数里到底哪一行慢”:
python# pip install line_profiler from line_profiler import LineProfiler def build_result(): result = [] for i in range(10000): result.append(i * 2) return sum(result) lp = LineProfiler() lp_wrapper = lp(build_result) lp_wrapper() lp.print_stats()
先用 cProfile 找函数,再用 line_profiler 看函数内部,通常比一开始就逐行分析更省时间。
算法比语法糖更值钱
Python 单行写法再漂亮,也救不了错误的复杂度。一个 O(n²) 的逻辑,在数据量小的时候看不出来,数据一大就会突然变成事故现场。
用 set 或 dict 降低查找成本
找重复元素时,双重循环是典型的慢写法:
python# O(n²) def find_duplicates_slow(items): duplicates = [] for i in range(len(items)): for j in range(i + 1, len(items)): if items[i] == items[j] and items[i] not in duplicates: duplicates.append(items[i]) return duplicates
用集合记录状态,复杂度可以降到 O(n):
python# O(n) def find_duplicates_fast(items): seen = set() duplicates = set() for item in items: if item in seen: duplicates.add(item) else: seen.add(item) return list(duplicates)
这种优化比把变量名缩短、把循环换成推导式有效得多。
减少重复计算
两两计算距离时,如果 (i, j) 和 (j, i) 都算一遍,工作量直接翻倍:
pythondef calculate_distances(points): distances = [] for i in range(len(points)): for j in range(i + 1, len(points)): dx = points[j][0] - points[i][0] dy = points[j][1] - points[i][1] distances.append((dx * dx + dy * dy) ** 0.5) return distances
还能继续优化吗?要看业务是否真的需要保存所有距离。如果只是找最小值,就不必把中间结果全部放进列表。
数据结构会直接影响速度和内存
Python 的列表、集合、字典、元组都很好用,但适用场景不一样。性能问题经常不是“Python 不够快”,而是数据结构选错了。
列表适合顺序遍历,集合适合成员判断
pythonusers = list(range(100000)) user_set = set(users) 99999 in users # 需要顺序扫描 99999 in user_set # 哈希查找,平均 O(1)
如果一段代码每天要做几百万次成员判断,把列表换成集合通常很划算。但集合会占用更多内存,也不保留重复元素,不能只看速度。
生成器适合流式处理
一次性构造大列表,会把所有结果放进内存。只需要逐个消费时,用生成器更稳。
pythondef squares(n): for i in range(n): yield i * i total = sum(squares(1_000_000))
列表推导式不是坏东西。需要多次遍历、随机访问、切片时,列表更方便。生成器的优势在于“边算边用”,不是所有场景都要换成生成器。
slots 能省内存,但会牺牲灵活性
当你需要创建大量结构相同的小对象时,__slots__ 可以减少每个对象的 __dict__ 开销。
pythonclass Point: __slots__ = ("x", "y") def __init__(self, x, y): self.x = x self.y = y
它的代价是不能随意添加新属性,和某些依赖 __dict__ 的库也可能不兼容。几十个对象没必要上 __slots__,几十万、几百万个对象才值得认真比较。
多用内置函数,但别迷信“一行代码”
sum()、max()、min()、sorted()、any()、all() 这类内置函数通常跑得更快,因为核心循环在 C 层实现。
pythondef manual_sum(items): total = 0 for item in items: total += item return total builtin_sum = sum
不过可读性也要算账。为了把三段清晰的业务逻辑压成一个嵌套表达式,最后没人敢改,这不是优化,是埋雷。
I/O 优化先看批量、缓冲和异步
文件、网络、数据库这些 I/O 场景,瓶颈通常不在 Python 运算本身,而在等待外部系统响应。优化方向也不一样。
文件读写尽量批量处理
逐行写入不是不能用,但如果数据已经在内存里,批量写入会少很多系统调用。
pythondef write_lines_fast(filename, lines): with open(filename, "w", encoding="utf-8") as f: f.write("\n".join(lines))
读取大文件时,不要为了“省事”直接 read() 全部加载:
pythondef process_large_file(filename): with open(filename, "r", encoding="utf-8") as f: for line in f: yield line.strip()
缓冲大小可以调,但别在文本模式下写 buffering=0,无缓冲只允许二进制模式。大多数情况下,默认缓冲已经够用。
网络 I/O 适合异步或线程池
如果任务主要在等网络返回,异步 I/O 或线程池都能提高吞吐。下面是 aiohttp 的常见写法:
pythonimport asyncio import aiohttp async def fetch(session, url): async with session.get(url, timeout=10) as response: response.raise_for_status() return await response.text() async def fetch_all(urls): async with aiohttp.ClientSession() as session: tasks = [fetch(session, url) for url in urls] return await asyncio.gather(*tasks)
真实项目里还要限制并发数,不要同时打出几千个请求把对方服务或自己机器打满。
并发要分清 CPU 密集还是 I/O 密集
Python 有 GIL,线程不适合加速纯 CPU 计算,但很适合等待网络、磁盘、数据库这类 I/O。CPU 密集任务通常考虑多进程、C 扩展、NumPy、Cython,或者把计算交给更合适的服务。
I/O 密集用 ThreadPoolExecutor
pythonfrom concurrent.futures import ThreadPoolExecutor import requests def download(url): response = requests.get(url, timeout=10) response.raise_for_status() return len(response.content) with ThreadPoolExecutor(max_workers=8) as executor: sizes = list(executor.map(download, urls))
CPU 密集用 ProcessPoolExecutor
pythonfrom concurrent.futures import ProcessPoolExecutor def compute(chunk): return sum(x * x for x in chunk) with ProcessPoolExecutor(max_workers=4) as executor: result = sum(executor.map(compute, chunks))
多进程有序列化和进程通信成本。数据块太小、任务太轻时,开进程可能比单进程更慢。
缓存能救热点,也能制造脏数据
缓存适合重复计算、重复查询、重复读取的场景。它不适合掩盖慢 SQL,也不适合缓存所有东西。
lru_cache 适合纯函数
pythonfrom functools import lru_cache @lru_cache(maxsize=1024) def get_user_permissions(user_id): return load_permissions_from_db(user_id)
lru_cache 的参数必须可哈希;结果最好由参数唯一决定。如果函数依赖当前时间、登录态、外部可变配置,就要谨慎。
Redis 适合跨进程缓存
pythonimport json import redis r = redis.Redis(host="localhost", port=6379, db=0) def set_cache(key, value, ttl=300): r.setex(key, ttl, json.dumps(value, ensure_ascii=False)) def get_cache(key): raw = r.get(key) return json.loads(raw) if raw else None
示例里用 JSON 是为了避免随手 pickle.loads() 带来的安全风险。只有在数据来源完全可信时,才考虑 pickle。
缓存一定要设计失效策略。TTL 太长会读到旧数据,太短又没有命中率;热点 key 还要考虑击穿、雪崩和并发回源。
字符串和对象创建也有细节
大量字符串拼接时,join() 通常比循环里的 + 更合适:
pythondef build_text(parts): return "".join(parts)
格式化字符串优先用 f-string,清晰也快:
pythonmessage = f"Name: {name}, Age: {age}"
处理大小写、替换、分割时,先看字符串内置方法。手写循环不仅慢,也更容易漏掉边界情况。
数据库优化别只盯 Python 代码
很多接口慢,Python 只背了一半锅,另一半在数据库。
连接池减少连接开销
pythonfrom sqlalchemy import create_engine engine = create_engine( "postgresql://user:password@localhost/dbname", pool_size=10, max_overflow=5, pool_pre_ping=True, )
连接池不是越大越好。池子太大会把数据库连接打满,反而拖垮整体服务。
批量写入比逐条写入稳定
pythondef insert_fast(cursor, items): cursor.executemany( "INSERT INTO events(name, value) VALUES (%s, %s)", [(item.name, item.value) for item in items], )
ORM 也通常提供 bulk insert,但要确认它是否跳过了模型钩子、默认值、校验逻辑。
索引要服务查询条件
sqlCREATE INDEX idx_users_name ON users(name); SELECT * FROM users WHERE name = 'Alice';
如果查询写成 WHERE LOWER(name) = 'alice',普通 name 索引可能用不上。要么调整查询方式,要么建立对应的函数索引。遇到慢查询,先看 EXPLAIN,不要凭感觉加索引。
监控决定优化有没有持续价值
本地优化通过之后,还要看线上是否真的变快。至少记录这些指标:接口耗时、P95/P99 延迟、错误率、内存峰值、队列长度、缓存命中率、数据库慢查询。
一个简单的耗时日志装饰器可以这样写:
pythonimport logging import time from functools import wraps logger = logging.getLogger(__name__) def log_cost(func): @wraps(func) def wrapper(*args, **kwargs): start = time.perf_counter() try: return func(*args, **kwargs) finally: cost = time.perf_counter() - start logger.info("%s cost %.4fs", func.__name__, cost) return wrapper
更完整的项目会把指标打到 Prometheus、StatsD、OpenTelemetry 或 APM 系统里。单次平均值不够,P95/P99 更能反映用户真实体感。
优化时必须保留的取舍
性能优化不是越快越好,至少要同时看四件事:
- 可读性:快 3%,但代码复杂一倍,通常不值得。
- 正确性:缓存、并发、批量写入都会引入一致性和边界问题。
- 内存:用空间换时间很常见,但容器内存有限。
- 可观测性:没有监控,优化效果很快会被后续改动吃掉。
如果只能记住一个顺序,那就是:先用 timeit、cProfile、内存分析和线上指标确认瓶颈;再从算法、数据结构、I/O、并发、缓存、数据库这些高收益位置下手;最后用同样的指标复测。Python 性能优化最怕“感觉上更快”,真正可靠的是前后两组数据。