1. 脚本化性能与稳定性测试的整体设计思路

1.1 为什么选择 Python 脚本而不是现成工具

做接口性能和稳定性测试，很多人第一反应是找现成的压测平台或者图形化工具。我早期也用过一些，但实际项目里很快会遇到几个绕不开的问题：一是接口鉴权逻辑复杂，签名、加密、动态 token 这些现成工具配置起来很别扭；二是测试数据需要从数据库或另一个接口动态生成，工具的参数化能力跟不上；三是稳定性测试往往要跑几个小时甚至几天，需要自定义的监控、日志和中断恢复逻辑，图形化工具很难做到精细控制。

Python 脚本的核心优势就在这里——它本质上是"用代码描述测试行为"，任何你能用代码表达的逻辑，都能塞进测试流程里。签名算法复杂？直接调项目里的签名函数。测试数据要动态生成？写个生成器。需要边压测边查数据库校验数据一致性？加一段查询逻辑就行。这种灵活性是通用压测工具给不了的。

当然，脚本化也有代价。你得自己处理并发模型、连接池、结果统计、异常重试这些基础设施。但好消息是，这些"轮子"在 Python 生态里都有成熟方案，我们更多是在做组装和业务适配，而不是从零造。

1.2 性能测试与稳定性测试的目标差异

这两个词经常被混着说，但目标其实不一样，脚本设计思路也不同。

性能测试关注的是"极限在哪"。核心指标是吞吐量（QPS/TPS）、响应时间（P50/P95/P99）、错误率随并发压力的变化曲线。它的典型做法是阶梯式加压：从低并发开始，逐步提升，观察系统在哪个点开始劣化，找到性能拐点和最大承载能力。脚本设计上，重点是并发控制要精准、计时要准确、结果要能按并发梯度分组统计。

稳定性测试关注的是"能不能一直扛住"。核心指标是长时间运行下的错误率是否稳定、内存和连接数是否泄漏、响应时间是否随时间漂移。它的典型做法是恒定压力长时间运行，比如用 70% 的预估最大承载量跑 8 到 24 小时。脚本设计上，重点是异常要能自愈（比如连接断了自动重连）、资源要能回收、日志要能滚动、中途挂了要能续跑。

我一般会把两者做成同一套脚本的两个模式，共用请求构造和结果收集模块，只是加压策略和运行时长不同。这样维护成本低，也不会出现"性能测试脚本"和"稳定性脚本"两套代码逻辑不一致的问题。

1.3 整体架构分层

一套能打的测试脚本，我习惯分成四层，各层职责清晰，方便替换和复用：

这样分层的好处是，换一个接口测试，只需要改请求构造层；换一种并发模型，只动调度层；统计口径要调整，只改统计层。实际项目里接口经常变，分层能省下大量重复劳动。

提示：不要一上来就追求"大而全"的框架。我见过太多人花两周写了个通用压测框架，结果业务接口一改就全废。先用最朴素的脚本把当前接口压通，再逐步抽象，才是务实的路子。

2. 核心细节解析与实操要点

2.1 并发模型的选择：线程、协程还是多进程

这是脚本化压测第一个要拍板的事，选错了后面全是坑。

多线程是最直观的。Python 的concurrent.futures.ThreadPoolExecutor几行代码就能起几百个并发。但要注意 GIL 的存在——不过对于 IO 密集型的接口测试，GIL 影响其实没那么大，因为线程大部分时间在等网络返回，释放了 GIL。实测下来，单机几百到一千并发用线程池是够用的。

协程（asyncio + aiohttp）在 IO 密集场景下效率更高，单机轻松撑起几千并发，内存占用也低得多。代价是代码要写成 async/await 风格，遇到同步的库（比如某些加密库、数据库驱动）会阻塞事件循环，得用run_in_executor包一层，稍微麻烦。

