本文目录导读：
- 使用 Python + Locust
- 使用 Python + aiohttp (异步高并发)
- 使用 Apache JMeter (GUI工具)
- 使用 k6 (高性能压测工具)
- 常见压测脚本模板
- 性能测试报告生成
- 最佳实践建议
我来详细介绍如何编写接口压测脚本,包含多种常见工具和方法。
使用 Python + Locust
安装
pip install locust
基础脚本示例
from locust import HttpUser, task, between
class WebsiteUser(HttpUser):
wait_time = between(1, 2) # 用户请求间隔
@task
def get_users(self):
self.client.get("/api/users")
@task(3) # 权重为3，执行概率更高
def create_user(self):
payload = {
"name": "test_user",
"email": "test@example.com"
}
self.client.post("/api/users", json=payload)
def on_start(self):
"""每个用户启动时执行"""
self.client.post("/login", {
"username": "test",
"password": "123456"
})
运行命令
# Web界面模式 locust -f locustfile.py --host=http://example.com # 无头模式 locust -f locustfile.py --headless -u 100 -r 10 --run-time 1m
使用 Python + aiohttp (异步高并发)
import asyncio
import aiohttp
import time
import statistics
class PerformanceTester:
def __init__(self, base_url, concurrency=50):
self.base_url = base_url
self.concurrency = concurrency
self.results = []
async def single_request(self, session, endpoint, method='GET', data=None):
start_time = time.time()
try:
url = f"{self.base_url}{endpoint}"
if method == 'GET':
async with session.get(url) as response:
status = response.status
else:
async with session.post(url, json=data) as response:
status = response.status
elapsed = time.time() - start_time
self.results.append({
'status': status,
'time': elapsed,
'success': status == 200
})
return elapsed
except Exception as e:
self.results.append({
'status': 0,
'time': time.time() - start_time,
'success': False,
'error': str(e)
})
async def run_test(self, endpoint, total_requests=1000):
connector = aiohttp.TCPConnector(limit=self.concurrency)
async with aiohttp.ClientSession(connector=connector) as session:
tasks = []
for _ in range(total_requests):
task = self.single_request(session, endpoint)
tasks.append(task)
await asyncio.gather(*tasks)
def report(self):
successful = [r for r in self.results if r['success']]
failed = [r for r in self.results if not r['success']]
times = [r['time'] for r in successful]
print(f"总请求数: {len(self.results)}")
print(f"成功请求: {len(successful)}")
print(f"失败请求: {len(failed)}")
if times:
print(f"平均响应时间: {statistics.mean(times):.3f}s")
print(f"最大响应时间: {max(times):.3f}s")
print(f"最小响应时间: {min(times):.3f}s")
print(f"P50: {sorted(times)[len(times)//2]:.3f}s")
print(f"P99: {sorted(times)[int(len(times)*0.99)]:.3f}s")
print(f"QPS: {len(successful)/sum(times):.2f}")
# 使用示例
async def main():
tester = PerformanceTester("http://api.example.com", concurrency=100)
await tester.run_test("/api/endpoint", total_requests=5000)
tester.report()
asyncio.run(main())
使用 Apache JMeter (GUI工具)
创建测试计划
<?xml version="1.0" encoding="UTF-8"?>
<jmeterTestPlan version="1.2">
<hashTree>
<TestPlan guiclass="TestPlanGui" testclass="TestPlan" testname="API压力测试">
<elementProp name="TestPlan.user_defined_variables" elementType="Arguments">
<collectionProp name="Arguments.arguments">
<elementProp name="base_url" elementType="Argument">
<stringProp name="Argument.name">base_url</stringProp>
<stringProp name="Argument.value">http://api.example.com</stringProp>
</elementProp>
</collectionProp>
</elementProp>
</TestPlan>
<hashTree>
<ThreadGroup guiclass="ThreadGroupGui" testclass="ThreadGroup" testname="用户组">
<stringProp name="ThreadGroup.num_threads">100</stringProp>
<stringProp name="ThreadGroup.ramp_time">10</stringProp>
<stringProp name="ThreadGroup.loop">-1</stringProp>
</ThreadGroup>
<hashTree>
<HTTPSamplerProxy guiclass="HttpTestSampleGui" testclass="HTTPSamplerProxy" testname="API请求">
<stringProp name="HTTPSampler.domain">${base_url}</stringProp>
<stringProp name="HTTPSampler.path">/api/endpoint</stringProp>
<stringProp name="HTTPSampler.method">GET</stringProp>
</HTTPSamplerProxy>
<hashTree/>
<ConstantTimer guiclass="ConstantTimerGui" testclass="ConstantTimer" testname="固定定时器">
<stringProp name="ConstantTimer.delay">1000</stringProp>
</ConstantTimer>
<hashTree/>
</hashTree>
</hashTree>
</hashTree>
</jmeterTestPlan>
JMeter CLI 运行
jmeter -n -t test_plan.jmx -l results.jtl
使用 k6 (高性能压测工具)
// stress_test.js
import http from 'k6/http';
import { check, sleep } from 'k6';
import { Rate, Trend } from 'k6/metrics';
// 自定义指标
const errorRate = new Rate('errors');
const responseTime = new Trend('response_time');
export const options = {
stages: [
{ duration: '2m', target: 100 }, // 逐渐增加到100用户
{ duration: '5m', target: 100 }, // 保持100用户
{ duration: '2m', target: 0 }, // 逐渐减少
],
thresholds: {
http_req_duration: ['p(95)<500'], // 95%的请求在500ms内
errors: ['rate<0.1'], // 错误率小于10%
},
};
export default function () {
const baseUrl = 'http://api.example.com';
// GET请求
const response = http.get(`${baseUrl}/api/users`);
// 检查响应
check(response, {
'状态码是200': (r) => r.status === 200,
'响应时间小于2s': (r) => r.timings.duration < 2000,
});
// 记录错误
errorRate.add(response.status !== 200);
responseTime.add(response.timings.duration);
sleep(1);
}
运行k6
k6 run stress_test.js
常见压测脚本模板
带认证的压测脚本
# locust_with_auth.py
from locust import HttpUser, task, between
import random
class AuthenticatedUser(HttpUser):
wait_time = between(0.5, 1.5)
def on_start(self):
# 获取token
response = self.client.post("/api/login", {
"username": "admin",
"password": "password123"
})
self.token = response.json().get("token")
self.headers = {"Authorization": f"Bearer {self.token}"}
@task
def search_products(self):
params = {
"q": random.choice(["phone", "laptop", "book"]),
"page": random.randint(1, 10),
"limit": 20
}
with self.client.get(
"/api/products",
params=params,
headers=self.headers,
catch_response=True
) as response:
if response.status_code != 200:
response.failure(f"Got status {response.status_code}")
elif response.elapsed.total_seconds() > 5:
response.failure("Response too slow")
混合请求类型
from locust import HttpUser, task, between
import json
class MixedLoadUser(HttpUser):
wait_time = between(0.5, 3)
@task(5) # 高频率
def view_products(self):
self.client.get("/api/products/1")
@task(3) # 中等频率
def add_to_cart(self):
self.client.post("/api/cart/add", json={
"product_id": 123,
"quantity": 1
})
@task(1) # 低频率
def checkout(self):
self.client.post("/api/orders", json={
"items": [{"product_id": 123, "quantity": 1}],
"payment_method": "credit_card"
})
性能测试报告生成
生成HTML报告 (Locust)
locust -f locustfile.py --host=http://example.com --csv=mytest --html=report.html
自定义报告函数
def generate_report(results, output_file='performance_report.txt'):
with open(output_file, 'w') as f:
f.write("=" * 50 + "\n")
f.write("性能测试报告\n")
f.write("=" * 50 + "\n\n")
# 基本统计
f.write(f"测试时间: {results['start_time']} - {results['end_time']}\n")
f.write(f"总请求数: {results['total_requests']}\n")
f.write(f"并发用户数: {results['concurrency']}\n\n")
# 响应时间分布
f.write("响应时间分布:\n")
for percentile in [50, 75, 90, 95, 99]:
time = results['percentiles'].get(percentile, 0)
f.write(f" P{percentile}: {time:.2f}ms\n")
f.write(f"\n最大响应时间: {results['max_time']:.2f}ms\n")
f.write(f"最小响应时间: {results['min_time']:.2f}ms\n")
f.write(f"平均响应时间: {results['avg_time']:.2f}ms\n")
f.write(f"吞吐量(QPS): {results['qps']:.2f}\n")
# 错误统计
f.write(f"\n错误统计:\n")
f.write(f" 总错误数: {results['error_count']}\n")
f.write(f" 错误率: {results['error_rate']:.2%}\n")
if results['errors']:
f.write("\n错误详情:\n")
for error_type, count in results['errors'].items():
f.write(f" {error_type}: {count}\n")
print(f"报告已生成: {output_file}")
最佳实践建议
压测前准备
-
设置合理的参数
- 从低并发开始,逐步增加
- 设置超时时间,避免请求挂起
-
数据准备
# 准备测试数据 test_data = { "users": [{"id": i, "name": f"user_{i}"} for i in range(100)], "tokens": ["token_" + str(i) for i in range(50)] } -
监控系统状态
import psutil import time
def monitor_system(): """监控系统资源使用情况""" while True: cpu_percent = psutil.cpu_percent(interval=1) memory = psutil.virtual_memory() print(f"CPU: {cpu_percent}%, Memory: {memory.percent}%") time.sleep(5)
### 常见问题处理
- **连接池耗尽**: 增加连接池大小或减少并发数
- **内存泄漏**: 监控内存使用，添加内存限制
- **超时**: 设置合理的超时时间，区分网络超时和业务超时
选择适合你项目的工具和方法，建议从简单的脚本开始，逐步完善压测方案。
