第8章 8.2 性能优化与压测：让你的代码跑得更快

上一章我们聊了缓存——把常用数据提前准备好，省得反复去数据库捞。缓存是「空间换时间」的经典思路。但光有缓存还不够，你有没有遇到过这种情况：代码逻辑都对，但访问一多就卡成 PPT？或者本地跑得飞起，放到服务器上慢得像爬行？

这就不是缓存能解决的了，这一章我们来聊聊性能优化和压测——让你的代码不仅逻辑正确，还能扛住真实流量。

🎯 开场 3 分钟：为什么你的代码跑不快？

场景切入：小明的外卖店

小明开了个外卖店，日均 200 单时厨房忙得过来。后来上了平台推荐，单量涨到 2000 单，结果出餐速度跟不上了——厨师还是那几个人，锅还是那口锅。

代码也一样：

• 数据量小的时候，for 循环跑一圈没啥感觉
• 数据量大了，10 万条数据一个个处理，浏览器转圈圈
• 并发一高，服务器开始排队响应，用户开始骂娘

你的痛点是不是也这样？

• 本地测试没问题，一上线就爆炸
• 不知道哪里慢，只会「感觉卡」
• 想优化但不知道从哪下手

学完这章，你能：

1\n\n\n\n\n\n. 用工具定位代码里谁在拖后腿
2. 用技巧让循环、查询变快
3. 用压测提前知道能扛多少流量

🧱 基础 25 分钟：性能优化核心概念

8.2.1 为什么需要性能优化？

生活类比：自来水 vs. 水桶接水

你家的自来水（服务器）每秒能出 10 升水。如果你拿勺子（低效代码）一勺一勺接，一分钟也接不了多少。但如果换成桶接（优化后的代码），一下能接一大桶，时间省一半。

性能优化的本质就是：减少不必要的步骤，让资源用得更合理。

8.2.2 时间复杂度：怎么衡量快慢？

代码「慢不慢」，有个硬指标叫时间复杂度，说白了就是——数据量涨的时候，代码执行次数涨得多快。

生活类比：排队买奶茶

• O(1)：直接报手机号取餐，不管队伍多长，你的时间不变
• O(n)：逐个核对姓名，100 个人排你就得等 100 次核对
• O(n²)：两两对比确认订单，100 个人要比对 10000 次

Python 里怎么判断？看你的循环嵌套了几层：

# O(1) - 常数时间，不管数据多少，只执行一次
def get_first(users):
return users[0]

# O(n) - 线性时间，数据有多少，就循环多少次
def find_admin(users):
for user in users:  # 一层循环
    if user['role'] == 'admin':
        return user

# O(n²) - 平方时间，嵌套循环
def find_pairs(users):
for a in users:           # 第一层
    for b in users:       # 第二层
        if a['id'] != b['id']:
            print(a['name'], b['name'])

8.2.3 列表推导式：Pythonic 的快写法

Python 里有个写循环的偷懒技巧，叫列表推导式。不光是写得短，执行效率也比普通 for 循环高。

生活类比：快餐流水线

普通循环是「点餐→做餐→装盘→递给你」，一条条来。
列表推导式是「流水线机器」，你报个菜单，它自动批量生产。

# 普通写法 - 3 行
numbers = [1, 2, 3, 4, 5]
squares = []
for n in numbers:
squares.append(n * n)

# 列表推导式 - 1 行搞定
squares = [n * n for n in numbers]

print(squares)  # [1, 4, 9, 16, 25]

同样算平方，列表推导式快 30% 左右。数据量大的时候，这省出来的就很可观了。

8.2.4 生成器：省内存的「自来水厂」

列表推导式虽快，但有个问题——它会先把所有结果存进内存。100 万个数就要占一大块内存。

生成器不一样，它像自来水厂：你开水龙头，水就流出来，不用提前在家里存一缸水。

# 列表推导式 - 一次性生成全部（占内存）
squares_list = [x * x for x in range(1000000)]
print(type(squares_list))  # <class 'list'>，内存占用大

# 生成器表达式 - 按需生成（省内存）
squares_gen = (x * x for x in range(1000000))
print(type(squares_gen))   # <class 'generator'>，内存占用小

# 用 next() 一个一个取，像开水龙头
print(next(squares_gen))   # 0
print(next(squares_gen))   # 1
print(next(squares_gen))   # 4

什么时候用？

• 数据量小、要用多次 → 列表推导式
• 数据量大、只遍历一次 → 生成器表达式