多进程主要用于突破单核 CPU 瓶颈，或者需要真正并行计算的场景。接口压测里一般用不上，除非你要在压测的同时做大量本地数据处理。

我的经验是：并发需求在 500 以内，直接用线程池，代码简单好维护；超过 500 或者要跑长时间稳定性测试，上 asyncio，省资源。下面给一个线程池的基础骨架：

import time import requests from concurrent.futures import ThreadPoolExecutor, as_completed def send_request(session, url, payload): start = time.perf_counter() try: resp = session.post(url, json=payload, timeout=5) cost = (time.perf_counter() - start) * 1000 return {"code": resp.status_code, "cost": cost, "ok": resp.status_code == 200} except Exception as e: cost = (time.perf_counter() - start) * 1000 return {"code": -1, "cost": cost, "ok": False, "err": str(e)} def run(concurrency, total): results = [] with ThreadPoolExecutor(max_workers=concurrency) as pool: with requests.Session() as session: futures = [pool.submit(send_request, session, URL, PAYLOAD) for _ in range(total)] for f in as_completed(futures): results.append(f.result()) return results

这里有个细节：requests.Session()要复用。每次请求新建连接开销很大，复用 Session 能走连接池，性能差别在高压下非常明显。但要注意 Session 不是线程安全的，多线程共享一个 Session 在高并发下可能出问题，稳妥做法是每个线程一个 Session，用threading.local()管理。

2.2 请求构造与鉴权处理

接口测试最容易翻车的地方就是鉴权。很多接口不是简单传个 token 就行，而是要求时间戳、随机数、签名一起算。我的做法是直接把项目里的签名逻辑抽出来复用，而不是在测试脚本里重新实现一遍——重新实现几乎必然和线上逻辑有偏差，测出来的结果没意义。

import hashlib import time import uuid def build_signed_payload(biz_data, secret): ts = str(int(time.time())) nonce = uuid.uuid4().hex raw = f"{biz_data}{ts}{nonce}{secret}" sign = hashlib.sha256(raw.encode()).hexdigest() return { "data": biz_data, "timestamp": ts, "nonce": nonce, "sign": sign }

时间戳和随机数每次请求都要重新生成，不能复用，否则服务端的防重放校验会直接拒绝。这一点在压测时特别容易忽略——有人为了"省事"把整个 payload 缓存下来反复发，结果全被服务端判重拦截，错误率 100%，还以为是接口扛不住。

注意：签名里的时间戳如果和服务端时间偏差过大（一般超过 5 分钟），也会被拒。压测机的时间要和服务端对齐，最好挂个时间同步。

2.3 结果统计的口径设计

统计口径决定了你报告的可信度。我见过有人只统计"成功请求的平均耗时"，结果把超时的、报错的都排除了，得出一个漂亮但毫无意义的数字。正确的做法是：

• 总请求数：发出去多少
• 成功数 / 失败数：按状态码和业务码分类
• 耗时分布：不能只看平均值，必须看 P50、P90、P95、P99
• 错误分类：超时、连接失败、4xx、5xx、业务错误码，分开统计

耗时分布的计算，简单做法是把所有耗时排序后取分位值。数据量大时可以用直方图分桶，避免内存爆掉。

import numpy as np def summarize(results): costs = [r["cost"] for r in results if r["ok"]] total = len(results) ok = sum(1 for r in results if r["ok"]) if costs: arr = np.array(costs) p50, p90, p95, p99 = np.percentile(arr, [50, 90, 95, 99]) else: p50 = p90 = p95 = p99 = 0 return { "total": total, "ok": ok, "error_rate": round((total - ok) / total * 100, 2), "p50": round(p50, 1), "p90": round(p90, 1), "p95": round(p95, 1), "p99": round(p99, 1), }

平均值会骗人，P99 才是用户真实感受到的"最差体验"。一个接口平均 50ms，但 P99 是 3 秒，说明有 1% 的请求慢得离谱，这在生产环境就是投诉来源。

