3. 测试脚本设计：用Python写压测脚本
好，咱们进入实战环节。脚本设计这块，说白了就是把你脑子里的压测方案，翻译成机器能跑起来的代码。我个人习惯先把整体架构想清楚，再动手写。你想想看，如果连连接怎么管、事务怎么模拟都没想明白，代码写出来也是一团乱麻。
核心要点：压测脚本不是一次性玩具，它要能复用、能调参、能扛得住长时间运行。我见过太多人写出来的脚本跑10分钟就崩了，原因就是没做好资源管理。
3.1 整体架构设计
先画个图，让你对脚本的整体结构有个直观认识。这是我做压测项目时常用的分层思路：
这个架构图我用了好几年，每次做新项目都拿它当模板。从上到下，配置驱动引擎，引擎调度数据，最后输出结果。每一层各司其职，改起来也方便。
3.2 连接池管理
连接池这东西，说白了就是「别每次干活都重新开门」。我早期做压测时犯过一个错——每个请求都新建数据库连接，结果压到500并发时，数据库那边连接数直接爆了，服务挂了。
后来我学乖了，用连接池来管理。Python里我常用 DBUtils 或者 psycopg2.pool。给你看个例子：
from dbutils.pooled_db import PooledDB
import pymysql
class ConnectionPool:
"""数据库连接池管理器"""
def __init__(self, config):
self.pool = PooledDB(
creator=pymysql,
maxconnections=config['max_connections'], # 最大连接数
mincached=config['min_cached'], # 最小空闲连接
maxcached=config['max_cached'], # 最大空闲连接
blocking=True, # 无连接时是否阻塞等待
host=config['host'],
port=config['port'],
user=config['user'],
password=config['password'],
database=config['database']
)
def get_conn(self):
"""从池中获取连接"""
return self.pool.connection()
def close_all(self):
"""关闭所有连接"""
self.pool.close()我的经验：连接池大小不是越大越好。我一般按「并发数 × 每个事务需要的连接数 × 1.2」来估算。比如1000并发，每个事务用1个连接，那就设1200。留20%的余量，够用又不浪费。
3.3 事务模拟
事务模拟是压测的核心。你不能只发一个查询就完事，得模拟真实用户的操作链路。比如一个下单事务，可能包含：查库存 → 扣库存 → 生成订单 → 更新余额。
我习惯把每个事务封装成一个类，这样清晰又好维护：
import time
import random
from contextlib import contextmanager
class TransactionSimulator:
"""事务模拟器"""
def __init__(self, conn_pool):
self.pool = conn_pool
@contextmanager
def transaction(self):
"""事务上下文管理器"""
conn = self.pool.get_conn()
try:
conn.begin()
yield conn
conn.commit()
except Exception as e:
conn.rollback()
raise e
finally:
conn.close()
def place_order(self, user_id, product_id, quantity):
"""模拟下单事务"""
with self.transaction() as conn:
cursor = conn.cursor()
# 步骤1：检查库存
cursor.execute(
"SELECT stock FROM products WHERE id = %s",
(product_id,)
)
stock = cursor.fetchone()[0]
if stock < quantity:
raise Exception("库存不足")
# 步骤2：扣减库存
cursor.execute(
"UPDATE products SET stock = stock - %s WHERE id = %s",
(quantity, product_id)
)
# 步骤3：生成订单
order_id = int(time.time() * 1000) % 1000000
cursor.execute(
"INSERT INTO orders (id, user_id, product_id, quantity) "
"VALUES (%s, %s, %s, %s)",
(order_id, user_id, product_id, quantity)
)
# 模拟思考时间（用户操作间隔）
time.sleep(random.uniform(0.01, 0.05))
return order_id注意：事务里的每个步骤都可能失败。我曾经遇到过一个坑——库存扣了但订单没生成，结果数据对不上了。所以一定要用事务包裹，要么全成功，要么全回滚。
3.4 随机数据生成
压测数据不能千篇一律。你想想看，如果所有请求都用同一个用户ID，数据库那边缓存一命中，压出来的结果根本不准。
我常用的数据生成策略是这样的：
import random
import string
from datetime import datetime, timedelta
class DataGenerator:
"""随机数据生成器"""
def __init__(self, seed=42):
random.seed(seed) # 固定种子，保证可复现
def random_user_id(self):
"""从预置用户池中随机选取"""
return random.randint(1001, 9999)
def random_product_id(self):
"""随机商品ID"""
return random.choice([101, 102, 103, 104, 105])
def random_quantity(self):
"""随机购买数量（1-5）"""
return random.randint(1, 5)
def random_string(self, length=8):
"""随机字符串"""
return ''.join(
random.choices(string.ascii_letters + string.digits, k=length)
)
def random_timestamp(self, days_back=30):
"""随机时间戳（最近30天内）"""
now = datetime.now()
delta = timedelta(days=random.randint(0, days_back))
return (now - delta).strftime('%Y-%m-%d %H:%M:%S')
def batch_generate(self, count=1000):
"""批量生成测试数据"""
data = []
for _ in range(count):
data.append({
'user_id': self.random_user_id(),
'product_id': self.random_product_id(),
'quantity': self.random_quantity(),
'order_time': self.random_timestamp()
})
return data关键点：数据生成要满足「随机性」和「可复现性」两个要求。随机性保证压测场景真实，可复现性保证问题能定位。我一般用固定种子（比如42），这样每次跑出来的数据序列都一样，出问题好排查。
3.5 整合：完整的压测脚本
好了，把上面这些组件拼起来，就是一个能用的压测脚本了：
import time
import threading
from concurrent.futures import ThreadPoolExecutor
class LoadTestRunner:
"""压测执行器"""
def __init__(self, config):
self.config = config
self.pool = ConnectionPool(config['database'])
self.transaction = TransactionSimulator(self.pool)
self.data_gen = DataGenerator()
self.results = []
self.lock = threading.Lock()
def run_single(self):
"""执行单次压测事务"""
start = time.time()
success = True
try:
user_id = self.data_gen.random_user_id()
product_id = self.data_gen.random_product_id()
quantity = self.data_gen.random_quantity()
self.transaction.place_order(
user_id, product_id, quantity
)
except Exception as e:
success = False
finally:
elapsed = time.time() - start
with self.lock:
self.results.append({
'success': success,
'elapsed': elapsed,
'timestamp': time.time()
})
def run(self):
"""启动压测"""
print(f"开始压测，并发数: {self.config['concurrency']}")
print(f"持续时间: {self.config['duration']}秒")
with ThreadPoolExecutor(
max_workers=self.config['concurrency']
) as executor:
futures = []
end_time = time.time() + self.config['duration']
while time.time() < end_time:
futures.append(executor.submit(self.run_single))
# 控制请求速率
time.sleep(1.0 / self.config['qps'])
# 等待所有任务完成
for f in futures:
f.result()
self._report()
def _report(self):
"""输出压测报告"""
total = len(self.results)
success = sum(1 for r in self.results if r['success'])
fail = total - success
avg_time = sum(r['elapsed'] for r in self.results) / total
print(f"\n=== 压测报告 ===")
print(f"总请求数: {total}")
print(f"成功: {success}")
print(f"失败: {fail}")
print(f"成功率: {success/total*100:.2f}%")
print(f"平均响应时间: {avg_time*1000:.2f}ms")
print(f"实际QPS: {total/self.config['duration']:.2f}")
# 使用示例
if __name__ == '__main__':
config = {
'concurrency': 50,
'duration': 60,
'qps': 100,
'database': {
'host': 'localhost',
'port': 3306,
'user': 'test',
'password': 'test123',
'database': 'order_db',
'max_connections': 60,
'min_cached': 10,
'max_cached': 30
}
}
runner = LoadTestRunner(config)
runner.run()避坑指南：我曾经在压测时忘了加 time.sleep 来控制请求速率，结果一启动就把数据库打满了。后来我养成了习惯——压测脚本里一定要有速率控制，哪怕只是简单的sleep。不然你测的不是系统性能，而是你自己的网络带宽。
3.6 脚本设计的几个原则
最后，总结几条我这些年摸爬滚打总结出来的原则：
- 配置化：所有参数都放配置文件里，别硬编码。我见过有人把数据库密码写死在代码里，结果代码传到Git上...嗯，你懂的。
- 可观测：每个关键步骤都要打日志。压测跑着跑着出问题了，没日志你根本不知道哪里挂了。
- 优雅退出：按Ctrl+C时，要能正常关闭连接池、保存中间结果。我早期写的脚本一中断就丢数据，后来加了信号处理才搞定。
- 幂等性：同一个请求跑两次，结果应该一样。不然你压测过程中重试几次，数据就乱套了。
说白了，写压测脚本跟写生产代码一样，要讲究工程规范。别因为是测试代码就随便写，最后坑的还是你自己。
公众号:蓝海资料掘金营，微信deep3321