8.2.5 压测工具：给代码做「体检」

知道哪里慢了，下一步就是压测——模拟很多人同时访问，看代码扛不扛得住。

Python 生态里有个轻量级的压测工具叫 locust，用 Python 写压测脚本，特别适合 Web API 压测。

先安装：

pip install locust

一个最简单的压测脚本：

from locust import HttpUser, task, between

class WebsiteUser(HttpUser):
# 每隔 1-3 秒访问一次
wait_time = between(1, 3)

@task
def index_page(self):
    self.client.get("/")

保存为 locustfile.py，命令行运行：

locust -f locustfile.py --host=http://localhost:5000

然后打开浏览器访问 http://localhost:8089，你会看到一个可视化界面，可以设置「并发用户数」，点 Start 就能看到压测结果。

8.2.6 profiling：找出慢代码的「监控摄像头」

压测是看整体，profiling 是看局部——哪个函数调用最多次、哪个函数最耗时。

Python 自带的 cProfile 不用装，直接用：

import cProfile
import pstats
from io import StringIO

def slow_function():
total = 0
for i in range(10000):
    total += i
return total

# 把分析结果存到字符串里
pr = cProfile.Profile()
pr.enable()

slow_function()
slow_function()

pr.disable()
s = StringIO()
ps = pstats.Stats(pr, stream=s).sort_stats('cumulative')
ps.print_stats(10)  # 只看前 10 行
print(s.getvalue())

输出会告诉你每个函数被调用了多少次、花了多少时间。新手重点关注 ncalls（调用次数）最多的和 cumtime（累计时间）最长的。

🔥 实战 35 分钟：3 个递进小项目

项目 1（5 分钟）：对比循环 vs 列表推导式速度

目标：亲手跑一遍，感受到差距

import time

# 数据准备：1 万个数字
data = list(range(10000))

# 方式一：普通 for 循环
start = time.time()
result1 = []
for n in data:
result1.append(n * 2)
time1 = time.time() - start

# 方式二：列表推导式
start = time.time()
result2 = [n * 2 for n in data]
time2 = time.time() - start

print(f"普通循环耗时：{time1:.4f} 秒")
print(f"列表推导式耗时：{time2:.4f} 秒")
print(f"列表推导式快了 {time1/time2:.2f} 倍")

预期输出：

普通循环耗时：0.0012 秒
列表推导式耗时：0.0008 秒
列表推导式快了 1.50 倍

解释：同样的逻辑，列表推导式少了一次 append 方法调用，攒出来的优势。

项目 2（15 分钟）：分析 CSV 数据里的慢查询

目标：从真实数据文件里找出「耗时最长」的记录

假设有个 orders.csv，内容如下：

order_id,user_id,amount,created_at
1001,201,89.5,2024-01-15 10:30
1002,202,120.0,2024-01-15 10:31
1003,201,45.0,2024-01-15 10:35
...

现在要找出「单笔金额最大的 5 笔订单」：

import csv
from collections import defaultdict

# 读取 CSV
orders = []
with open('orders.csv', 'r') as f:
reader = csv.DictReader(f)
for row in reader:
    orders.append({
        'order_id': int(row['order_id']),
        'user_id': int(row['user_id']),
        'amount': float(row['amount']),
        'created_at': row['created_at']
    })

# 找出金额最大的 5 笔
# 方式一：普通排序（会遍历全部数据）
top_orders = sorted(orders, key=lambda x: x['amount'], reverse=True)[:5]

print("金额最大的 5 笔订单：")
for order in top_orders:
print(f"  订单号：{order['order_id']}，金额：{order['amount']}，用户：{order['user_id']}")

# 方式二：用堆（heapq），数据量大时更快
import heapq

# 如果数据量是百万级，用 heap 找 top N 比排序快
top5_heap = heapq.nlargest(5, orders, key=lambda x: x['amount'])
print("\n用堆找的 top 5：")
for order in top5_heap:
print(f"  订单号：{order['order_id']}，金额：{order['amount']}")

预期输出：

金额最大的 5 笔订单：
订单号：1045，金额：2999.0，用户：215
订单号：1088，金额：2580.5，用户：203
订单号：1023，金额：2200.0，用户：198
订单号：1067，金额：1899.0，用户：221
订单号：1012，金额：1560.0，用户：205

用堆找的 top 5：
订单号：1045，金额：2999.0，用户：215
...

解释：排序是 O(n log n)，找 top N 用堆是 O(n log k)，数据大的时候差距明显。

