副标题: 90%的人不知道,性能优化首先要找到瓶颈
痛点:为什么你的程序总是运行缓慢?
2025年某数据处理系统,处理100万条记录需要30分钟。问题出在哪?工程师盲目优化,没有先找到真正的瓶颈。
真相:性能优化第一步是profiling,用数据说话,不是凭感觉。
| 优化方式 | 效果 | 风险 |
| 盲目优化 | 低 | 高(可能引入bug) |
| 基于profiling | 高 | 低(数据驱动) |
一、性能分析基础
1.1 什么是profiling
Profiling是测量程序运行时的性能指标,包括:
-
●函数调用次数-
●函数执行时间-
●内存分配-
●CPU使用率-
1.2 cProfile基础
import cProfile
import pstats
def slow_function():
total = 0
for i in range(1000000):
total += i
return total
def fast_function():
return sum(range(1000000))
基本profiling
cProfile.run('slow_function()')
cProfile.run('fast_function()')
保存结果
profiler = cProfile.Profile()
profiler.enable()
slow_function()
profiler.disable()
stats = pstats.Stats(profiler).sort_stats('cumtime')
stats.print_stats(10)
1.3 输出解读
3 function calls in 0.089 seconds
Ordered by: cumulative time
ncalls tottime percall cumtime percall filename:lineno(function)
1 0.089 0.089 0.089 0.089 <string>:1(<module>)
1 0.089 0.089 0.089 0.089 profile.py:1(slow_function)
1 0.000 0.000 0.000 0.000 {built-in method builtins.sum}
| 列名 | 含义 |
| ncalls | 调用次数 |
| tottime | 函数自身执行时间(不含子函数) |
| cumtime | 累计时间(含子函数) |
| percall | 平均每次调用时间 |
二、高级profiling工具
2.1 line_profiler(逐行分析)
# 安装: pip install line_profiler
from line_profiler import LineProfiler
def slow_function(n):
total = 0
for i in range(n):
total += i
return total
profiler = LineProfiler()
profiler.add_function(slow_function)
profiler.runctx('slow_function(1000000)', globals(), locals())
profiler.print_stats()
输出示例:
Timer unit: 1e-06 s
Total time: 0.089123 s
File: test.py
Function: slow_function at line 1
Line # Hits Time Per Hit % Time Line Contents
==============================================================
1 def slow_function(n):
2 1 0.000 0.000 0.0 total = 0
3 1000001 89123.0 0.1 99.9 for i in range(n):
4 1000000 89120.0 0.1 99.9 total += i
5 1 0.000 0.000 0.0 return total
2.2 memory_profiler(内存分析)
# 安装: pip install memory_profiler
from memory_profiler import profile
@profile
def memory_intensive():
data = []
for i in range(1000000):
data.append([i] * 100)
return data
memory_intensive()
输出示例:
Line # Mem usage Increment Occurrences Line Contents
============================================================
1 45.0 MiB 45.0 MiB 1 @profile
2 def memory_intensive():
3 45.0 MiB 0.0 MiB 1 data = []
4 845.0 MiB 800.0 MiB 1000000 for i in range(1000000):
5 845.0 MiB 0.0 MiB 1000000 data.append([i] * 100)
6 845.0 MiB 0.0 MiB 1 return data
2.3 py-spy(生产环境profiling)
# 安装: pip install py-spy
对运行中的程序profiling
py-spy top --pid 12345
生成火焰图
py-spy record -o profile.svg --pid 12345
采样模式(低开销)
py-spy record -o profile.svg --pid 12345 --rate 100
三、常见性能瓶颈
3.1 循环优化
# ❌ 低效:多次查找
def find_duplicates(items):
duplicates = []
for item in items:
if items.count(item) > 1 and item not in duplicates:
duplicates.append(item)
return duplicates
✅ 高效:使用集合
def find_duplicates(items):
seen = set()
duplicates = set()
for item in items:
if item in seen:
duplicates.add(item)
seen.add(item)
return list(duplicates)
性能对比
import timeit
items = list(range(1000)) * 10
timeit.timeit(lambda: find_duplicates(items), number=100)
低效: ~5秒 高效: ~0.01秒
3.2 字符串拼接
# ❌ 低效:字符串不可变,每次拼接都创建新对象
def build_string(n):
result = ""
for i in range(n):
result += str(i)
return result
✅ 高效:使用列表收集,最后join
def build_string(n):
parts = []
for i in range(n):
parts.append(str(i))
return "".join(parts)
或者用生成器表达式
def build_string(n):
return "".join(str(i) for i in range(n))
3.3 列表推导式vs循环
# ✅ 列表推导式通常更快
squares = [x**2 for x in range(1000000)]
❌ 显式循环
squares = []
for x in range(1000000):
squares.append(x**2)
性能对比
import timeit
timeit.timeit(lambda: [x**2 for x in range(1000000)], number=10)
timeit.timeit(lambda: (squares := [], [squares.append(x**2) for x in range(1000000)][1], squares)[2], number=10)
3.4 函数调用开销
# ❌ 频繁函数调用
def process(items):
results = []
for item in items:
results.append(transform(item))
return results
def transform(x):
return x * 2
✅ 内联或批量处理
def process(items):
return [x * 2 for x in items]
或者用map
def process(items):
return list(map(lambda x: x * 2, items))
四、内存优化
4.1 使用生成器
# ❌ 占用大量内存
def read_all_lines(filename):
with open(filename) as f:
return f.readlines() # 一次性加载所有行
✅ 逐行处理
def read_lines(filename):
with open(filename) as f:
for line in f:
yield line.strip()
处理大文件
for line in read_lines('large_file.txt'):
process(line)
4.2 使用__slots__
# ❌ 每个实例都有__dict__,占用额外内存
class Point:
def __init__(self, x, y):
self.x = x
self.y = y
✅ 使用__slots__减少内存
class Point:
__slots__ = ['x', 'y']
def __init__(self, x, y):
self.x = x
self.y = y
内存对比
import sys
p1 = Point(1, 2)
print(sys.getsizeof(p1)) # 使用__slots__后更小
4.3 避免不必要的对象创建
# ❌ 在循环中创建对象
def process(items):
results = []
for item in items:
config = {'mode': 'fast', 'cache': True}
results.append(transform(item, config))
return results
✅ 对象复用
def process(items):
config = {'mode': 'fast', 'cache': True}
results = []
for item in items:
results.append(transform(item, config))
return results
五、CPU优化
5.1 使用内置函数
# ❌ 手动实现
def sum_list(items):
total = 0
for item in items:
total += item
return total
✅ 使用内置sum
def sum_list(items):
return sum(items)
性能提升: 2-5倍
5.2 使用numpy进行数值计算
import numpy as np
❌ Python原生
def compute_squares(items):
return [x**2 for x in items]
✅ numpy向量化
def compute_squares(items):
arr = np.array(items)
return (arr ** 2).tolist()
性能对比(100万条数据)
Python: ~0.1秒 numpy: ~0.001秒(100倍提升)
5.3 使用numba加速
# 安装: pip install numba
from numba import jit
@jit(nopython=True)
def compute_squares_numba(items):
result = np.empty(len(items))
for i in range(len(items)):
result[i] = items[i] ** 2
return result
性能对比(100万条数据)
Python: ~0.1秒 numba: ~0.0005秒(200倍提升)
六、实战案例
6.1 数据分析优化
import pandas as pd
import numpy as np
❌ 低效:逐行处理
def process_dataframe(df):
results = []
for idx, row in df.iterrows():
if row['value'] > 100:
results.append(row['value'] * 2)
return results
✅ 高效:向量化操作
def process_dataframe(df):
mask = df['value'] > 100
return (df.loc[mask, 'value'] * 2).tolist()
或者用numpy
def process_dataframe(df):
values = df['value'].values
return (values[values > 100] * 2).tolist()
6.2 数据库查询优化
# ❌ N+1查询问题
def get_users_with_orders():
users = User.query.all()
result = []
for user in users:
orders = Order.query.filter_by(user_id=user.id).all()
result.append({'user': user, 'orders': orders})
return result
✅ 使用joinedload
from sqlalchemy.orm import joinedload
def get_users_with_orders():
users = User.query.options(joinedload('orders')).all()
return [{'user': u, 'orders': u.orders} for u in users]
查询次数: N+1 → 1
6.3 并发优化
import asyncio
import aiohttp
❌ 串行请求
async def fetch_all(urls):
results = []
async with aiohttp.ClientSession() as session:
for url in urls:
async with session.get(url) as response:
results.append(await response.text())
return results
✅ 并发请求
async def fetch_all(urls):
async with aiohttp.ClientSession() as session:
tasks = [session.get(url) for url in urls]
responses = await asyncio.gather(*tasks)
return [await r.text() for r in responses]
性能提升: 10个请求,串行10秒 → 并发1秒
七、优化检查清单
| 检查项 | 优化方法 | 预期提升 |
| 循环内查找 | 使用set/dict | 10-100倍 |
| 字符串拼接 | 使用join | 10-100倍 |
| 列表创建 | 列表推导式 | 2-5倍 |
| 数值计算 | numpy/numba | 10-100倍 |
| 大文件读取 | 生成器逐行 | 内存减少99% |
| 对象创建 | 对象复用 | 减少GC压力 |
| 数据库查询 | 批量查询 | 10-100倍 |
| I/O等待 | 异步并发 | 10-100倍 |
常见坑自查清单
| 坑 | 现象 | 自查方法 | 修复方案 |
| 盲目优化 | 优化后更慢 | 先profiling | 用cProfile定位 |
| 过早优化 | 代码难读 | 先写对再优化 | 遵循KISS原则 |
| 忽略算法 | 数据结构不当 | 检查算法复杂度 | 选择合适的结构 |
| 内存泄漏 | 内存持续增长 | 用memory_profiler | 检查循环引用 |
结语
关键洞察:
-
●先profiling,再优化-
●80%的时间花在20%的代码上-
●算法优化 > 代码优化-
●不要过早优化-
互动
-
1.你用哪种profiling工具最多?-
2.遇到过最棘手的性能问题是什么?-
3.你觉得numpy值得学习吗?-
版本: V1.0 | 2026-05-26 | Python性能优化系列
📚 推荐阅读
📝 摘要:今天深入学习静态代码分析技术,这是安全审计的核心技能。从 Python AST 模块到检测模式设计,收获满满!
发布于 202603
01-Python 环境搭建与第一个脚本
发布于 202603
【优化】Python代码优化与调试技巧
发布于 202603
KEYWORDS
IL, Python, python, 函数, 循环
💡 如果你觉得这篇文章有帮助,请点个在看,分享给更多需要的人!
📝 关注我,获取更多实用干货~
🤝 有问题欢迎评论区留言交流!