2.4 稳定性测试的资源管理

稳定性测试跑几个小时，资源管理是生死线。几个必须处理的点：

连接泄漏：如果每次请求都新建连接不复用，跑几小时后本地端口会被耗尽，报 "Cannot assign requested address"。必须用连接池，并设置合理的pool_maxsize。

内存增长：结果列表如果一直往内存里塞，跑几百万请求后内存会爆。做法是分批聚合，比如每 1 万条统计一次就清空，只保留聚合结果。

日志滚动：日志文件不滚动，几小时能写到几个 G。用logging.handlers.RotatingFileHandler限制单文件大小和保留数量。

断点续跑：长时间测试中途可能因为网络抖动、机器重启中断。把进度定期写到一个状态文件里，重启后从断点继续，能省下大量重跑时间。

3. 实操过程与核心环节实现

3.1 环境准备与依赖安装

先把基础环境搭起来。Python 版本建议 3.8 以上，依赖不多：

pip install requests aiohttp numpy

如果要用协程方案，aiohttp是必须的；纯线程方案只要requests和numpy（用于分位数计算）。压测机的文件描述符上限要调高，否则高并发下会报 "Too many open files"：

ulimit -n 65535

这个命令只对当前 shell 生效，要永久生效得改/etc/security/limits.conf。我踩过的坑是：本地测试几百并发没事，一上到两千并发就各种连接错误，排查半天才发现是文件描述符限制。

3.2 阶梯加压的实现

性能测试的核心是找到拐点，所以加压要阶梯式。我的做法是定义一组并发梯度，比如 10、50、100、200、500，每个梯度跑固定时长（比如 60 秒），记录该梯度的稳定指标，然后进入下一梯度。

def step_load(url, payload, steps, duration_per_step): report = [] for concurrency in steps: print(f"=== 并发 {concurrency} 开始 ===") results = [] end_time = time.time() + duration_per_step with ThreadPoolExecutor(max_workers=concurrency) as pool: with requests.Session() as session: while time.time() < end_time: batch = [pool.submit(send_request, session, url, payload) for _ in range(concurrency)] for f in as_completed(batch): results.append(f.result()) summary = summarize(results) summary["concurrency"] = concurrency report.append(summary) print(summary) return report

这里每个梯度内是"发一批、等一批"的节奏，保证并发数稳定。更精细的做法是用固定速率发射（比如每秒发 N 个），但那需要更复杂的调度器，一般阶梯加压够用了。

3.3 稳定性测试的长时间运行

稳定性测试我一般用固定并发跑固定时长，重点是监控指标随时间的变化。做法是每隔一段时间（比如 30 秒）输出一次当前窗口的统计，这样能看出指标是否漂移。

def stability_test(url, payload, concurrency, total_duration): window = 30 start = time.time() with ThreadPoolExecutor(max_workers=concurrency) as pool: with requests.Session() as session: while time.time() - start < total_duration: window_results = [] window_end = time.time() + window while time.time() < window_end: batch = [pool.submit(send_request, session, url, payload) for _ in range(concurrency)] for f in as_completed(batch): window_results.append(f.result()) s = summarize(window_results) s["elapsed"] = round(time.time() - start, 1) print(s)

跑稳定性测试时，我习惯同时开一个终端用top或htop盯着压测机的 CPU 和内存，再开一个盯着被测服务的监控面板。三方对照，才能判断问题出在压测端还是服务端。

3.4 结果落盘与报告生成

测试跑完，结果要落成文件，方便后续分析。我一般输出两份：一份是每个梯度的汇总（CSV），一份是原始明细（可选，量大时抽样）。

import csv def save_report(report, path="perf_report.csv"): keys = ["concurrency", "total", "ok", "error_rate", "p50", "p90", "p95", "p99"] with open(path, "w", newline="") as f: writer = csv.DictWriter(f, fieldnames=keys) writer.writeheader() for row in report: writer.writerow({k: row.get(k, "") for k in keys})