项目 3（15 分钟）：做个「网站压测报告生成器」

目标：组合前两个项目的技能，写一个压测结果分析小工具

import csv
import time
from datetime import datetime

# 模拟压测结果数据（实际可以用 locust 的 JSON 导出）
sample_results = [
{'timestamp': '2024-01-15 10:30:00', '并发数': 10, 'QPS': 520, '平均响应': 45, '最大响应': 120},
{'timestamp': '2024-01-15 10:31:00', '并发数': 20, 'QPS': 980, '平均响应': 58, '最大响应': 180},
{'timestamp': '2024-01-15 10:32:00', '并发数': 50, 'QPS': 2100, '平均响应': 85, '最大响应': 350},
{'timestamp': '2024-01-15 10:33:00', '并发数': 100, 'QPS': 3800, '平均响应': 150, '最大响应': 800},
{'timestamp': '2024-01-15 10:34:00', '并发数': 200, 'QPS': 5200, '平均响应': 280, '最大响应': 1500},
]

# 保存到 CSV
with open('stress_test_report.csv', 'w', newline='') as f:
writer = csv.DictWriter(f, fieldnames=['timestamp', '并发数', 'QPS', '平均响应', '最大响应'])
writer.writeheader()
writer.writerows(sample_results)

# 读取并分析
print("=" * 50)
print("压测报告分析")
print("=" * 50)

with open('stress_test_report.csv', 'r') as f:
reader = csv.DictReader(f)
rows = list(reader)

# 找出 QPS 最高的节点
best_qps = max(rows, key=lambda x: int(x['QPS']))
print(f"\n最佳性能点：并发 {best_qps['并发数']}，QPS {best_qps['QPS']}")

# 找出响应时间超过 200ms 的记录
slow_requests = [r for r in rows if int(r['平均响应']) > 200]
print(f"\n响应时间超标次数：{len(slow_requests)}")
for r in slow_requests:
print(f"  并发 {r['并发数']} 时，平均响应 {r['平均响应']}ms")

# 生成简单结论
print("\n结论：")
print(f"  - 当前服务器在并发 {best_qps['并发数']} 时性能最佳")
print(f"  - 并发超过 100 时，响应时间开始明显上升")
print(f"  - 建议：设置并发上限 {best_qps['并发数']}，超过后排队或拒绝")

预期输出：

==================================================
压测报告分析
==================================================

最佳性能点：并发 200，QPS 5200

响应时间超标次数：2
并发 100 时，平均响应 150ms
并发 200 时，平均响应 280ms

结论：
- 当前服务器在并发 200 时性能最佳
- 并发超过 100 时，响应时间开始明显上升
- 建议：设置并发上限 200，超过后排队或拒绝

解释：这个工具模拟了真实压测场景，能帮你快速看出性能拐点在哪里。

💪 进阶 20 分钟：常见坑 + 性能小贴士

坑 1：字符串拼接用 + 还是 join()？

# ❌ 错误：用 + 拼接字符串，每次都创建新字符串对象
result = ""
for i in range(10000):
result += str(i)

# ✅ 正确：用 join()，只创建一次
parts = [str(i) for i in range(10000)]
result = "".join(parts)

字符串是不可变对象，+ 每次都生成新字符串，1 万次拼接就要创建 1 万个中间对象。join() 只需要分配一次内存。

坑 2：查询数据库用 for 循环查 N 次

# ❌ 错误：100 个 ID，循环查 100 次数据库
user_ids = [101, 102, 103, ..., 200]
users = []
for uid in user_ids:
users.append(db.query("SELECT * FROM users WHERE id = ?", uid))

# ✅ 正确：一次查询，带上所有 ID
ids_str = ",".join(str(uid) for uid in user_ids)
users = db.query(f"SELECT * FROM users WHERE id IN ({ids_str})")

数据库查询有网络开销，N 次查询是 N 倍延迟。

坑 3：读写文件不用 with 语句

# ❌ 错误：忘记关文件，资源泄露
f = open('data.txt', 'r')
content = f.read()
# 如果后面代码报错，f.close() 永远不会执行

# ✅ 正确：用 with 自动关
with open('data.txt', 'r') as f:
content = f.read()

坑 4：在循环里查长度

# ❌ 错误：每次循环都调用 len()
for i in range(len(items)):
if i < len(items):  # 重复调用 len
    print(items[i])

# ✅ 正确：提前算好
n = len(items)
for i in range(n):
if i < n:
    print(items[i])

