调试与性能优化
# 第 9 章 调试与性能优化
# 目录介绍
# 9.1 调试与日志
# 9.1.1 print 与 pdb
💬 典型场景:你的爬虫跑了 5000 条数据后突然报
KeyError——你只知道出错了,不知道哪个 URL、哪条数据触发的。你在代码里塞了 20 个print(),跑了一次又一次——每次花 10 分钟等待复现。
print() 调试法在小规模时够用——但有三大致命缺陷:
| 缺陷 | print() | 专业调试器(pdb) |
|---|---|---|
| 查看变量 | 必须在代码里提前写 print(x) | 随时看任何变量——不需要改代码 |
| 暂停执行 | ❌ 不可能——一路到底 | ✅ 断点停下,单步走 |
| 调用栈 | ❌ 完全看不见 | ✅ bt 命令看完整调用链 |
| 改后重新跑 | 每次加 print 都要重跑 | 停下来就能看——不需要重跑 |
🔑 print() 不是没用——快速定位大概位置时它仍然是最快的。但一旦进入"这个变量到底在什么时候变成了 None"的阶段,就该上 pdb 了。
# 9.1.2 pdb 调试器
pdb 是 Python 自带的调试器——零安装成本,任何环境都能用:
# ===== 方式一:在代码里插入断点 =====
def process_order(order_id):
order = fetch_order(order_id)
import pdb; pdb.set_trace() # ← 程序到这里会暂停,进入交互模式
total = order.amount * order.quantity
return total
# ===== 方式二:从命令行启动 =====
# python -m pdb my_script.py
# ===== 方式三:Python 3.7+ 内置 breakpoint() =====
def process_order(order_id):
order = fetch_order(order_id)
breakpoint() # ← 更简洁——不需要 import
total = order.amount * order.quantity
return total
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
pdb 核心命令速查:
| 命令 | 简写 | 作用 |
|---|---|---|
list / ll | l / ll | 显示当前位置代码 / 显示整个函数 |
next | n | 执行下一行(不进入函数) |
step | s | 进入函数内部 |
continue | c | 继续执行直到下一个断点 |
return | r | 执行到当前函数返回 |
print(var) | p var | 打印变量的值 |
pp vars(obj) | pp | 漂亮打印对象所有属性 |
where / bt | w | 打印完整调用栈 |
up / down | u / d | 在调用栈中上移/下移一层 |
args | a | 打印当前函数的参数 |
break 42 | b 42 | 在第 42 行加断点 |
break func | b func | 在函数入口加断点 |
condition 1 x > 100 | 条件断点——x > 100 才停下 | |
quit | q | 退出调试器 |
实战对话:
# buggy_script.py
def calculate_average(numbers):
total = 0
for n in numbers:
total += n
breakpoint()
return total / len(numbers)
def main():
data = get_data_from_api() # 返回 [](空列表!)
avg = calculate_average(data)
print(f"平均:{avg}")
main()
2
3
4
5
6
7
8
9
10
11
12
13
14
$ python buggy_script.py
> calculate_average() line 5, in calculate_average
-> return total / len(numbers)
(Pdb) p total # 查看 total
0
(Pdb) p numbers # 查看 numbers
[] # ← 空的!这就是 bug 的根源
(Pdb) bt # 看调用栈
calculate_average()
main()
<module>
(Pdb) u # 上到 main() 帧
> main() line 9 -> avg = calculate_average(data)
(Pdb) p data # data 是什么?
[] # ← get_data_from_api 返回了空列表
(Pdb) q # 退出——修复 get_data_from_api()
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
🔑 条件断点——最强大的调试技巧之一:
# 循环 10000 次,只想知道第 9999 次发生了什么——
# 手动 continue 9998 次是不可能的
for i, item in enumerate(items):
breakpoint() # ❌ 每次循环都停——崩溃
process(item)
# ✅ 条件断点——只在满足条件时停下
for i, item in enumerate(items):
if i == 9999:
breakpoint() # ✅ 只在第 9999 次停下
process(item)
# 或更优雅:pdb 命令行中
# (Pdb) break 10, i == 9999 ← 在代码第 10 行加条件断点
2
3
4
5
6
7
8
9
10
11
12
13
14
# 9.1.3 VS Code 调试
命令行 pdb 已经很强——但 VS Code 的图形化调试更直观:
配置 launch.json:
{
"version": "0.2.0",
"configurations": [
{
"name": "Python:当前文件",
"type": "debugpy",
"request": "launch",
"program": "${file}",
"console": "integratedTerminal",
"justMyCode": false,
"env": {
"PYTHONPATH": "${workspaceFolder}"
}
},
{
"name": "Python:带参数",
"type": "debugpy",
"request": "launch",
"program": "${file}",
"args": ["--input", "data.csv", "--verbose"],
"console": "integratedTerminal"
}
]
}
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
🔑 VS Code 调试核心操作:
| 操作 | 快捷键 (macOS) | 作用 |
|---|---|---|
| 加断点 | F9 | 点击行号左侧 |
| 开始调试 | F5 | 运行到第一个断点 |
| 单步跳过 | F10 | 下一行(不进入函数) |
| 单步进入 | F11 | 进入函数内部 |
| 单步跳出 | Shift+F11 | 跳出当前函数 |
| 继续 | F5 | 运行到下一个断点 |
| 查看变量 | 左侧 VARIABLES 面板 | 所有变量一目了然 |
| 监视表达式 | 左侧 WATCH 面板 | 自定义表达式如 len(items) > 100 |
# 9.1.4 logging 组件
print() 在代码里"不干净"、不能控制级别、不能写文件——logging 是生产环境唯一的日志方案:
import logging
# ===== 最速配置(一行能用) =====
logging.basicConfig(
level=logging.INFO, # 最低显示级别
format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
datefmt="%Y-%m-%d %H:%M:%S",
handlers=[
logging.FileHandler("app.log", encoding="utf-8"), # 写文件
logging.StreamHandler(), # 同时打控制台
],
)
# ===== 使用 =====
logger = logging.getLogger(__name__) # 每个模块一个 logger——名字自动是模块名
logger.debug("详细的调试信息——通常不显示")
logger.info("用户 %s 登录成功", "张三") # %s 风格——logging 社区惯例(不是 f-string)
logger.warning("磁盘使用率 %d%%——接近上限", 85)
logger.error("数据库连接失败", exc_info=True) # exc_info=True 自动附带堆栈
logger.critical("系统崩溃——立即处理!")
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
🔑 为什么 logging 里用 %s 而不是 f-string?
# f-string 在 log 调用前求值——即使级别不够也会执行
logger.debug(f"计算结果:{expensive_calculation()}") # ❌ 计算了但不输出——浪费
# %s 延迟求值——如果 DEBUG < INFO,expensive() 不会被调用
logger.debug("计算结果:%s", expensive_calculation()) # ✅ 不输出就不计算
# Python 3.6+ 也可以惰性日志(但不如惰性求值彻底)
logger.debug("计算结果:%s", expensive_calculation())
2
3
4
5
6
7
8
logging 四大组件:
┌─────────────┐ ┌───────────────┐ ┌──────────────┐ ┌──────────────┐
│ Logger │ → │ Handler │ → │ Formatter │ → │ Filter │
│ 记录器 │ │ 处理器 │ │ 格式化器 │ │ 过滤器 │
│ │ │ │ │ │ │ │
│ 产生日志 │ │ 输出到哪里 │ │ 长什么样 │ │ 哪些要/不要 │
└─────────────┘ └───────────────┘ └──────────────┘ └──────────────┘
↑ ↑ ↑
logger.info() FileHandler → app.log 只记录 ERROR 级别
logger.error() StreamHandler → 控制台
2
3
4
5
6
7
8
9
# ===== 生产级 logging 配置(完整四组件) =====
import logging
from logging.handlers import RotatingFileHandler
def setup_logging():
# 1. Logger
logger = logging.getLogger("my_app")
logger.setLevel(logging.DEBUG) # Logger 级别——总闸
# 2. Formatter
file_fmt = logging.Formatter(
"%(asctime)s | %(levelname)-8s | %(name)s:%(lineno)d | %(message)s"
)
console_fmt = logging.Formatter(
"%(levelname)-8s | %(message)s"
)
# 3. Handler(文件——带轮转,最大 10MB,保留 5 个备份)
fh = RotatingFileHandler("app.log", maxBytes=10*1024*1024, backupCount=5,
encoding="utf-8")
fh.setLevel(logging.INFO) # 文件只记 INFO 以上
fh.setFormatter(file_fmt)
# 4. Handler(控制台)
ch = logging.StreamHandler()
ch.setLevel(logging.DEBUG)
ch.setFormatter(console_fmt)
# 5. 组装
logger.addHandler(fh)
logger.addHandler(ch)
return logger
logger = setup_logging()
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
# 9.1.5 日志轮转
# ===== RotatingFileHandler:按大小轮转 =====
from logging.handlers import RotatingFileHandler
fh = RotatingFileHandler("app.log", maxBytes=10*1024*1024, backupCount=5)
# 文件到 10MB → 自动滚动:app.log → app.log.1 → app.log.2 → ... → 最多 5 个
# ===== TimedRotatingFileHandler:按时间轮转 =====
from logging.handlers import TimedRotatingFileHandler
fh = TimedRotatingFileHandler("app.log", when="midnight", backupCount=30)
# 每天午夜自动滚动——保留最近 30 天
2
3
4
5
6
7
8
9
结构化日志——JSON 格式(方便接入 ELK/Splunk 等日志系统):
import json
import logging
from datetime import datetime
class JsonFormatter(logging.Formatter):
"""输出 JSON 格式日志——方便日志平台解析"""
def format(self, record):
log_entry = {
"timestamp": datetime.utcnow().isoformat(),
"level": record.levelname,
"logger": record.name,
"message": record.getMessage(),
"module": record.module,
"line": record.lineno,
}
if record.exc_info and record.exc_info[1]:
log_entry["exception"] = str(record.exc_info[1])
return json.dumps(log_entry, ensure_ascii=False)
handler = logging.FileHandler("app.jsonl")
handler.setFormatter(JsonFormatter())
logger = logging.getLogger("structured")
logger.addHandler(handler)
logger.info("用户登录", extra={"user_id": 12345}) # 这里需要用结构化方式
# 输出到 app.jsonl:{"timestamp":"...", "level":"INFO", "message":"用户登录", ...}
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
# 9.1.6 loguru 日志
loguru 把 logging 的四大组件全部封装到一行——新人最优选择:
from loguru import logger
# 一行配置——同时输出到控制台 + 文件(自动轮转 + 彩色)
logger.add("app_{time}.log", rotation="10 MB", retention="7 days",
level="INFO", encoding="utf-8")
# 使用——和 logging 一样简单
logger.debug("调试信息")
logger.info("处理文件:{}", filename)
logger.warning("磁盘使用率 {}%——接近上限", 85)
logger.error("数据库连接失败")
# 自动捕获异常栈——不需要 exc_info=True
try:
1 / 0
except Exception:
logger.exception("计算错误") # ← 自动附带完整堆栈
# 装饰器——自动记录函数调用
@logger.catch
def risky_operation():
raise ValueError("出错了") # ← 异常自动记录到日志
# 绑定上下文——日志自动带上用户/请求信息
context_logger = logger.bind(user_id=123, request_id="abc-456")
context_logger.info("请求处理开始") # 自动带 user_id 和 request_id
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
🔑 loguru vs logging 选择:
| 特性 | logging | loguru |
|---|---|---|
| 配置行数 | ~20 行 | 1 行 |
| 彩色输出 | 需额外配置 | 默认彩色 |
| 异常捕获 | exc_info=True | logger.exception() 自动 |
| 日志轮转 | 需选 Handler 类型 | rotation="10 MB" 搞定 |
| 结构化日志 | serialize=True | json.dumps() |
| 性能 | 稍好(C 加速) | 够用 |
| 适用场景 | 框架/库(避免引入依赖) | 应用/脚本(开发体验优先) |
# 9.2 性能优化
性能优化的第一原则——先测后改。不测就改 = 瞎改。Python 提供了四层性能分析工具。
# 9.2.1 timeit 计时
import timeit
# 比较两种实现——哪个更快?
setup_code = """
data = list(range(10000))
"""
# 方式 1:列表推导式
t1 = timeit.timeit("[x*2 for x in data]", setup=setup_code, number=1000)
# 方式 2:map + lambda
t2 = timeit.timeit("list(map(lambda x: x*2, data))", setup=setup_code, number=1000)
print(f"列表推导式:{t1:.4f}s")
print(f"map+lambda:{t2:.4f}s")
print(f"列表推导式 快 {t2/t1:.1f}x")
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
# 命令行快速对比
# python -m timeit -s "data=list(range(10000))" "[x*2 for x in data]"
# python -m timeit -s "data=list(range(10000))" "list(map(lambda x:x*2, data))"
# %%timeit(Jupyter Notebook)
# In [1]: %%timeit
# ...: data = list(range(10000))
# ...: [x*2 for x in data]
# Out[1]: 257 µs ± 3.2 µs per loop
2
3
4
5
6
7
8
9
# 9.2.2 cProfile
cProfile 告诉你每个函数花了多少时间——相当于程序的"体检报告":
import cProfile
import pstats
from io import StringIO
def slow_sorting():
data = list(range(10000))
for _ in range(100):
data.sort(reverse=True)
data.sort()
def slow_computation():
total = 0
for i in range(100000):
total += i ** 2
return total
def main():
slow_sorting()
slow_computation()
# ===== 方式一:代码内启动 =====
profiler = cProfile.Profile()
profiler.enable()
main()
profiler.disable()
# 输出统计——按累计时间排序
stats = pstats.Stats(profiler, stream=StringIO()).sort_stats("cumulative")
stats.print_stats(10)
# ncalls tottime percall cumtime percall filename:lineno(function)
# 100 0.001 0.000 0.320 0.003 ... sort()
# 1 0.180 0.180 0.180 0.180 ... slow_computation()
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
# ===== 方式二:命令行启动 =====
python -m cProfile -s cumulative my_script.py
# 或输出到文件后用 pstats 交互分析
python -m cProfile -o output.prof my_script.py
python -m pstats output.prof
# output.prof% sort cumulative
# output.prof% stats 15
# output.prof% quit
2
3
4
5
6
7
8
9
🔑 cProfile 输出各列含义:
| 列 | 含义 |
|---|---|
ncalls | 调用次数 |
tottime | 本函数自身耗时(不含子函数) |
cumtime | 累计耗时(含所有子函数)← 看这个找瓶颈 |
percall | 平均每次调用耗时 |
# 9.2.3 逐行分析
cProfile 告诉你哪个函数慢——line_profiler 告诉你函数里哪一行慢:
pip install line_profiler
# 在要分析的函数上加 @profile 装饰器
@profile
def process_records(records):
cleaned = []
for r in records:
if r["score"] > 60: # ① 过滤
r["grade"] = "A" if r["score"] > 90 else "B" # ② 分级
r["processed"] = r["score"] * 1.5 + r.get("bonus", 0) # ③ 计算
cleaned.append(r)
return cleaned
# 生成模拟数据
records = [{"score": i, "bonus": i % 10} for i in range(100000)]
process_records(records)
# 运行分析
# kernprof -l -v script.py
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
# line_profiler 输出示例
Line # Hits Time Per Hit % Time Line Contents
==========================================================
1 @profile
2 def process_records(records):
3 1 2.0 2.0 0.0 cleaned = []
4 100001 13000.0 0.1 1.5 for r in records:
5 100000 45000.0 0.5 5.3 if r["score"] > 60:
6 60000 12000.0 0.2 1.4 r["grade"] = "A" if ...
7 60000 780000.0 13.0 91.8 ⚠️ r["processed"] = r["score"] * 1.5 + r.get("bonus", 0)
8 60000 3000.0 0.1 0.4 cleaned.append(r)
9 1 2.0 2.0 0.0 return cleaned
# ↑ 91.8% 的时间花在这一行!——优化目标锁定
2
3
4
5
6
7
8
9
10
11
12
13
# 9.2.4 内存分析
pip install memory_profiler
from memory_profiler import profile
@profile
def memory_hungry():
a = [0] * 10_000_000 # ↓ 内存变化
# Line 5: Mem usage 76.2 MiB Increment 76.2 MiB
b = [i * 2 for i in a]
# Line 7: Mem usage 152.5 MiB Increment 76.3 MiB ← 峰值!
del a # 手动释放
# Line 9: Mem usage 76.3 MiB Increment -76.2 MiB ← 降下来了
return sum(b)
memory_hungry()
# 命令行
# python -m memory_profiler script.py
# 按时间采样——看内存波动趋势
# mprof run script.py
# mprof plot → 生成内存时间曲线图
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
# 9.2.5 优化技巧
| # | 技巧 | 示例 | 原理 |
|---|---|---|---|
| 1 | 局部变量缓存 | upper = str.upper; [upper(s) for s in items] | 省去每次全局查找 |
| 2 | 用 join 拼接字符串 | "".join(chunks) | 避免 += 反复创建新字符串 |
| 3 | 用集合检查成员 | if x in my_set 代替 if x in my_list | O(1) vs O(n) |
| 4 | 用生成器而非列表 | sum(x*2 for x in data) 代替 sum([x*2 for x in data]) | 省去中间列表内存 |
| 5 | functools.lru_cache | @lru_cache; def fib(n): ... | 自动记忆化 |
| 6 | NumPy 向量化 | np.sum(arr * 2) 代替 sum(x*2 for x in arr) | C 层运算 |
| 7 | __slots__ 减少内存 | class Point: __slots__ = ('x','y') | 禁用 __dict__,每对象省 ~200B |
| 8 | PyPy 解释器 | pypy script.py | JIT 编译——某些场景快 5-10 倍 |
# ===== 技巧 1 详解——局部变量缓存 =====
import timeit
# 每次都全局查找 str.upper
code1 = """
items = ["hello"] * 1000
result = [str.upper(s) for s in items]
"""
# 缓存到局部变量
code2 = """
items = ["hello"] * 1000
upper = str.upper
result = [upper(s) for s in items]
"""
print(f"全局查找:{timeit.timeit(code1, number=10000):.4f}s")
print(f"局部缓存:{timeit.timeit(code2, number=10000):.4f}s")
# 典型结果:局部缓存约快 15~20%
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
# ===== 技巧 3 详解——集合 O(1) 查找 =====
import time
n = 100_000
items = list(range(n))
items_set = set(items)
target = n - 1 # 找最后一个——列表的最坏情况
# 列表查找——O(n)
start = time.perf_counter()
_ = target in items
print(f"列表查找:{time.perf_counter() - start:.6f}s")
# 集合查找——O(1)
start = time.perf_counter()
_ = target in items_set
print(f"集合查找:{time.perf_counter() - start:.6f}s")
# 列表:0.001s 集合:0.0000004s——快 2500 倍!
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
# 9.3 新手陷阱
| # | 陷阱 | 说明 |
|---|---|---|
| 1 | pdb.set_trace() 忘删就提交 | import pdb; pdb.set_trace() 留在代码里——生产环境卡住。用 breakpoint() + pre-commit 拦截 |
| 2 | logging 里用 f-string | logger.debug(f"x={x}") 即使级别不够也会求值。改用 %s |
| 3 | 性能优化前不测试 | 凭感觉改代码——改完发现更慢。永远先用 cProfile 定位热点 |
| 4 | logger = logging.getLogger(__name__) 放在模块顶层 | 子模块导入时 logging 还没配置好——用惰性初始化 |
| 5 | 生产环境开 DEBUG 日志 | 每毫秒一条 DEBUG——磁盘一天填满。生产用 INFO/WARNING |
陷阱 1 详解——pre-commit 拦截:
# .pre-commit-config.yaml 加一条
- repo: https://github.com/pre-commit/pre-commit-hooks
rev: v4.5.0
hooks:
- id: debug-statements # 检查 import pdb; set_trace(); breakpoint()
2
3
4
5
# 9.4 综合思考题
pdb vs VS Code 调试器 vs print:三者各有什么不可替代的优势?在什么场景下,
print()反而比图形化调试器更快解决问题?日志级别选择策略:DEBUG / INFO / WARNING / ERROR / CRITICAL 五级日志——在一个 Web 应用中,哪些行为应该记 INFO、哪些记 WARNING?级别选错了有什么后果(信息过载 vs 漏掉关键错误)?
cProfile 的观察者效应:cProfile 在分析时会减慢程序 10~20%——这本身就会改变程序的行为(尤其是 IO 密集型程序)。你能想到什么方法来减少这种"测不准"效应?
内存泄漏排查:一个 Python 进程的内存从启动时的 50MB 涨到 2GB——但你确认没有循环引用。除了
memory_profiler,还有哪些工具能帮你找出内存泄漏的根源?(提示:objgraph、tracemalloc、gc模块)Python 性能上限:Python 再怎么优化也快不过 C/Rust——但为什么那么多公司仍然选择 Python 做后端服务?在什么场景下,研发效率 > 运行效率,什么场景下反之?