CSV 的好处是能直接丢进表格软件画曲线图。并发数做横轴，QPS 和 P99 做纵轴，拐点一目了然。

3.5 一个完整的调用示例

把上面的模块拼起来，一个完整的测试入口大概长这样：

if __name__ == "__main__": URL = "http://your-service/api/endpoint" PAYLOAD = build_signed_payload({"userId": 1001}, "your_secret") # 性能测试：阶梯加压 steps = [10, 50, 100, 200, 500] report = step_load(URL, PAYLOAD, steps, duration_per_step=60) save_report(report, "perf_report.csv") # 稳定性测试：固定并发跑 2 小时 # stability_test(URL, PAYLOAD, concurrency=100, total_duration=7200)

实际用的时候，性能和稳定性分开跑，不要一次全上。先跑性能找到拐点，再用拐点的 70% 左右作为稳定性测试的并发量，这样最有参考价值。

4. 常见问题与排查技巧实录

4.1 压测端自身成为瓶颈

这是最隐蔽的坑。你以为是服务端扛不住，其实是压测机 CPU 跑满了、或者 Python 单进程处理不过来。判断方法很简单：压测时看压测机的 CPU 和内存。如果压测机 CPU 已经 90% 以上，那测出来的"服务端性能"其实是压测机的性能。

解决办法：一是换协程方案降低单机开销；二是多机分布式压测，每台机器跑一部分并发；三是减少压测端的额外开销，比如关掉不必要的日志、别在请求路径里做复杂计算。

4.2 连接池配置不当导致的假错误

requests默认的连接池大小是 10，如果你开 100 并发，多出来的请求会排队等连接，表现为响应时间虚高。要显式配置：

from requests.adapters import HTTPAdapter session = requests.Session() adapter = HTTPAdapter(pool_connections=100, pool_maxsize=100) session.mount("http://", adapter) session.mount("https://", adapter)

pool_maxsize要大于等于你的并发数，否则连接复用不起来。

4.3 超时设置与错误分类

超时一定要设，而且连接超时和读取超时要分开设。不设超时的话，一个卡住的请求会一直占着线程，并发数虚高实际有效并发很低。

resp = session.post(url, json=payload, timeout=(3, 10))

(3, 10)表示连接超时 3 秒、读取超时 10 秒。超时后要区分是连接超时还是读取超时，前者说明网络或服务不可达，后者说明服务处理慢，排查方向完全不同。

4.4 常见问题速查表

4.5 几个压箱底的经验

先单请求验证再上并发。我每次写新脚本，第一件事是发一个请求，把响应打印出来，确认签名、参数、返回都正常，再上并发。跳过这步直接压，出问题根本不知道是脚本错还是服务错。

测试数据要贴近真实分布。如果所有请求都用同一个 userId，服务端的缓存会全部命中，测出来的性能虚高。要用一批随机或轮询的 ID，模拟真实流量。

关注服务端指标，不只看客户端。客户端看到的响应时间包含了网络传输，服务端看到的才是真实处理时间。两边对照，才能定位瓶颈在网络还是在服务。

稳定性测试要留观察期。跑完别急着下结论，停压后再观察 10 分钟，看服务端资源是否回落、有没有积压的任务在慢慢消化。有些问题在停压后才暴露。

报告要带上下文。光给一堆数字没意义，要写清楚测试环境、数据量、并发模型、被测服务版本。不然过两周自己都看不懂当时测的是什么。

我个人在实际操作中的体会是，脚本化压测最值钱的不是脚本本身，而是你对"什么算好、什么算坏"的判断力。工具谁都能写，但能从一堆数字里看出服务端的真实状态、能预判哪个指标会先崩，这个得靠一次次实测积累。刚开始别怕测错，多测几次，把每次的异常和排查过程记下来，慢慢就有手感了。