本文目录导读：
可以，实用脚本”正是压力测试自动化的核心，完全可以通过脚本自动执行压力测试，包括启动测试、监控指标、生成报告,甚至根据结果自动调整策略。
下面从如何实现、常用工具和一个实战脚本示例三个方面来说明。
脚本如何实现自动化压力测试？
一个自动化的压力测试脚本通常需要完成以下步骤：
- 初始化环境：确保被测服务、监控工具（如Prometheus）可用。
- 配置测试参数：设置并发用户数、请求速率、持续时间、请求地址等。
- 执行压测：调用压测工具（如
ab、wrk、vegeta、locust）的命令行接口。 - 实时监控：在压测过程中，定时抓取服务器CPU、内存、QPS、错误率等指标。
- 动态调整（高级）：如果错误率过高，自动降低并发；如果性能有余量,自动提升并发。
- 停止与清理：测试结束后，停止压测工具,收集日志。
- 生成报告：解析压测输出和监控数据，生成HTML/JSON报告。
常用工具与脚本语言
- 压测引擎（命令行友好）：
vegeta：纯Go编写，输出为JSON,非常适合脚本解析。wrk/wrk2：非常高效,适合高并发。ab(ApacheBench)：简单,但功能有限。locust：Python编写，支持分布式和Web UI，可通过--headless模式无界面运行。
- 脚本语言：
- Bash/Shell：最轻量,适合快速启动和简单监控。
- Python：最佳选择，生态丰富（
subprocess、requests、json、matplotlib）,能轻松处理复杂逻辑和数据可视化。 - PowerShell：Windows环境下的首选。
实战脚本示例（Python + Vegeta）
这个脚本可以自动执行一个分阶段（递增并发）的压力测试,并输出一份友好的摘要报告。
#!/usr/bin/env python3
"""
自动压力测试脚本 - 使用 Vegeta 引擎 (需要先安装: brew install vegeta)
步骤:
1. 定义测试目标 (URL, 速率, 持续时间)
2. 使用 subprocess 调用 vegeta attack
3. 解析 vegeta 输出的 JSON 结果
4. 打印和保存测试摘要
"""
import subprocess
import json
import sys
import time
from datetime import datetime
# === 配置项 ===
TARGET_URL = "http://localhost:8080/api/health" # 替换为你的测试地址
RATE = 100 # 每秒请求数 (RPS)
DURATION = "10s" # 持续时间
REPORT_FILE = f"report_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json"
def run_pressure_test():
"""执行压力测试并返回解析后的结果"""
print(f"[{datetime.now()}] 开始压力测试...")
print(f"目标: {TARGET_URL}")
print(f"速率: {RATE} RPS, 持续: {DURATION}")
# 1. 构造 vegeta 命令 (使用管道: echo "GET url" | vegeta attack -rate ...)
# 这样不需要额外创建 target 文件
vegeta_attack = [
"vegeta", "attack",
"-targets=", # 从 stdin 读取目标
"-rate", str(RATE),
"-duration", DURATION,
"-format", "json"
]
# 构造目标输入
target_input = f"GET {TARGET_URL}\n"
try:
# 启动 vegeta attack 进程
attack_proc = subprocess.Popen(
vegeta_attack,
stdin=subprocess.PIPE,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True
)
# 发送目标并等待完成
stdout, stderr = attack_proc.communicate(input=target_input, timeout=120)
if attack_proc.returncode != 0:
print(f"错误: vegeta 返回非零状态码 {attack_proc.returncode}")
print(f"Stderr: {stderr}")
sys.exit(1)
# 2. 解析 vegeta 的二进制输出 (stdout) 为 JSON
# vegeta 默认输出二进制编码，需要用 vegeta encode 转码
encode_proc = subprocess.Popen(
["vegeta", "encode", "--format", "json"],
stdin=subprocess.PIPE,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=False # 二进制模式
)
encoded_stdout, encode_stderr = encode_proc.communicate(
input=stdout.encode(), timeout=30
)
if encode_proc.returncode != 0:
print(f"编码错误: {encode_stderr.decode()}")
sys.exit(1)
# 3. 解析 JSON 结果 (每个请求一行)
results = []
for line in encoded_stdout.decode().strip().split("\n"):
if line:
results.append(json.loads(line))
# 4. 生成摘要报告 (使用 vegeta report)
report_proc = subprocess.run(
["vegeta", "report", "--type", "json"],
input=encoded_stdout,
capture_output=True,
text=False,
timeout=30
)
if report_proc.returncode == 0:
report_data = json.loads(report_proc.stdout)
else:
report_data = {}
return report_data, results
except subprocess.TimeoutExpired:
print("错误: 测试超时 (120s)")
sys.exit(1)
except FileNotFoundError:
print("错误: 未找到 vegeta，请先安装: https://github.com/tsenart/vegeta")
sys.exit(1)
def print_report(report):
"""打印人类可读的测试报告"""
print("\n" + "="*50)
print("压力测试报告摘要")
print("="*50)
if not report:
print("无法生成报告")
return
# 提取关键指标
latency = report.get("latencies", {})
status_codes = report.get("status_codes", {})
print(f"总请求数: {report.get('requests', 0)}")
print(f"成功率: {report.get('success', 0)*100:.2f}%")
print(f"平均延迟: {latency.get('mean', 'N/A')}")
print(f"P50延迟: {latency.get('50th', 'N/A')}")
print(f"P95延迟: {latency.get('95th', 'N/A')}")
print(f"P99延迟: {latency.get('99th', 'N/A')}")
print(f"最大延迟: {latency.get('max', 'N/A')}")
print(f"吞吐量: {report.get('throughput', 0):.2f} 请求/秒")
# 状态码分布
print("\nHTTP状态码分布:")
for code, count in sorted(status_codes.items()):
print(f" {code}: {count}")
# === 执行 ===
if __name__ == "__main__":
report, raw_results = run_pressure_test()
print_report(report)
# 保存完整报告到文件
with open(REPORT_FILE, "w") as f:
json.dump(report, f, indent=2)
print(f"\n详细报告已保存至: {REPORT_FILE}")
如何让脚本更“实用”？
-
支持动态参数：通过命令行参数传入
--url、--rate、--duration,而不是硬编码。# 使用 argparse 或 click 库 python stress_test.py --url http://myapp.com/api --rate 500 --duration 30s
-
集成监控告警：在脚本中调用
curl或requests查询云厂商或Prometheus的API，获取服务器CPU、数据库连接数等指标,并在超出阈值时自动终止或降级。 -
循环与变体：自动执行多轮测试，每轮递增并发（50, 100, 200, 500），便于找出服务的“拐点”（性能极限）。
-
结果对比与阈值断言：在脚本最后自动判断：如果P99延迟 > 500ms 或错误率 > 1%，则标记为
FAIL并发送通知（邮件、企业微信、钉钉）。 -
集成到CI/CD：将脚本加入Jenkins、GitLab CI或GitHub Actions，每次代码合并前自动执行一次轻量压力测试,防止回归。
注意事项
- 先测本地，再测远程：先在开发环境验证脚本逻辑,避免直接对生产环境造成影响。
- 控制客户端资源：压测脚本本身也消耗资源（CPU、内存、网络连接），如果并发要求极高，考虑使用分布式压测（Locust或Vegeta的集群模式）。
- 监控被测系统：压测时务必同时监控服务器的CPU、内存、磁盘I/O和网络带宽,否则光看客户端延迟可能误判。
是的，实用脚本完全可以自动执行压力测试。 而且这正是现代DevOps和性能工程的最佳实践，通过组合vegeta/wrk（作为执行引擎）和Python/Bash（作为控制逻辑），你可以构建一个可重复、可衡量、可断言的自动化压测系统，上面的Python + Vegeta脚本是一个可以直接运行的起点。
