性能测试与并发优化
大约 17 分钟
性能测试与并发优化
前言：让测试"飞"起来
还记得我第一次运行大规模自动化测试时的场景，1000个测试用例跑了整整一个晚上，第二天来公司发现还在跑...那种心情就像等快递一样焦急。后来我意识到，好的性能优化就像给汽车装上涡轮增压器，不仅跑得快，还很稳定。
今天我们就来探讨这个pytest框架中的性能测试和并发优化实现，看看如何让测试框架在大规模场景下依然表现出色，让测试真正"飞"起来。
性能测试设计理念
性能测试的层次
"""
性能测试的三个层次
就像体检一样，从基础指标到深度分析
"""
performance_levels = {
"基础性能": {
"指标": ["响应时间", "吞吐量", "成功率"],
"目标": "确保基本性能指标达标",
"场景": "单接口性能验证"
},
"压力测试": {
"指标": ["并发用户数", "系统资源使用率", "错误率"],
"目标": "找到系统性能瓶颈",
"场景": "模拟高并发访问"
},
"稳定性测试": {
"指标": ["长时间运行稳定性", "内存泄漏", "性能衰减"],
"目标": "验证系统长期运行稳定性",
"场景": "7x24小时持续运行"
}
}1. 性能测试模块设计
# src/utils/performance.py
"""
性能测试模块 - 让测试跑得更快更稳
"""
import time
import asyncio
import threading
import statistics
from concurrent.futures import ThreadPoolExecutor, as_completed
from dataclasses import dataclass
from typing import List, Dict, Any, Callable, Optional
import psutil
import json
from datetime import datetime
@dataclass
class PerformanceMetrics:
"""性能指标数据类"""
response_times: List[float]
success_count: int
error_count: int
start_time: float
end_time: float
concurrent_users: int
@property
def total_requests(self) -> int:
return self.success_count + self.error_count
@property
def success_rate(self) -> float:
if self.total_requests == 0:
return 0.0
return self.success_count / self.total_requests * 100
@property
def total_duration(self) -> float:
return self.end_time - self.start_time
@property
def throughput(self) -> float:
"""吞吐量 (请求/秒)"""
if self.total_duration == 0:
return 0.0
return self.total_requests / self.total_duration
@property
def avg_response_time(self) -> float:
"""平均响应时间"""
if not self.response_times:
return 0.0
return statistics.mean(self.response_times)
@property
def min_response_time(self) -> float:
"""最小响应时间"""
if not self.response_times:
return 0.0
return min(self.response_times)
@property
def max_response_time(self) -> float:
"""最大响应时间"""
if not self.response_times:
return 0.0
return max(self.response_times)
@property
def percentile_95(self) -> float:
"""95%分位数"""
if not self.response_times:
return 0.0
return statistics.quantiles(self.response_times, n=20)[18] # 95th percentile
@property
def percentile_99(self) -> float:
"""99%分位数"""
if not self.response_times:
return 0.0
return statistics.quantiles(self.response_times, n=100)[98] # 99th percentile
class PerformanceTester:
"""性能测试器 - 测试性能的专家"""
def __init__(self):
self.results: List[PerformanceMetrics] = []
self.system_monitor = SystemMonitor()
def run_load_test(self,
test_func: Callable,
concurrent_users: int = 10,
duration: int = 60,
ramp_up_time: int = 10) -> PerformanceMetrics:
"""
运行负载测试
Args:
test_func: 测试函数
concurrent_users: 并发用户数
duration: 测试持续时间(秒)
ramp_up_time: 启动时间(秒)
"""
print(f"🚀 开始负载测试: {concurrent_users}并发用户, 持续{duration}秒")
response_times = []
success_count = 0
error_count = 0
start_time = time.time()
end_time = start_time + duration
# 启动系统监控
self.system_monitor.start_monitoring()
with ThreadPoolExecutor(max_workers=concurrent_users) as executor:
# 提交初始任务
futures = []
for _ in range(concurrent_users):
future = executor.submit(self._worker_thread, test_func, end_time)
futures.append(future)
# 渐进式启动
if ramp_up_time > 0:
time.sleep(ramp_up_time / concurrent_users)
# 收集结果
for future in as_completed(futures):
try:
worker_results = future.result()
response_times.extend(worker_results['response_times'])
success_count += worker_results['success_count']
error_count += worker_results['error_count']
except Exception as e:
print(f"❌ 工作线程异常: {e}")
error_count += 1
# 停止系统监控
system_metrics = self.system_monitor.stop_monitoring()
actual_end_time = time.time()
metrics = PerformanceMetrics(
response_times=response_times,
success_count=success_count,
error_count=error_count,
start_time=start_time,
end_time=actual_end_time,
concurrent_users=concurrent_users
)
self.results.append(metrics)
self._print_results(metrics, system_metrics)
return metrics
def _worker_thread(self, test_func: Callable, end_time: float) -> Dict[str, Any]:
"""工作线程函数"""
response_times = []
success_count = 0
error_count = 0
while time.time() < end_time:
try:
start = time.time()
test_func()
end = time.time()
response_times.append(end - start)
success_count += 1
except Exception as e:
error_count += 1
print(f"⚠️ 请求失败: {e}")
# 短暂休息，避免过度消耗CPU
time.sleep(0.01)
return {
'response_times': response_times,
'success_count': success_count,
'error_count': error_count
}
def run_spike_test(self,
test_func: Callable,
base_users: int = 10,
spike_users: int = 100,
spike_duration: int = 30) -> PerformanceMetrics:
"""
运行尖峰测试
Args:
test_func: 测试函数
base_users: 基础用户数
spike_users: 尖峰用户数
spike_duration: 尖峰持续时间
"""
print(f"⚡ 开始尖峰测试: {base_users} → {spike_users}用户, 尖峰持续{spike_duration}秒")
# 先运行基础负载
print("📊 运行基础负载...")
base_metrics = self.run_load_test(test_func, base_users, 60)
# 然后运行尖峰负载
print("⚡ 运行尖峰负载...")
spike_metrics = self.run_load_test(test_func, spike_users, spike_duration)
# 分析尖峰影响
self._analyze_spike_impact(base_metrics, spike_metrics)
return spike_metrics
def _analyze_spike_impact(self, base_metrics: PerformanceMetrics, spike_metrics: PerformanceMetrics):
"""分析尖峰测试影响"""
print("\n📈 尖峰测试影响分析:")
response_time_increase = (spike_metrics.avg_response_time - base_metrics.avg_response_time) / base_metrics.avg_response_time * 100
success_rate_decrease = base_metrics.success_rate - spike_metrics.success_rate
print(f" 响应时间变化: {response_time_increase:+.1f}%")
print(f" 成功率变化: {success_rate_decrease:+.1f}%")
if response_time_increase > 50:
print(" ⚠️ 响应时间显著增加，系统可能存在性能瓶颈")
if success_rate_decrease > 5:
print(" ⚠️ 成功率显著下降，系统可能无法处理尖峰负载")
def _print_results(self, metrics: PerformanceMetrics, system_metrics: Dict[str, Any]):
"""打印测试结果"""
print("\n📊 性能测试结果:")
print(f" 总请求数: {metrics.total_requests}")
print(f" 成功请求: {metrics.success_count}")
print(f" 失败请求: {metrics.error_count}")
print(f" 成功率: {metrics.success_rate:.2f}%")
print(f" 吞吐量: {metrics.throughput:.2f} 请求/秒")
print(f" 平均响应时间: {metrics.avg_response_time:.3f}秒")并发优化策略
1. pytest并发执行优化
# conftest.py - pytest并发配置
"""
pytest并发执行配置
让测试用例并行执行，提升整体效率
"""
import pytest
import os
from concurrent.futures import ThreadPoolExecutor
def pytest_configure(config):
"""pytest配置钩子"""
# 设置并发执行参数
if not config.getoption("--dist"):
# 自动检测CPU核心数
cpu_count = os.cpu_count()
workers = min(cpu_count, 8) # 最多8个worker
config.option.dist = "worksteal"
config.option.numprocesses = workers
print(f"🚀 自动配置并发执行: {workers} workers")
@pytest.fixture(scope="session")
def thread_pool():
"""线程池fixture"""
max_workers = min(os.cpu_count() * 2, 20)
with ThreadPoolExecutor(max_workers=max_workers) as executor:
yield executor
# pytest-xdist配置示例
"""
# pytest.ini
[tool:pytest]
addopts =
-n auto
--dist worksteal
--maxfail=5
--tb=short
-v
markers =
slow: 标记慢速测试
fast: 标记快速测试
parallel: 可并行执行的测试
serial: 必须串行执行的测试
"""
class ParallelTestRunner:
"""并行测试运行器"""
def __init__(self, max_workers: int = None):
self.max_workers = max_workers or min(os.cpu_count() * 2, 20)
self.executor = ThreadPoolExecutor(max_workers=self.max_workers)
def run_tests_parallel(self, test_functions: List[Callable]) -> List[Any]:
"""并行运行测试函数"""
print(f"🔄 并行执行 {len(test_functions)} 个测试，使用 {self.max_workers} 个线程")
futures = []
for test_func in test_functions:
future = self.executor.submit(test_func)
futures.append(future)
results = []
for i, future in enumerate(futures):
try:
result = future.result(timeout=300) # 5分钟超时
results.append(result)
print(f"✅ 测试 {i+1}/{len(test_functions)} 完成")
except Exception as e:
print(f"❌ 测试 {i+1}/{len(test_functions)} 失败: {e}")
results.append(None)
return results
def __enter__(self):
return self
def __exit__(self, exc_type, exc_val, exc_tb):
self.executor.shutdown(wait=True)
# 使用示例
def test_parallel_execution():
"""并行执行测试示例"""
def create_test_function(test_id):
def test_func():
# 模拟测试逻辑
time.sleep(1)
return f"测试{test_id}完成"
return test_func
# 创建多个测试函数
test_functions = [create_test_function(i) for i in range(10)]
# 并行执行
with ParallelTestRunner(max_workers=5) as runner:
results = runner.run_tests_parallel(test_functions)
# 验证结果
assert len(results) == 10
assert all(result is not None for result in results)2. 连接池优化
# src/client/optimized_client.py
"""
优化的HTTP客户端 - 连接池和会话管理
"""
import requests
from requests.adapters import HTTPAdapter
from requests.packages.urllib3.util.retry import Retry
from requests.packages.urllib3.poolmanager import PoolManager
import threading
from typing import Dict, Optional
class OptimizedHTTPClient:
"""优化的HTTP客户端"""
_instances: Dict[str, 'OptimizedHTTPClient'] = {}
_lock = threading.Lock()
def __new__(cls, base_url: str, **kwargs):
"""单例模式，每个base_url一个实例"""
with cls._lock:
if base_url not in cls._instances:
instance = super().__new__(cls)
cls._instances[base_url] = instance
return cls._instances[base_url]
def __init__(self, base_url: str,
pool_connections: int = 20,
pool_maxsize: int = 50,
max_retries: int = 3):
if hasattr(self, '_initialized'):
return
self.base_url = base_url.rstrip('/')
self.session = requests.Session()
# 配置连接池
self._setup_connection_pool(pool_connections, pool_maxsize, max_retries)
# 配置默认头部
self._setup_default_headers()
self._initialized = True
print(f"🔗 优化HTTP客户端初始化: {base_url}")
def _setup_connection_pool(self, pool_connections: int, pool_maxsize: int, max_retries: int):
"""配置连接池"""
# 重试策略
retry_strategy = Retry(
total=max_retries,
backoff_factor=0.3,
status_forcelist=[429, 500, 502, 503, 504],
method_whitelist=["HEAD", "GET", "PUT", "DELETE", "OPTIONS", "TRACE"]
)
# HTTP适配器
http_adapter = HTTPAdapter(
pool_connections=pool_connections,
pool_maxsize=pool_maxsize,
max_retries=retry_strategy,
pool_block=False
)
# HTTPS适配器
https_adapter = HTTPAdapter(
pool_connections=pool_connections,
pool_maxsize=pool_maxsize,
max_retries=retry_strategy,
pool_block=False
)
# 挂载适配器
self.session.mount('http://', http_adapter)
self.session.mount('https://', https_adapter)
print(f"🏊 连接池配置: {pool_connections}连接, {pool_maxsize}最大池大小")
def _setup_default_headers(self):
"""设置默认头部"""
self.session.headers.update({
'User-Agent': 'OptimizedTestClient/1.0',
'Accept': 'application/json',
'Accept-Encoding': 'gzip, deflate',
'Connection': 'keep-alive'
})
def get_pool_status(self) -> Dict[str, Any]:
"""获取连接池状态"""
pool_info = {}
for prefix, adapter in self.session.adapters.items():
if hasattr(adapter, 'poolmanager'):
pool_manager = adapter.poolmanager
if isinstance(pool_manager, PoolManager):
pool_info[prefix] = {
'num_pools': len(pool_manager.pools),
'pools': {}
}
for key, pool in pool_manager.pools.items():
pool_info[prefix]['pools'][str(key)] = {
'num_connections': pool.pool.qsize() if hasattr(pool.pool, 'qsize') else 'unknown',
'maxsize': getattr(pool, 'maxsize', 'unknown')
}
return pool_info
def request(self, method: str, endpoint: str, **kwargs) -> requests.Response:
"""发送请求"""
url = f"{self.base_url}/{endpoint.lstrip('/')}"
return self.session.request(method, url, **kwargs)
def close(self):
"""关闭会话"""
self.session.close()
@classmethod
def close_all(cls):
"""关闭所有实例"""
with cls._lock:
for instance in cls._instances.values():
instance.close()
cls._instances.clear()
# 连接池监控
class ConnectionPoolMonitor:
"""连接池监控器"""
def __init__(self, client: OptimizedHTTPClient):
self.client = client
self.monitoring = False
self.monitor_thread = None
self.stats_history = []
def start_monitoring(self, interval: int = 5):
"""开始监控"""
self.monitoring = True
self.monitor_thread = threading.Thread(
target=self._monitor_loop,
args=(interval,)
)
self.monitor_thread.daemon = True
self.monitor_thread.start()
print(f"📊 连接池监控已启动，间隔{interval}秒")
def stop_monitoring(self):
"""停止监控"""
self.monitoring = False
if self.monitor_thread:
self.monitor_thread.join(timeout=10)
print("📊 连接池监控已停止")
def _monitor_loop(self, interval: int):
"""监控循环"""
while self.monitoring:
try:
stats = self.client.get_pool_status()
stats['timestamp'] = time.time()
self.stats_history.append(stats)
# 保留最近100条记录
if len(self.stats_history) > 100:
self.stats_history.pop(0)
except Exception as e:
print(f"⚠️ 连接池监控异常: {e}")
time.sleep(interval)
def get_stats_summary(self) -> Dict[str, Any]:
"""获取统计摘要"""
if not self.stats_history:
return {}
latest_stats = self.stats_history[-1]
return {
'latest_stats': latest_stats,
'history_count': len(self.stats_history),
'monitoring_duration': latest_stats['timestamp'] - self.stats_history[0]['timestamp'] if len(self.stats_history) > 1 else 0
}3. 异步HTTP客户端
# src/client/async_client.py
"""
异步HTTP客户端 - 高并发场景的利器
"""
import asyncio
import aiohttp
import time
from typing import Dict, Any, Optional, List
from dataclasses import dataclass
@dataclass
class AsyncResponse:
"""异步响应包装"""
status: int
headers: Dict[str, str]
text: str
json_data: Optional[Dict[str, Any]]
response_time: float
def json(self) -> Dict[str, Any]:
"""获取JSON数据"""
return self.json_data
class AsyncHTTPClient:
"""异步HTTP客户端"""
def __init__(self, base_url: str,
connector_limit: int = 100,
timeout: int = 30):
self.base_url = base_url.rstrip('/')
self.connector_limit = connector_limit
self.timeout = aiohttp.ClientTimeout(total=timeout)
self.session: Optional[aiohttp.ClientSession] = None
async def __aenter__(self):
"""异步上下文管理器入口"""
connector = aiohttp.TCPConnector(
limit=self.connector_limit,
limit_per_host=self.connector_limit // 2,
keepalive_timeout=30,
enable_cleanup_closed=True
)
self.session = aiohttp.ClientSession(
connector=connector,
timeout=self.timeout,
headers={
'User-Agent': 'AsyncTestClient/1.0',
'Accept': 'application/json'
}
)
return self
async def __aexit__(self, exc_type, exc_val, exc_tb):
"""异步上下文管理器出口"""
if self.session:
await self.session.close()
async def request(self, method: str, endpoint: str, **kwargs) -> AsyncResponse:
"""发送异步请求"""
if not self.session:
raise RuntimeError("客户端未初始化，请使用async with语句")
url = f"{self.base_url}/{endpoint.lstrip('/')}"
start_time = time.time()
try:
async with self.session.request(method, url, **kwargs) as response:
text = await response.text()
# 尝试解析JSON
json_data = None
if 'application/json' in response.headers.get('Content-Type', ''):
try:
json_data = await response.json()
except:
pass
response_time = time.time() - start_time
return AsyncResponse(
status=response.status,
headers=dict(response.headers),
text=text,
json_data=json_data,
response_time=response_time
)
except asyncio.TimeoutError:
raise TimeoutError(f"请求超时: {method} {url}")
except Exception as e:
raise RuntimeError(f"请求失败: {method} {url} - {e}")
async def get(self, endpoint: str, **kwargs) -> AsyncResponse:
"""GET请求"""
return await self.request('GET', endpoint, **kwargs)
async def post(self, endpoint: str, **kwargs) -> AsyncResponse:
"""POST请求"""
return await self.request('POST', endpoint, **kwargs)
async def put(self, endpoint: str, **kwargs) -> AsyncResponse:
"""PUT请求"""
return await self.request('PUT', endpoint, **kwargs)
async def delete(self, endpoint: str, **kwargs) -> AsyncResponse:
"""DELETE请求"""
return await self.request('DELETE', endpoint, **kwargs)
# 异步批量请求工具
class AsyncBatchRequester:
"""异步批量请求器"""
def __init__(self, base_url: str, max_concurrent: int = 50):
self.base_url = base_url
self.max_concurrent = max_concurrent
self.semaphore = asyncio.Semaphore(max_concurrent)
async def batch_requests(self, requests: List[Dict[str, Any]]) -> List[AsyncResponse]:
"""批量发送请求"""
async with AsyncHTTPClient(self.base_url) as client:
tasks = []
for request_config in requests:
task = self._limited_request(client, request_config)
tasks.append(task)
results = await asyncio.gather(*tasks, return_exceptions=True)
# 处理异常
responses = []
for i, result in enumerate(results):
if isinstance(result, Exception):
print(f"❌ 请求 {i+1} 失败: {result}")
responses.append(None)
else:
responses.append(result)
return responses
async def _limited_request(self, client: AsyncHTTPClient, request_config: Dict[str, Any]) -> AsyncResponse:
"""限制并发的请求"""
async with self.semaphore:
method = request_config.pop('method', 'GET')
endpoint = request_config.pop('endpoint', '/')
return await client.request(method, endpoint, **request_config)
# 使用示例
async def demo_async_performance():
"""异步性能测试示例"""
# 创建大量请求
requests = []
for i in range(1000):
requests.append({
'method': 'GET',
'endpoint': f'/api/users/{i}',
'params': {'page': i % 10 + 1}
})
# 批量发送
batch_requester = AsyncBatchRequester('https://httpbin.org', max_concurrent=50)
start_time = time.time()
responses = await batch_requester.batch_requests(requests)
end_time = time.time()
# 统计结果
successful_responses = [r for r in responses if r and r.status == 200]
print(f"📊 异步批量请求结果:")
print(f" 总请求数: {len(requests)}")
print(f" 成功请求: {len(successful_responses)}")
print(f" 总耗时: {end_time - start_time:.2f}秒")
print(f" 平均QPS: {len(requests) / (end_time - start_time):.2f}")
# 运行异步示例
if __name__ == "__main__":
asyncio.run(demo_async_performance())实战应用：性能测试实践
1. 完整的性能测试套件
# tests/test_performance_suite.py
"""
完整的性能测试套件
展示如何在实际项目中应用性能测试
"""
import pytest
import asyncio
from src.client.base_client import BaseClient
from src.client.async_client import AsyncHTTPClient, AsyncBatchRequester
from src.utils.performance import PerformanceTester, AsyncPerformanceTester
from src.utils.data_driver import data_provider
class TestPerformanceSuite:
"""性能测试套件"""
def setup_method(self):
self.base_url = "https://httpbin.org"
self.client = BaseClient(self.base_url)
self.perf_tester = PerformanceTester()
@pytest.mark.performance
def test_single_api_performance(self):
"""单接口性能测试"""
def test_get_request():
"""测试GET请求"""
response = self.client.get("/get")
assert response.status_code == 200
# 运行负载测试
metrics = self.perf_tester.run_load_test(
test_func=test_get_request,
concurrent_users=10,
duration=30
)
# 性能断言
assert metrics.success_rate >= 95.0, f"成功率过低: {metrics.success_rate}%"
assert metrics.avg_response_time <= 2.0, f"平均响应时间过长: {metrics.avg_response_time}s"
assert metrics.percentile_95 <= 5.0, f"95%分位数过高: {metrics.percentile_95}s"
@pytest.mark.performance
def test_post_api_performance(self):
"""POST接口性能测试"""
def test_post_request():
"""测试POST请求"""
test_data = data_provider.generator.generate_user_data()
response = self.client.post("/post", json=test_data)
assert response.status_code == 200
# 运行负载测试
metrics = self.perf_tester.run_load_test(
test_func=test_post_request,
concurrent_users=20,
duration=60
)
# 性能断言
assert metrics.success_rate >= 90.0
assert metrics.avg_response_time <= 3.0
assert metrics.throughput >= 5.0 # 至少5 QPS
@pytest.mark.performance
@pytest.mark.slow
def test_spike_performance(self):
"""尖峰性能测试"""
def test_api_call():
response = self.client.get("/get")
assert response.status_code == 200
# 运行尖峰测试
metrics = self.perf_tester.run_spike_test(
test_func=test_api_call,
base_users=5,
spike_users=50,
spike_duration=30
)
# 尖峰测试的容忍度更高
assert metrics.success_rate >= 80.0
assert metrics.avg_response_time <= 10.0
@pytest.mark.asyncio
@pytest.mark.performance
async def test_async_performance(self):
"""异步性能测试"""
async def async_test_func():
"""异步测试函数"""
async with AsyncHTTPClient(self.base_url) as client:
response = await client.get("/get")
assert response.status == 200
# 运行异步负载测试
async_tester = AsyncPerformanceTester()
metrics = await async_tester.run_async_load_test(
async_test_func=async_test_func,
concurrent_users=50,
duration=30
)
# 异步测试通常有更高的吞吐量
assert metrics.success_rate >= 95.0
assert metrics.throughput >= 20.0 # 异步应该有更高的QPS
@pytest.mark.performance
def test_batch_requests_performance(self):
"""批量请求性能测试"""
async def batch_test():
# 创建批量请求
requests = []
for i in range(100):
requests.append({
'method': 'GET',
'endpoint': '/get',
'params': {'id': i}
})
# 批量发送
batch_requester = AsyncBatchRequester(self.base_url, max_concurrent=20)
responses = await batch_requester.batch_requests(requests)
# 验证结果
successful_count = sum(1 for r in responses if r and r.status == 200)
assert successful_count >= 95 # 至少95%成功
# 运行批量测试
start_time = time.time()
asyncio.run(batch_test())
end_time = time.time()
duration = end_time - start_time
qps = 100 / duration
print(f"📊 批量请求性能: {qps:.2f} QPS, 耗时 {duration:.2f}s")
assert qps >= 10.0 # 至少10 QPS
class TestPerformanceRegression:
"""性能回归测试"""
def setup_method(self):
self.baseline_file = "performance_baseline.json"
self.perf_tester = PerformanceTester()
def load_baseline(self) -> Dict[str, float]:
"""加载性能基线"""
try:
with open(self.baseline_file, 'r') as f:
return json.load(f)
except FileNotFoundError:
return {}
def save_baseline(self, metrics: Dict[str, float]):
"""保存性能基线"""
with open(self.baseline_file, 'w') as f:
json.dump(metrics, f, indent=2)
@pytest.mark.performance
def test_performance_regression(self):
"""性能回归测试"""
def test_api():
response = BaseClient("https://httpbin.org").get("/get")
assert response.status_code == 200
# 运行性能测试
metrics = self.perf_tester.run_load_test(
test_func=test_api,
concurrent_users=10,
duration=30
)
current_metrics = {
'avg_response_time': metrics.avg_response_time,
'percentile_95': metrics.percentile_95,
'throughput': metrics.throughput,
'success_rate': metrics.success_rate
}
# 加载基线数据
baseline = self.load_baseline()
if not baseline:
# 首次运行，保存基线
self.save_baseline(current_metrics)
print("📊 性能基线已保存")
return
# 性能回归检查
regression_threshold = 0.2 # 20%的性能衰减阈值
for metric_name, current_value in current_metrics.items():
baseline_value = baseline.get(metric_name)
if baseline_value is None:
continue
if metric_name in ['avg_response_time', 'percentile_95']:
# 响应时间类指标，值越小越好
regression_ratio = (current_value - baseline_value) / baseline_value
if regression_ratio > regression_threshold:
pytest.fail(f"性能回归检测: {metric_name} 衰减 {regression_ratio:.1%}")
elif metric_name in ['throughput', 'success_rate']:
# 吞吐量和成功率，值越大越好
regression_ratio = (baseline_value - current_value) / baseline_value
if regression_ratio > regression_threshold:
pytest.fail(f"性能回归检测: {metric_name} 衰减 {regression_ratio:.1%}")
print("✅ 性能回归检测通过")
# pytest配置
@pytest.fixture(scope="session")
def performance_report():
"""性能测试报告fixture"""
report_data = {
'test_results': [],
'start_time': time.time()
}
yield report_data
# 生成性能测试报告
end_time = time.time()
report_data['end_time'] = end_time
report_data['total_duration'] = end_time - report_data['start_time']
generate_performance_report(report_data)
def generate_performance_report(report_data: Dict[str, Any]):
"""生成性能测试报告"""
report_content = f"""
# 性能测试报告
## 测试概览
- 测试开始时间: {datetime.fromtimestamp(report_data['start_time'])}
- 测试结束时间: {datetime.fromtimestamp(report_data['end_time'])}
- 总测试时长: {report_data['total_duration']:.2f}秒
- 测试用例数: {len(report_data['test_results'])}
## 测试结果
"""
for result in report_data['test_results']:
report_content += f"""
### {result['test_name']}
- 成功率: {result['success_rate']:.2f}%
- 平均响应时间: {result['avg_response_time']:.3f}s
- 吞吐量: {result['throughput']:.2f} QPS
- 95%分位数: {result['percentile_95']:.3f}s
"""
# 保存报告
with open('performance_report.md', 'w', encoding='utf-8') as f:
f.write(report_content)
print("📊 性能测试报告已生成: performance_report.md")性能优化最佳实践
1. 测试执行优化
"""
测试执行优化策略
"""
optimization_strategies = {
"并发控制": {
"策略": "根据系统资源动态调整并发数",
"实现": "监控CPU和内存使用率，自动调整worker数量",
"工具": "pytest-xdist, multiprocessing"
},
"连接复用": {
"策略": "使用连接池和会话复用",
"实现": "配置合适的连接池大小，启用keep-alive",
"工具": "requests.Session, aiohttp.ClientSession"
},
"数据预加载": {
"策略": "预先生成和缓存测试数据",
"实现": "使用session级别的fixture预加载数据",
"工具": "pytest fixtures, 内存缓存"
},
"智能重试": {
"策略": "区分临时性错误和永久性错误",
"实现": "只对网络超时等临时错误进行重试",
"工具": "urllib3.Retry, 自定义重试逻辑"
}
}2. 监控与告警
# src/utils/performance_monitor.py
"""
性能监控与告警系统
"""
import time
import threading
from typing import Dict, List, Callable
from dataclasses import dataclass, asdict
@dataclass
class PerformanceAlert:
"""性能告警"""
metric_name: str
current_value: float
threshold: float
severity: str # 'warning', 'critical'
timestamp: float
message: str
class PerformanceMonitor:
"""性能监控器"""
def __init__(self):
self.thresholds = {
'avg_response_time': {'warning': 2.0, 'critical': 5.0},
'success_rate': {'warning': 95.0, 'critical': 90.0},
'throughput': {'warning': 10.0, 'critical': 5.0}
}
self.alerts: List[PerformanceAlert] = []
self.alert_callbacks: List[Callable] = []
def add_alert_callback(self, callback: Callable[[PerformanceAlert], None]):
"""添加告警回调"""
self.alert_callbacks.append(callback)
def check_metrics(self, metrics: PerformanceMetrics):
"""检查性能指标"""
metric_values = {
'avg_response_time': metrics.avg_response_time,
'success_rate': metrics.success_rate,
'throughput': metrics.throughput
}
for metric_name, value in metric_values.items():
self._check_single_metric(metric_name, value)
def _check_single_metric(self, metric_name: str, value: float):
"""检查单个指标"""
thresholds = self.thresholds.get(metric_name, {})
for severity, threshold in thresholds.items():
if self._is_threshold_exceeded(metric_name, value, threshold, severity):
alert = PerformanceAlert(
metric_name=metric_name,
current_value=value,
threshold=threshold,
severity=severity,
timestamp=time.time(),
message=f"{metric_name} {severity}: {value} (阈值: {threshold})"
)
self.alerts.append(alert)
self._trigger_alert(alert)
def _is_threshold_exceeded(self, metric_name: str, value: float, threshold: float, severity: str) -> bool:
"""判断是否超过阈值"""
if metric_name in ['avg_response_time']:
# 响应时间类指标，值越大越差
return value > threshold
elif metric_name in ['success_rate', 'throughput']:
# 成功率和吞吐量，值越小越差
return value < threshold
return False
def _trigger_alert(self, alert: PerformanceAlert):
"""触发告警"""
print(f"🚨 性能告警: {alert.message}")
for callback in self.alert_callbacks:
try:
callback(alert)
except Exception as e:
print(f"⚠️ 告警回调执行失败: {e}")
def get_alert_summary(self) -> Dict[str, Any]:
"""获取告警摘要"""
if not self.alerts:
return {"total_alerts": 0}
warning_count = sum(1 for alert in self.alerts if alert.severity == 'warning')
critical_count = sum(1 for alert in self.alerts if alert.severity == 'critical')
return {
"total_alerts": len(self.alerts),
"warning_count": warning_count,
"critical_count": critical_count,
"latest_alert": asdict(self.alerts[-1]) if self.alerts else None
}
# 告警回调示例
def send_alert_notification(alert: PerformanceAlert):
"""发送告警通知"""
if alert.severity == 'critical':
# 发送紧急通知（邮件、短信、钉钉等）
print(f"📧 发送紧急通知: {alert.message}")
else:
# 发送普通通知
print(f"📝 记录告警: {alert.message}")
# 使用示例
def demo_performance_monitoring():
"""性能监控示例"""
monitor = PerformanceMonitor()
monitor.add_alert_callback(send_alert_notification)
# 模拟性能测试
perf_tester = PerformanceTester()
def test_api():
time.sleep(0.1) # 模拟API调用
metrics = perf_tester.run_load_test(test_api, concurrent_users=5, duration=10)
# 检查性能指标
monitor.check_metrics(metrics)
# 获取告警摘要
summary = monitor.get_alert_summary()
print(f"📊 告警摘要: {summary}")总结
性能测试与并发优化就像给测试框架装上了"涡轮增压器"，不仅让测试跑得更快，还能发现系统的性能瓶颈。通过这篇文章，我们深入了解了：
- 性能测试设计：负载测试、压力测试、稳定性测试的实现
- 并发优化策略：pytest并发、连接池、异步客户端的应用
- 实战应用：完整的性能测试套件和回归测试
- 监控告警：性能指标监控和自动告警机制
- 最佳实践：优化策略和经验总结
这套性能测试和并发优化方案不仅提升了测试执行效率，更重要的是为系统性能提供了全面的保障。它让我们能够在开发阶段就发现性能问题，避免在生产环境中出现性能瓶颈。
下一篇文章，我们将对整个项目进行总结，分享项目开发的经验和最佳实践。
推荐阅读
print(f" 最小响应时间: {metrics.min_response_time:.3f}秒")
print(f" 最大响应时间: {metrics.max_response_time:.3f}秒")
print(f" 95%分位数: {metrics.percentile_95:.3f}秒")
print(f" 99%分位数: {metrics.percentile_99:.3f}秒")
print("\n💻 系统资源使用:")
print(f" 平均CPU使用率: {system_metrics['avg_cpu']:.1f}%")
print(f" 平均内存使用率: {system_metrics['avg_memory']:.1f}%")
print(f" 峰值CPU使用率: {system_metrics['max_cpu']:.1f}%")
print(f" 峰值内存使用率: {system_metrics['max_memory']:.1f}%")
class SystemMonitor:
"""系统监控器"""
def __init__(self):
self.monitoring = False
self.cpu_samples = []
self.memory_samples = []
self.monitor_thread = None
def start_monitoring(self):
"""开始监控"""
self.monitoring = True
self.cpu_samples = []
self.memory_samples = []
self.monitor_thread = threading.Thread(target=self._monitor_loop)
self.monitor_thread.daemon = True
self.monitor_thread.start()
def stop_monitoring(self) -> Dict[str, float]:
"""停止监控并返回统计数据"""
self.monitoring = False
if self.monitor_thread:
self.monitor_thread.join(timeout=5)
if not self.cpu_samples or not self.memory_samples:
return {
'avg_cpu': 0.0,
'max_cpu': 0.0,
'avg_memory': 0.0,
'max_memory': 0.0
}
return {
'avg_cpu': statistics.mean(self.cpu_samples),
'max_cpu': max(self.cpu_samples),
'avg_memory': statistics.mean(self.memory_samples),
'max_memory': max(self.memory_samples)
}
def _monitor_loop(self):
"""监控循环"""
while self.monitoring:
try:
cpu_percent = psutil.cpu_percent(interval=1)
memory_percent = psutil.virtual_memory().percent
self.cpu_samples.append(cpu_percent)
self.memory_samples.append(memory_percent)
except Exception as e:
print(f"⚠️ 系统监控异常: {e}")
time.sleep(1)
class AsyncPerformanceTester:
"""异步性能测试器"""
def __init__(self):
self.results: List[PerformanceMetrics] = []
async def run_async_load_test(self,
async_test_func: Callable,
concurrent_users: int = 10,
duration: int = 60) -> PerformanceMetrics:
"""运行异步负载测试"""
print(f"🚀 开始异步负载测试: {concurrent_users}并发用户, 持续{duration}秒")
response_times = []
success_count = 0
error_count = 0
start_time = time.time()
end_time = start_time + duration
# 创建并发任务
tasks = []
for _ in range(concurrent_users):
task = asyncio.create_task(
self._async_worker(async_test_func, end_time)
)
tasks.append(task)
# 等待所有任务完成
results = await asyncio.gather(*tasks, return_exceptions=True)
# 收集结果
for result in results:
if isinstance(result, Exception):
error_count += 1
print(f"❌ 异步任务异常: {result}")
else:
response_times.extend(result['response_times'])
success_count += result['success_count']
error_count += result['error_count']
actual_end_time = time.time()
metrics = PerformanceMetrics(
response_times=response_times,
success_count=success_count,
error_count=error_count,
start_time=start_time,
end_time=actual_end_time,
concurrent_users=concurrent_users
)
self.results.append(metrics)
self._print_async_results(metrics)
return metrics
async def _async_worker(self, async_test_func: Callable, end_time: float) -> Dict[str, Any]:
"""异步工作函数"""
response_times = []
success_count = 0
error_count = 0
while time.time() < end_time:
try:
start = time.time()
await async_test_func()
end = time.time()
response_times.append(end - start)
success_count += 1
except Exception as e:
error_count += 1
print(f"⚠️ 异步请求失败: {e}")
# 短暂休息
await asyncio.sleep(0.01)
return {
'response_times': response_times,
'success_count': success_count,
'error_count': error_count
}
def _print_async_results(self, metrics: PerformanceMetrics):
"""打印异步测试结果"""
print("\n📊 异步性能测试结果:")
print(f" 总请求数: {metrics.total_requests}")
print(f" 成功率: {metrics.success_rate:.2f}%")
print(f" 吞吐量: {metrics.throughput:.2f} 请求/秒")
print(f" 平均响应时间: {metrics.avg_response_time:.3f}秒")