坑 5：深拷贝 vs 浅拷贝

# ❌ 错误：修改了原数据
original = [{'name': 'Alice'}, {'name': 'Bob'}]
copied = original
copied[0]['name'] = 'Charlie'
print(original[0]['name'])  # Charlie，原数据被改了！

# ✅ 正确：浅拷贝（拷贝第一层）
import copy
original = [{'name': 'Alice'}, {'name': 'Bob'}]
copied = copy.copy(original)  # 外层是新对象
copied[0]['name'] = 'Charlie'
print(original[0]['name'])  # Alice，没被改

# ✅ 正确：深拷贝（递归拷贝所有层）
copied = copy.deepcopy(original)

性能小贴士：善用 collections 里的高效数据结构

from collections import defaultdict, Counter

# 用 defaultdict 避免 KeyError
user_orders = defaultdict(list)
user_orders['Alice'].append(1001)
user_orders['Alice'].append(1002)

# 用 Counter 快速统计频次
votes = ['Alice', 'Bob', 'Alice', 'Charlie', 'Bob', 'Alice']
counter = Counter(votes)
print(counter.most_common(2))  # [('Alice', 3), ('Bob', 2)]

调试技巧：用 logging 代替 print

import logging

# 配置日志级别和格式
logging.basicConfig(
level=logging.DEBUG,
format='%(asctime)s - %(levelname)s - %(message)s'
)

logging.debug("这是调试信息")
logging.info("这是普通信息")
logging.warning("这是警告信息")

# 上线后改成 WARNING 级别，就不会打印 debug 了

✏️ 练习题 + 作业题

练习题（10 分钟）

练习 1（2 分钟）：抄改列表推导式
- 输入：用 [1, 2, 3, 4, 5] 生成每个数的立方
- 预期输出：[1, 8, 27, 64, 125]
- 提示：把 n * n 改成 n ** 3

练习 2（2 分钟）：加个 if 判断
- 输入：在练习 1 基础上，只保留大于 10 的立方
- 预期输出：[27, 64, 125]
- 提示：列表推导式后面可以加 if 条件

练习 3（2 分钟）：用 heapq 找 top 3
- 输入：[10, 5, 8, 20, 3]，用 heapq.nlargest 找最大的 3 个
- 预期输出：[20, 10, 8]
- 提示：import heapq 后直接用

练习 4（2 分钟）：分析压测 CSV
- 输入：给一个 test.csv，内容是 {"并发": 50, "QPS": 1000} 等行
- 预期输出：打印最大 QPS 那一行
- 提示：参考项目 2 的 max(..., key=...) 用法

练习 5（2 分钟）：找茬题
- 输入：下面代码哪里慢了？

text = ""
for i in range(1000):
text = text + str(i)

• 预期输出：指出问题 + 给出优化版本
• 提示：用 join() 代替 +

作业题（30 分钟 - 2 小时）

作业：做一个「API 性能压测小工具」

• 需求描述：写一个脚本，对指定的 URL 发起并发请求，统计响应时间分布
• 功能点：
  1. 支持指定并发数、请求次数
  2. 记录每次请求的响应时间
  3. 输出统计报告：平均响应、最小、最大、95 分位
  4. 超过 1 秒的请求标记为「超时」
• 加分项：
  1. 生成一个简单的文本柱状图
  2. 支持把结果保存到 CSV
• 验收标准：能跑起来 + 输出统计报告 + 代码有注释
• 提交方式：评论区贴代码或 GitHub 链接

📚 总结 + 资源

这一章我们学了：

1. 时间复杂度：用 O(1)、O(n)、O(n²) 衡量代码快慢
2. 列表推导式 + 生成器：Pythonic 的高效写法
3. 压测 + profiling：用工具定位性能瓶颈

延伸学习资源：

• Python 官方性能优化文档 —— 性能测试内置模块
• 《Python 高性能编程》—— 进阶必读，讲了很多底层优化
• Locust 官方文档 —— 强大的 Python 压测工具

互动钩子：

你的项目有没有遇到过「本地飞起、上线卡死」的情况？当时是怎么解决的？评论区聊聊，老粉优先回复！

下一章我们要聊点不一样的——安全。代码跑得快是好事，但如果被人钻了空子，轻则数据泄露，重则服务器被黑。下一章我们会讲 XSS、SQL 注入、CSRF 这些常见攻击原理，以及怎么用我们这一章学的性能思维，构建更安全的应用。