1. 为什么我最终选择了脚本化压测这条路做后端或者中间件开发的朋友大概都有过这样的经历新版本上线前老板或者测试同学跑过来问一句“这个接口能扛多少并发”然后你就得硬着头皮去准备一轮性能测试。早些年我用过不少现成的压测工具图形化界面点一点参数填一填报告导出来看着也挺唬人。但真到了要复现问题、要精细控制请求节奏、要在CI流水线里自动跑一轮回归的时候那些工具就开始各种别扭了。后来我慢慢把主力方案收敛到了“用Python脚本直接调接口”这条路上。原因其实很朴素接口测试的本质就是发请求、收响应、统计结果而Python在这三件事上都有极其成熟的库胶水能力又强想怎么拼就怎么拼。你不需要去学某个工具特有的DSL也不用被它的插件体系绑架脚本本身就是最好的文档改一行参数就能换一个场景。这篇文章我想聊的不是某个库的API手册而是我这些年踩坑踩出来的一整套“套路”——怎么设计压测脚本的结构、怎么控制并发模型、怎么统计出真正有意义的指标、怎么把稳定性和性能两件事一起测了。适合已经会写Python、但对性能测试还没有系统方法论的同学也适合那些用惯了图形化工具、想往自动化方向走一步的同行。核心关键词就三个接口调用、性能测试、稳定性验证下面所有的内容都围绕它们展开。2. 压测脚本的整体设计思路拆解2.1 先想清楚你要测的到底是“性能”还是“稳定性”很多人把这两个词混着用但它们的测试目标、脚本设计、结果判读方式完全不同混在一起做最后往往两头都不讨好。性能测试关注的是“极限在哪”。你要回答的问题是在多少并发下响应时间的P99会突破阈值吞吐量的拐点出现在哪个并发数这时候脚本的设计重点是梯度加压从低并发逐步往上顶观察指标随压力的变化曲线。你需要的是精确的并发控制、干净的计时点、以及对服务端资源的同步监控。稳定性测试关注的是“能不能一直跑”。你要回答的问题是持续跑8小时、24小时内存会不会涨连接会不会泄漏错误率会不会随时间漂移这时候脚本的重点变成了长时间匀速运行、异常自愈、以及趋势数据的采集。并发数通常不高但要跑得久脚本本身不能先崩了。我一般的做法是同一套脚本骨架通过配置切换两种模式。性能模式用梯度并发稳定性模式用固定并发加长时长。这样代码复用率高也不用维护两套逻辑。下面会详细讲这个骨架怎么搭。2.2 为什么不用现成压测工具而自己写脚本这个问题我被问过很多次。现成工具当然有它的价值比如开箱即用、报告漂亮。但自己写脚本有几个绕不开的优势协议适配灵活现成工具对自定义签名、加密参数、动态token的处理往往很别扭要么写插件要么用前置脚本折腾半天。Python里就是一个函数的事。场景编排自由真实业务往往是有依赖的比如先登录拿token再查列表再下单。脚本里就是几个函数顺序调用清晰可控。结果统计可定制我想要的分位数、错误分类、时间窗口聚合脚本里想怎么算就怎么算不用被工具的报告模板限制。易于集成直接扔进CI配合pytest或者自己写的runner每次发版自动跑一轮冒烟压测这是图形化工具很难做到的。当然代价是你要自己处理并发、自己算统计量。但这部分工作是一次性的搭好骨架之后后面都是填业务逻辑性价比很高。2.3 并发模型的选择线程、协程还是多进程这是脚本设计里最关键的一个决策选错了后面全是坑。我把三种模型的适用场景整理成一张表方便你对照自己的需求并发模型适用场景优点缺点多线程IO密集型、请求量中等几百并发编程模型直观库兼容性好GIL限制线程数过高时上下文切换开销大协程asyncioIO密集型、高并发上千并发资源占用低并发能力强需要异步库支持同步代码要改造多进程CPU密集型、需要压满多核绕开GIL真正并行进程间通信麻烦内存开销大对于绝大多数HTTP接口压测请求本身是IO密集型的瓶颈在等待服务端响应不在本地CPU。所以协程是理论上最优的选择。但现实是很多同学的业务代码里用了同步的SDK改造成异步成本高。这种情况下多线程是更务实的选择几百并发用线程池完全扛得住。我自己的习惯是新项目一律用asyncio aiohttp老项目如果改造困难就用线程池 requests。两种方案的骨架我会在下一节都给出。2.4 计时点的选择决定了你的数据有没有意义这是一个特别容易被忽视的细节。你测出来的“响应时间”到底是从哪个时刻到哪个时刻我见过不少脚本是这么写的在循环里记录一个开始时间发请求收到响应记录结束时间两者相减。看起来没问题但如果你的脚本里还有参数组装、日志打印、结果写入这些操作它们的时间就被算进去了测出来的数字偏大。正确的做法是把计时点尽量贴近网络调用本身。用requests的话可以在发送前打一个时间戳收到响应后立刻打一个时间戳中间不要插入任何其他操作。用aiohttp也是同理。另外要注意连接建立的时间要不要算进去——如果测的是服务端处理能力通常希望复用连接把TCP握手和TLS握手排除掉如果测的是端到端的用户体验那就应该算进去。这个要在脚本里明确不能含糊。3. 核心细节解析与实操要点3.1 请求构造把可变部分参数化压测脚本最忌讳把请求写死。你今天测的是用户ID为1的查询明天要测1000个不同用户的查询如果ID是硬编码的脚本就废了。我的做法是把请求的所有可变部分抽成配置包括URL、方法、headers、body模板、参数取值范围。用一个配置字典或者dataclass来管理脚本主体只负责读取配置、渲染请求、发送、收集结果。这样换一个接口测试只需要改配置不用动代码。from dataclasses import dataclass, field from typing import Any dataclass class RequestSpec: method: str url: str headers: dict field(default_factorydict) params: dict field(default_factorydict) json_body: dict field(default_factorydict) timeout: float 5.0参数化还有一个好处是方便做数据隔离。比如压测下单接口如果所有请求都用同一个用户可能会触发服务端的限流或者幂等逻辑测出来的数据就不准。用不同的用户ID、不同的商品ID能更真实地模拟线上流量。3.2 并发控制信号量和线程池的正确用法并发控制的核心是限制同时在飞的请求数量。如果不加限制你起1000个协程它们会瞬间全部发出去本地端口可能先耗尽测出来的错误全是连接失败跟服务端能力没关系。用asyncio的话标准做法是用asyncio.Semaphoreimport asyncio import aiohttp async def worker(session, sem, spec, results): async with sem: start time.perf_counter() try: async with session.request( spec.method, spec.url, headersspec.headers, paramsspec.params, jsonspec.json_body, timeoutspec.timeout ) as resp: body await resp.read() elapsed time.perf_counter() - start results.append({ status: resp.status, elapsed: elapsed, size: len(body) }) except Exception as e: elapsed time.perf_counter() - start results.append({ status: -1, elapsed: elapsed, error: str(e) })这里的sem就是信号量它的值决定了最大并发数。注意async with sem要包住整个请求过程包括读body否则并发限制是不完整的。用线程池的话concurrent.futures.ThreadPoolExecutor的max_workers就是并发上限。但要注意线程池的并发是“同时存在的线程数”不是“同时在飞的请求数”如果每个任务里有多次请求实际并发会更高。这个区别在精细压测时要留意。3.3 结果统计别只看平均值平均值是性能测试里最没用的指标之一。它会把长尾的慢请求平均掉让你误以为系统很健康。真正要看的是分位数尤其是P95、P99它们反映了大多数用户的真实体验和极端情况。我一般会在脚本里收集每个请求的耗时最后统一计算import statistics def summarize(results): latencies sorted(r[elapsed] for r in results if r[status] ! -1) total len(results) errors sum(1 for r in results if r[status] -1 or r[status] 500) def percentile(data, p): if not data: return 0 k (len(data) - 1) * p f int(k) c min(f 1, len(data) - 1) return data[f] (data[c] - data[f]) * (k - f) return { total: total, errors: errors, error_rate: errors / total if total else 0, avg: statistics.mean(latencies) if latencies else 0, p50: percentile(latencies, 0.50), p95: percentile(latencies, 0.95), p99: percentile(latencies, 0.99), max: max(latencies) if latencies else 0, }除了分位数还要关注错误率的构成。是连接超时还是服务端返回5xx还是业务层面的错误码不同错误对应的排查方向完全不同。我习惯在结果里保留状态码和错误信息最后按类型聚合。3.4 稳定性测试的额外考量趋势和自愈稳定性测试和性能测试最大的区别是时间维度。跑8小时你要看的不是某一时刻的指标而是指标随时间的变化趋势。我的做法是按时间窗口聚合比如每5分钟统计一次错误率、P99、吞吐量最后画成趋势。如果错误率随时间缓慢上升或者P99逐渐恶化即使绝对值还在阈值内也说明系统有隐患比如内存泄漏、连接池耗尽、日志文件撑爆磁盘。另外稳定性脚本要能自愈。网络抖动导致的偶发失败不应该让整个脚本崩掉要有重试机制但重试次数要限制否则会掩盖真实问题。我的经验是连接类错误重试2次业务类错误不重试因为业务错误重试也没用反而会污染数据。4. 完整实操流程与核心环节实现4.1 环境准备与依赖安装先把基础环境搭起来。Python版本建议3.8以上因为asyncio的一些特性在低版本上不完善。依赖库根据你选的并发模型来# 协程方案 pip install aiohttp # 线程方案 pip install requests # 统计和报告可选 pip install numpy tabulate如果你要做更专业的统计numpy的percentile函数比自己写的更准也更快。tabulate用来把结果打成表格方便贴到报告里。提示压测机和被测服务不要在同一台机器上否则压测脚本本身消耗的CPU和内存会影响服务端表现数据不可信。如果条件有限至少要把压测脚本的日志输出关掉或者异步化减少干扰。4.2 脚本骨架搭建配置、执行、统计三段式我把脚本分成三个部分配置区、执行区、统计区。配置区定义请求规格和压测参数执行区负责并发调度统计区负责结果聚合和输出。这样结构清晰改哪部分都不会影响其他部分。import asyncio import aiohttp import time from dataclasses import dataclass, field dataclass class LoadConfig: concurrency: int 50 duration: int 60 # 秒稳定性测试用 total_requests: int 0 # 性能测试用0表示按duration跑 ramp_up: int 0 # 梯度加压的步长0表示不梯度 dataclass class RequestSpec: method: str GET url: str headers: dict field(default_factorydict) params: dict field(default_factorydict) json_body: dict field(default_factorydict) timeout: float 5.0执行区是核心我一般写一个run_load函数根据配置决定是跑固定时长还是固定请求数是否梯度加压async def run_load(spec, config): results [] sem asyncio.Semaphore(config.concurrency) connector aiohttp.TCPConnector(limitconfig.concurrency * 2) async with aiohttp.ClientSession(connectorconnector) as session: if config.total_requests 0: # 性能模式固定请求数 tasks [ worker(session, sem, spec, results) for _ in range(config.total_requests) ] await asyncio.gather(*tasks) else: # 稳定性模式固定时长 end_time time.time() config.duration tasks [] while time.time() end_time: tasks.append(asyncio.create_task( worker(session, sem, spec, results) )) await asyncio.sleep(0.001) # 控制发起速率 await asyncio.gather(*tasks) return results这里有个细节稳定性模式下我用while循环持续创建任务但用sem限制同时在飞的数量。await asyncio.sleep(0.001)是为了避免创建任务的速度过快把内存撑爆。这个值要根据你的并发数和请求耗时调整原则是任务创建速度略快于完成速度让信号量始终处于饱和状态。4.3 梯度加压的实现与拐点识别性能测试最有价值的部分是找到吞吐量的拐点。做法是从低并发开始每隔一段时间提升一档并发观察吞吐量和响应时间的变化。async def run_ramp_up(spec, config): all_results [] step config.ramp_up current step while current config.concurrency: print(f--- 当前并发: {current} ---) step_results await run_load(spec, LoadConfig( concurrencycurrent, duration30, # 每档跑30秒 total_requests0 )) summary summarize(step_results) summary[concurrency] current all_results.append(summary) print(summary) current step return all_results判读拐点的经验是当并发提升后吞吐量不再增长而P99开始明显上升这个点就是拐点。拐点之前的并发数是系统的健康容量拐点之后的并发数就是危险区。实际生产中我会把健康容量打个七折作为线上限流阈值留出余量。4.4 稳定性测试的长跑与监控稳定性测试我一般跑4小时起步重要服务跑24小时。脚本本身要能扛住长时间运行几个要点结果不要全存内存跑24小时如果QPS是100就是864万条记录全存内存会爆。我的做法是定期落盘比如每10万条写一次文件内存里只保留聚合数据。日志要克制每条请求都打日志磁盘IO会成为瓶颈。只在错误时打日志正常请求只计数。加一个心跳输出每5分钟打印一次当前统计方便你实时观察也方便判断脚本是不是卡死了。async def stability_run(spec, config, report_interval300): results [] last_report time.time() end_time time.time() config.duration async with aiohttp.ClientSession() as session: sem asyncio.Semaphore(config.concurrency) while time.time() end_time: task asyncio.create_task(worker(session, sem, spec, results)) await asyncio.sleep(0.001) if time.time() - last_report report_interval: summary summarize(results[-10000:]) # 最近1万条 print(f[{time.strftime(%H:%M:%S)}] {summary}) last_report time.time() await asyncio.gather(*[t for t in asyncio.all_tasks() if t is not asyncio.current_task()]) return results4.5 结果输出与报告生成最后把统计结果整理成可读的报告。我习惯输出两部分控制台摘要和CSV明细。摘要给人看明细给后续分析用。def print_report(summary): print(\n 压测报告 ) print(f总请求数: {summary[total]}) print(f错误数: {summary[errors]} ({summary[error_rate]:.2%})) print(f平均耗时: {summary[avg]*1000:.2f} ms) print(fP50: {summary[p50]*1000:.2f} ms) print(fP95: {summary[p95]*1000:.2f} ms) print(fP99: {summary[p99]*1000:.2f} ms) print(f最大耗时: {summary[max]*1000:.2f} ms) print(f吞吐量: {summary[total]/summary[duration]:.2f} req/s)吞吐量的计算要注意分母是实际测试时长不是配置时长。因为脚本启动和收尾有时间开销用配置时长会略微高估。5. 常见问题与排查技巧实录5.1 压测结果不可信的几种典型情况压测最怕的不是数字难看而是数字好看但不可信。下面这几种情况我踩过不止一次现象可能原因排查方法错误率极高全是连接失败本地端口耗尽或文件描述符限制检查ulimit -n调大用连接池复用连接P99异常高但P50正常少量请求触发了慢路径如GC、锁竞争看服务端日志定位慢请求的特征吞吐量上不去CPU也不高压测机成为瓶颈或服务端有连接数限制换更强的压测机检查服务端连接池配置结果波动极大每次跑都不一样环境噪声如其他进程干扰固定环境关闭无关进程多跑几次取中位数注意压测前一定要确认压测机的网络带宽和文件描述符限制。我遇到过压测机网卡打满结果测出来服务端“很闲”的情况排查了半天才发现是压测机的问题。5.2 连接池与超时设置的坑用aiohttp的时候默认的连接池大小是100如果你设的并发是200超出的请求会排队等连接而不是真正并发。这时候你测出来的并发其实是假的。解决办法是显式设置connector的limit让它大于等于你的并发数。connector aiohttp.TCPConnector( limitconfig.concurrency * 2, limit_per_hostconfig.concurrency * 2, ttl_dns_cache300 )超时设置也有讲究。timeout参数如果设得太短正常但稍慢的请求会被判为失败错误率虚高设得太长脚本会被慢请求拖住吞吐量上不去。我的经验是先跑一轮小并发看看P99是多少然后把超时设成P99的3到5倍。这样既能捕获真正的异常又不会误伤正常请求。5.3 稳定性测试中的内存与句柄泄漏排查稳定性测试跑久了脚本本身可能会泄漏。最常见的是响应对象没有正确释放。用aiohttp的时候async with session.request(...)会自动释放但如果你用session.get()然后忘了读body连接就不会归还到池里跑着跑着就卡住了。另一个坑是结果列表无限增长。前面说过要定期落盘但如果你忘了清空内存里的列表跑24小时内存照样爆。我的做法是每落盘一次就results.clear()只保留聚合用的计数器。排查脚本自身泄漏的方法很简单在心跳输出里加上进程的内存占用和打开的文件描述符数量。如果这两个数字随时间单调上升基本可以确定是脚本泄漏而不是服务端问题。import os import psutil def print_self_status(): proc psutil.Process(os.getpid()) mem proc.memory_info().rss / 1024 / 1024 fds proc.num_fds() if hasattr(proc, num_fds) else N/A print(f脚本内存: {mem:.1f} MB, 文件描述符: {fds})5.4 如何区分是服务端问题还是压测脚本问题这是排查时最核心的问题。我的判断逻辑是换一个简单的接口测如果简单接口也大量失败大概率是压测脚本或压测机的问题。降低并发再测如果低并发下一切正常高并发下才出问题那更可能是服务端容量问题。看错误类型连接超时、连接拒绝通常是压测机或网络问题服务端返回5xx、响应变慢通常是服务端问题。同步监控服务端压测时盯着服务端的CPU、内存、GC、连接数如果服务端资源已经打满那瓶颈就在服务端。我一般会在压测脚本里加一个**“单请求验证”模式**先发一个请求确认接口通不通再开始压测。这样能排除掉URL写错、参数不对这类低级问题。5.5 独家避坑技巧汇总最后分享几个文档里不会写、但实际很管用的技巧预热很重要JVM类服务需要预热刚启动时性能很差。正式压测前先跑1到2分钟低并发预热让JIT编译和连接池都进入状态。多轮取中位数单轮压测受环境影响大我一般跑3轮取中位数作为最终结果。如果三轮差异超过20%说明环境不稳定要排查。记录压测时的服务端版本和配置不然过两周你看到报告根本想不起来当时测的是哪个版本数据就白测了。把压测脚本纳入版本管理接口变了脚本也要跟着改。放在Git里和代码一起review能避免脚本腐化。错误信息要完整保留不要只记状态码把响应体的前几百个字符也记下来。很多时候错误原因就藏在响应体里。这套脚本骨架我用了好几年从简单的查询接口到复杂的下单链路基本都能覆盖。核心思想就一句话把压测当成一个正经的软件项目来做而不是临时写几行代码应付差事。骨架搭好了后面每次压测都是填配置、跑脚本、看报告效率比每次从头折腾高得多。