性能测试
概述 性能测试衡量系统在各种负载条件下的行为，包括响应时间、吞吐量、资源利用率和可扩展性。它有助于识别瓶颈、验证性能要求，并确保系统能够处理预期的负载。
何时使用
- 验证响应时间要求
- 测量API吞吐量和延迟
- 测试数据库查询性能
- 识别性能瓶颈
- 比较算法效率
- 优化前后的基准测试
- 验证缓存效果
- 测试并发用户容量
性能测试类型
- 负载测试：正常预期负载
- 压力测试：超出正常容量（见压力测试技能）
- 峰值测试：突然的负载增加
- 耐久性测试：持续负载超过时间
- 可扩展性测试：不同规模下的性能
- 基准测试：组件级性能
指令
1. k6用于API负载测试
// load-test.js
import http from 'k6/http';
import { check, sleep } from 'k6';
import { Rate, Trend } from 'k6/metrics';
// 自定义指标
const errorRate = new Rate('errors');
const orderDuration = new Trend('order_duration');
// 测试配置
export const options = {
stages: [
{ duration: '2m', target: 10 }, // 逐渐增加到10个用户
{ duration: '5m', target: 10 }, // 保持10个用户
{ duration: '2m', target: 50 }, // 逐渐增加到50个用户
{ duration: '5m', target: 50 }, // 保持50个用户
{ duration: '2m', target: 0 }, // 逐渐减少到0
],
thresholds: {
http_req_duration: ['p(95)<500'], // 95%的请求在500ms以下
http_req_failed: ['rate<0.01'], // 错误率低于1%
errors: ['rate<0.1'], // 自定义错误率低于10%
},
};
// 测试数据
const BASE_URL = 'https://api.example.com';
let authToken;
export function setup() {
// 登录一次并获取授权令牌
const loginRes = http.post(`${BASE_URL}/auth/login`, {
email: 'test@example.com',
password: 'password123',
});
return { token: loginRes.json('token') };
}
export default function (data) {
// 测试1：获取产品（读操作繁重）
const productsRes = http.get(`${BASE_URL}/products`, {
headers: { Authorization: `Bearer ${data.token}` },
});
check(productsRes, {
'products status is 200': (r) => r.status === 200,
'products response time < 200ms': (r) => r.timings.duration < 200,
'has products array': (r) => Array.isArray(r.json('products')),
}) || errorRate.add(1);
sleep(1);
// 测试2：创建订单（写操作繁重）
const orderPayload = JSON.stringify({
userId: 'user-123',
items: [
{ productId: 'prod-1', quantity: 2 },
{ productId: 'prod-2', quantity: 1 },
],
});
const orderRes = http.post(`${BASE_URL}/orders`, orderPayload, {
headers: {
'Content-Type': 'application/json',
Authorization: `Bearer ${data.token}`,
},
});
const orderSuccess = check(orderRes, {
'order status is 201': (r) => r.status === 201,
'order has id': (r) => r.json('id') !== undefined,
});
if (!orderSuccess) {
errorRate.add(1);
}
orderDuration.add(orderRes.timings.duration);
sleep(2);
// 测试3：获取订单详情
if (orderSuccess) {
const orderId = orderRes.json('id');
const orderDetailRes = http.get(`${BASE_URL}/orders/${orderId}`, {
headers: { Authorization: `Bearer ${data.token}` },
});
check(orderDetailRes, {
'order detail status is 200': (r) => r.status === 200,
}) || errorRate.add(1);
sleep(1);
}
export function teardown(data) {
// 如有需要则清理
}
# 运行k6测试
k6 run load-test.js
# 运行不同场景
k6 run --vus 100 --duration 30s load-test.js
# 输出到InfluxDB以进行可视化
k6 run --out influxdb=http://localhost:8086/k6 load-test.js
2. Apache JMeter
<!-- test-plan.jmx（简化表示） -->
<?xml version="1.0" encoding="UTF-8"?>
<jmeterTestPlan version="1.2">
<hashTree>
<TestPlan testname="API Performance Test">
<ThreadGroup testname="Users">
<elementProp name="ThreadGroup.main_controller">
<stringProp name="ThreadGroup.num_threads">50</stringProp>
<stringProp name="ThreadGroup.ramp_time">60</stringProp>
<stringProp name="ThreadGroup.duration">300</stringProp>
</elementProp>
<!-- HTTP Request: Get Products -->
<HTTPSamplerProxy testname="GET /products">
<stringProp name="HTTPSampler.domain">api.example.com</stringProp>
<stringProp name="HTTPSampler.path">/products</stringProp>
<stringProp name="HTTPSampler.method">GET</stringProp>
</HTTPSamplerProxy>
<!-- Assertions -->
<ResponseAssertion testname="Response Code 200">
<stringProp name="Assertion.test_field">Assertion.response_code</stringProp>
<stringProp name="Assertion.test_type">8</stringProp>
<stringProp name="Assertion.test_string">200</stringProp>
</ResponseAssertion>
<DurationAssertion testname="Response Time < 500ms">
<stringProp name="DurationAssertion.duration">500</stringProp>
</DurationAssertion>
<!-- Timers -->
<ConstantTimer testname="Think Time">
<stringProp name="ConstantTimer.delay">2000</stringProp>
</ConstantTimer>
</ThreadGroup>
</TestPlan>
</hashTree>
</jmeterTestPlan>
# 运行JMeter测试
jmeter -n -t test-plan.jmx -l results.jtl -e -o report/
# 从结果生成报告
jmeter -g results.jtl -o report/
3. pytest-benchmark用于Python
# test_performance.py
import pytest
from app.services import DataProcessor, SearchEngine
from app.models import User, Product
class TestDataProcessorPerformance:
@pytest.fixture
def large_dataset(self):
"""为测试创建大型数据集。"""
return [{'id': i, 'value': i * 2} for i in range(10000)]
def test_process_data_performance(self, benchmark, large_dataset):
"""基准测试数据处理。"""
processor = DataProcessor()
result = benchmark(processor.process, large_dataset)
assert len(result) == len(large_dataset)
# 基准测试将报告执行时间
def test_filter_data_performance(self, benchmark, large_dataset):
"""测试不同条件下的过滤性能。"""
processor = DataProcessor()
def filter_operation():
return processor.filter(large_dataset, lambda x: x['value'] > 5000)
result = benchmark(filter_operation)
assert all(item['value'] > 5000 for item in result)
@pytest.mark.parametrize('dataset_size', [100, 1000, 10000])
def test_scalability(self, benchmark, dataset_size):
"""在不同规模下测试性能。"""
processor = DataProcessor()
data = [{'id': i, 'value': i} for i in range(dataset_size)]
benchmark(processor.process, data)
class TestSearchPerformance:
@pytest.fixture
def search_engine(self):
"""用索引数据设置搜索引擎。"""
engine = SearchEngine()
# 索引10,000个产品
for i in range(10000):
engine.index(Product(id=i, name=f"Product {i}"))
return engine
def test_search_performance(self, benchmark, search_engine):
"""基准测试搜索查询。"""
result = benchmark(search_engine.search, "Product 5000")
assert len(result) > 0
def test_search_with_filters(self, benchmark, search_engine):
"""基准测试带有过滤器的搜索。"""
def search_with_filters():
return search_engine.search(
"Product",
filters={'price_min': 10, 'price_max': 100}
)
result = benchmark(search_with_filters)
# 运行基准测试
# pytest test_performance.py --benchmark-only
# pytest test_performance.py --benchmark-compare
4. JMH用于Java基准测试
// PerformanceBenchmark.java
import org.openjdk.jmh.annotations.*;
import java.util.*;
import java.util.concurrent.TimeUnit;
@BenchmarkMode(Mode.AverageTime)
@OutputTimeUnit(TimeUnit.MICROSECONDS)
@State(Scope.Benchmark)
@Fork(value = 1, warmups = 1)
@Warmup(iterations = 3, time = 1)
@Measurement(iterations = 5, time = 1)
public class CollectionPerformanceBenchmark {
@Param({"100", "1000", "10000"})
private int size;
private List<Integer> arrayList;
private List<Integer> linkedList;
private Set<Integer> hashSet;
@Setup
public void setup() {
arrayList = new ArrayList<>();
linkedList = new LinkedList<>();
hashSet = new HashSet<>();
for (int i = 0; i < size; i++) {
arrayList.add(i);
linkedList.add(i);
hashSet.add(i);
}
}
@Benchmark
public void arrayListIteration() {
int sum = 0;
for (Integer num : arrayList) {
sum += num;
}
}
@Benchmark
public void linkedListIteration() {
int sum = 0;
for (Integer num : linkedList) {
sum += num;
}
}
@Benchmark
public void arrayListRandomAccess() {
for (int i = 0; i < size; i++) {
arrayList.get(i);
}
}
@Benchmark
public void linkedListRandomAccess() {
for (int i = 0; i < size; i++) {
linkedList.get(i);
}
}
@Benchmark
public boolean hashSetContains() {
return hashSet.contains(size / 2);
}
public static void main(String[] args) throws Exception {
org.openjdk.jmh.Main.main(args);
}
}
// 运行：mvn clean install
// java -jar target/benchmarks.jar
5. 数据库查询性能
# test_database_performance.py
import pytest
import time
from sqlalchemy import create_engine
from app.models import User, Order
class TestDatabasePerformance:
@pytest.fixture
def db_session(self):
"""创建测试数据库会话。"""
engine = create_engine('postgresql://localhost/testdb')
Session = sessionmaker(bind=engine)
return Session()
def test_query_without_index(self, db_session, benchmark):
"""在没有索引的情况下测量查询性能。"""
def query():
return db_session.query(User).filter(
User.email == 'test@example.com'
).first()
result = benchmark(query)
def test_query_with_index(self, db_session, benchmark):
"""在有索引的情况下测量查询性能。"""
# 假设email列已建立索引
def query():
return db_session.query(User).filter(
User.email == 'test@example.com'
).first()
result = benchmark(query)
def test_n_plus_one_query(self, db_session):
"""识别N+1查询问题。"""
start = time.time()
# ❌ N+1查询
users = db_session.query(User).all()
for user in users:
orders = user.orders # 为每个用户触发单独的查询
n_plus_one_time = time.time() - start
# ✅ 预加载
start = time.time()
users = db_session.query(User).options(
joinedload(User.orders)
).all()
eager_time = time.time() - start
assert eager_time < n_plus_one_time, "预加载应该更快"
print(f"N+1: {n_plus_one_time:.3f}s, Eager: {eager_time:.3f}s")
def test_pagination_performance(self, db_session, benchmark):
"""测试分页效率。"""
def paginate():
return db_session.query(Product)\
.order_by(Product.id)\
.limit(50)\
.offset(1000)\
.all()
results = benchmark(paginate)
assert len(results) == 50
6. 实时监控
// performance-monitor.js
class PerformanceMonitor {
constructor() {
this.metrics = [];
}
async measureEndpoint(name, fn) {
const start = performance.now();
const startMemory = process.memoryUsage();
try {
const result = await fn();
const duration = performance.now() - start;
const endMemory = process.memoryUsage();
this.metrics.push({
name,
duration,
memoryDelta: {
heapUsed: endMemory.heapUsed - startMemory.heapUsed,
external: endMemory.external - startMemory.external,
},
timestamp: new Date(),
});
return result;
} catch (error) {
throw error;
}
}
getStats(name) {
const measurements = this.metrics.filter(m => m.name === name);
const durations = measurements.map(m => m.duration);
return {
count: measurements.length,
mean: durations.reduce((a, b) => a + b, 0) / durations.length,
min: Math.min(...durations),
max: Math.max(...durations),
p50: this.percentile(durations, 50),
p95: this.percentile(durations, 95),
p99: this.percentile(durations, 99),
};
}
percentile(values, p) {
const sorted = values.sort((a, b) => a - b);
const index = Math.ceil((p / 100) * sorted.length) - 1;
return sorted[index];
}
}
// 在测试中的使用
const monitor = new PerformanceMonitor();
test('API endpoint performance', async () => {
for (let i = 0; i < 100; i++) {
await monitor.measureEndpoint('getUsers', async () => {
return await fetch('/api/users').then(r => r.json());
});
}
const stats = monitor.getStats('getUsers');
expect(stats.p95).toBeLessThan(500); // 95百分位低于500ms
expect(stats.mean).toBeLessThan(200); // 平均值低于200ms
});
性能指标
响应时间
- 平均值：平均响应时间
- 中位数（P50）：第50百分位
- P95：第95百分位（SLA目标）
- P99：第99百分位（最差情况）
- 最大值：最慢响应
吞吐量
- 每秒请求数：处理的请求数量
- 每秒事务数：完成的事务
- 数据传输量：MB/s或GB/s
资源利用率
- CPU：使用百分比
- 内存：堆，RSS，外部
- 网络：带宽使用
- 磁盘I/O：读写操作
最佳实践
✅ 做
- 明确定义性能要求（SLAs）
- 使用真实数据量进行测试
- 监控资源利用率
- 测试缓存效果
- 使用百分位数（P95，P99）而不是平均值
- 测量前预热
- 在类似生产的环境中运行测试
- 识别并修复N+1查询问题
❌ 不做
- 仅使用小数据集进行测试
- 忽略内存泄漏
- 在不切实际的环境中测试
- 只关注平均响应时间
- 跳过数据库索引分析
- 仅测试快乐路径
- 忽略网络延迟
- 无统计意义的比较
工具
- 负载测试：k6, JMeter, Gatling, Locust, Artillery
- 基准测试：JMH（Java）, pytest-benchmark（Python）, Benchmark.js（Node.js）
- 分析：Chrome DevTools, py-spy, VisualVM
- 监控：Prometheus, Grafana, New Relic, Datadog
- 数据库：EXPLAIN ANALYZE, pg_stat_statements
示例
另见：压力测试，持续测试，代码度量分析，进行全面的性能分析。