Locust性能优化与调优
大约 7 分钟
Locust性能优化与调优
🔧 压测跑起来了，但是效果不理想？别急，让我来教你如何调优！
就像汽车改装一样，我们要让Locust这台"压测跑车"跑得更快更稳。
今天分享一些实战中总结的性能优化技巧，让你的压测事半功倍！⚡
🎯 性能瓶颈识别
常见性能问题
作为测试工程师，我经常遇到这些问题：
- 客户端瓶颈：Locust本身成为性能瓶颈
- 网络瓶颈：带宽不足或延迟过高
- 脚本问题：编写不当导致效率低下
- 资源限制：CPU、内存、文件描述符不足
性能监控脚本
import psutil
import time
from locust import HttpUser, task, events
import threading
class PerformanceMonitor:
"""性能监控器"""
def __init__(self):
self.monitoring = False
self.stats = {
"cpu_usage": [],
"memory_usage": [],
"network_io": [],
"disk_io": [],
"connections": []
}
def start_monitoring(self):
"""开始监控"""
self.monitoring = True
monitor_thread = threading.Thread(target=self._monitor_loop)
monitor_thread.daemon = True
monitor_thread.start()
print("📊 性能监控已启动")
def stop_monitoring(self):
"""停止监控"""
self.monitoring = False
self._generate_report()
def _monitor_loop(self):
"""监控循环"""
while self.monitoring:
# CPU使用率
cpu_percent = psutil.cpu_percent(interval=1)
self.stats["cpu_usage"].append(cpu_percent)
# 内存使用率
memory = psutil.virtual_memory()
self.stats["memory_usage"].append(memory.percent)
# 网络IO
net_io = psutil.net_io_counters()
self.stats["network_io"].append({
"bytes_sent": net_io.bytes_sent,
"bytes_recv": net_io.bytes_recv,
"timestamp": time.time()
})
# 连接数
connections = len(psutil.net_connections())
self.stats["connections"].append(connections)
# 检查告警条件
self._check_alerts(cpu_percent, memory.percent, connections)
time.sleep(5)
def _check_alerts(self, cpu, memory, connections):
"""检查告警条件"""
if cpu > 80:
print(f"⚠️ CPU使用率过高: {cpu:.1f}%")
if memory > 80:
print(f"⚠️ 内存使用率过高: {memory:.1f}%")
if connections > 10000:
print(f"⚠️ 连接数过多: {connections}")
def _generate_report(self):
"""生成监控报告"""
if not self.stats["cpu_usage"]:
return
avg_cpu = sum(self.stats["cpu_usage"]) / len(self.stats["cpu_usage"])
max_cpu = max(self.stats["cpu_usage"])
avg_memory = sum(self.stats["memory_usage"]) / len(self.stats["memory_usage"])
max_connections = max(self.stats["connections"]) if self.stats["connections"] else 0
print("\n📋 性能监控报告:")
print(f" 平均CPU使用率: {avg_cpu:.1f}%")
print(f" 最高CPU使用率: {max_cpu:.1f}%")
print(f" 平均内存使用率: {avg_memory:.1f}%")
print(f" 最大连接数: {max_connections}")
# 全局监控实例
monitor = PerformanceMonitor()
@events.test_start.add_listener
def on_test_start(environment, **kwargs):
monitor.start_monitoring()
@events.test_stop.add_listener
def on_test_stop(environment, **kwargs):
monitor.stop_monitoring()⚡ 客户端性能优化
1. HTTP连接优化
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
from urllib3.poolmanager import PoolManager
import ssl
class OptimizedHTTPAdapter(HTTPAdapter):
"""优化的HTTP适配器"""
def __init__(self, *args, **kwargs):
# 连接池配置
self.pool_connections = kwargs.pop('pool_connections', 50)
self.pool_maxsize = kwargs.pop('pool_maxsize', 100)
self.pool_block = kwargs.pop('pool_block', False)
super().__init__(*args, **kwargs)
def init_poolmanager(self, *args, **kwargs):
"""初始化连接池管理器"""
kwargs['maxsize'] = self.pool_maxsize
kwargs['block'] = self.pool_block
# SSL优化
kwargs['ssl_context'] = ssl.create_default_context()
kwargs['ssl_context'].check_hostname = False
kwargs['ssl_context'].verify_mode = ssl.CERT_NONE
return super().init_poolmanager(*args, **kwargs)
class OptimizedUser(HttpUser):
wait_time = between(0.1, 0.5) # 减少等待时间
host = "https://httpbin.org"
def on_start(self):
"""优化HTTP客户端"""
# 配置重试策略
retry_strategy = Retry(
total=3,
backoff_factor=0.1,
status_forcelist=[429, 500, 502, 503, 504],
raise_on_status=False
)
# 使用优化的适配器
adapter = OptimizedHTTPAdapter(
pool_connections=50,
pool_maxsize=100,
max_retries=retry_strategy
)
self.client.mount("http://", adapter)
self.client.mount("https://", adapter)
# 设置超时
self.client.timeout = (2, 10) # 连接超时2秒，读取超时10秒
# 禁用SSL验证（测试环境）
self.client.verify = False
# 设置请求头
self.client.headers.update({
'User-Agent': 'Locust-Performance-Test',
'Connection': 'keep-alive',
'Accept-Encoding': 'gzip, deflate'
})
@task
def optimized_request(self):
"""优化的请求"""
self.client.get("/get")2. 异步请求优化
import asyncio
import aiohttp
from locust import User, task, events
import time
class AsyncUser(User):
"""异步用户类"""
wait_time = between(0.1, 0.3)
host = "https://httpbin.org"
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self.session = None
async def on_start(self):
"""异步初始化"""
# 创建异步HTTP会话
connector = aiohttp.TCPConnector(
limit=100, # 总连接池大小
limit_per_host=50, # 每个主机的连接数
ttl_dns_cache=300, # DNS缓存时间
use_dns_cache=True,
keepalive_timeout=30,
enable_cleanup_closed=True
)
timeout = aiohttp.ClientTimeout(total=10, connect=2)
self.session = aiohttp.ClientSession(
connector=connector,
timeout=timeout,
headers={
'User-Agent': 'Locust-Async-Test',
'Connection': 'keep-alive'
}
)
async def on_stop(self):
"""清理资源"""
if self.session:
await self.session.close()
@task
async def async_request(self):
"""异步请求"""
start_time = time.time()
try:
async with self.session.get(f"{self.host}/get") as response:
await response.text()
# 手动记录统计
total_time = int((time.time() - start_time) * 1000)
events.request_success.fire(
request_type="GET",
name="/get",
response_time=total_time,
response_length=len(await response.text())
)
except Exception as e:
total_time = int((time.time() - start_time) * 1000)
events.request_failure.fire(
request_type="GET",
name="/get",
response_time=total_time,
response_length=0,
exception=e
)🔧 系统级优化
1. 操作系统调优
#!/bin/bash
# system_tuning.sh - 系统调优脚本
echo "🔧 开始系统调优..."
# 增加文件描述符限制
echo "调整文件描述符限制..."
echo "* soft nofile 65536" >> /etc/security/limits.conf
echo "* hard nofile 65536" >> /etc/security/limits.conf
ulimit -n 65536
# 调整TCP参数
echo "调整TCP参数..."
sysctl -w net.core.somaxconn=65535
sysctl -w net.ipv4.tcp_max_syn_backlog=65535
sysctl -w net.core.netdev_max_backlog=5000
sysctl -w net.ipv4.tcp_fin_timeout=30
sysctl -w net.ipv4.tcp_keepalive_time=1200
sysctl -w net.ipv4.tcp_rmem="4096 65536 16777216"
sysctl -w net.ipv4.tcp_wmem="4096 65536 16777216"
# 调整内存参数
echo "调整内存参数..."
sysctl -w vm.swappiness=10
sysctl -w vm.dirty_ratio=15
sysctl -w vm.dirty_background_ratio=5
echo "✅ 系统调优完成"2. Python环境优化
# performance_config.py - Python性能配置
import gc
import sys
import os
def optimize_python_environment():
"""优化Python运行环境"""
# 禁用垃圾回收（短期测试）
gc.disable()
# 设置递归限制
sys.setrecursionlimit(10000)
# 环境变量优化
os.environ['PYTHONUNBUFFERED'] = '1' # 禁用输出缓冲
os.environ['PYTHONDONTWRITEBYTECODE'] = '1' # 禁用字节码生成
# 设置线程栈大小
import threading
threading.stack_size(2**20) # 1MB栈大小
print("🐍 Python环境优化完成")
# 在Locust脚本开头调用
optimize_python_environment()📊 内存管理优化
1. 对象池模式
import queue
import threading
from locust import HttpUser, task
class ObjectPool:
"""对象池，减少对象创建开销"""
def __init__(self, factory_func, max_size=100):
self.factory_func = factory_func
self.pool = queue.Queue(maxsize=max_size)
self.lock = threading.Lock()
def get_object(self):
"""获取对象"""
try:
return self.pool.get_nowait()
except queue.Empty:
return self.factory_func()
def return_object(self, obj):
"""归还对象"""
try:
# 重置对象状态
if hasattr(obj, 'reset'):
obj.reset()
self.pool.put_nowait(obj)
except queue.Full:
pass # 池满时丢弃对象
class RequestData:
"""请求数据对象"""
def __init__(self):
self.reset()
def reset(self):
"""重置对象状态"""
self.url = ""
self.params = {}
self.headers = {}
self.data = None
# 创建对象池
request_pool = ObjectPool(RequestData, max_size=1000)
class PooledUser(HttpUser):
wait_time = between(0.1, 0.3)
host = "https://httpbin.org"
@task
def pooled_request(self):
"""使用对象池的请求"""
# 从池中获取对象
req_data = request_pool.get_object()
try:
# 设置请求数据
req_data.url = "/get"
req_data.params = {"test": "value"}
# 发送请求
response = self.client.get(
req_data.url,
params=req_data.params
)
finally:
# 归还对象到池中
request_pool.return_object(req_data)2. 内存监控
import tracemalloc
import psutil
import gc
from locust import events
class MemoryMonitor:
"""内存监控器"""
def __init__(self):
self.start_memory = 0
self.peak_memory = 0
def start_monitoring(self):
"""开始内存监控"""
tracemalloc.start()
process = psutil.Process()
self.start_memory = process.memory_info().rss / 1024 / 1024 # MB
print(f"📊 内存监控开始，初始内存: {self.start_memory:.1f}MB")
def check_memory(self):
"""检查内存使用"""
process = psutil.Process()
current_memory = process.memory_info().rss / 1024 / 1024 # MB
if current_memory > self.peak_memory:
self.peak_memory = current_memory
# 内存增长过快时触发GC
if current_memory > self.start_memory * 2:
print(f"⚠️ 内存使用过高: {current_memory:.1f}MB，触发垃圾回收")
gc.collect()
return current_memory
def get_top_memory_usage(self):
"""获取内存使用排行"""
if tracemalloc.is_tracing():
snapshot = tracemalloc.take_snapshot()
top_stats = snapshot.statistics('lineno')
print("\n🔍 内存使用排行:")
for index, stat in enumerate(top_stats[:5], 1):
print(f" {index}. {stat}")
# 全局内存监控
memory_monitor = MemoryMonitor()
@events.test_start.add_listener
def on_test_start(environment, **kwargs):
memory_monitor.start_monitoring()
@events.request_success.add_listener
def on_request_success(request_type, name, response_time, response_length, **kwargs):
# 每100个请求检查一次内存
if hasattr(on_request_success, 'counter'):
on_request_success.counter += 1
else:
on_request_success.counter = 1
if on_request_success.counter % 100 == 0:
memory_monitor.check_memory()
@events.test_stop.add_listener
def on_test_stop(environment, **kwargs):
memory_monitor.get_top_memory_usage()
print(f"📊 峰值内存使用: {memory_monitor.peak_memory:.1f}MB")🚀 并发优化策略
1. 智能负载控制
from locust import LoadTestShape
import math
class AdaptiveLoadShape(LoadTestShape):
"""自适应负载模型"""
def __init__(self):
self.target_response_time = 500 # 目标响应时间(ms)
self.max_users = 1000
self.min_users = 10
self.current_users = self.min_users
self.adjustment_interval = 30 # 调整间隔(秒)
def tick(self):
run_time = self.get_run_time()
if run_time < self.adjustment_interval:
return (self.current_users, 5)
# 获取当前性能指标
stats = self.runner.stats.total
avg_response_time = stats.avg_response_time
# 根据响应时间调整用户数
if avg_response_time < self.target_response_time * 0.8:
# 响应时间良好，增加用户
self.current_users = min(
self.current_users + 10,
self.max_users
)
elif avg_response_time > self.target_response_time * 1.2:
# 响应时间过长，减少用户
self.current_users = max(
self.current_users - 5,
self.min_users
)
print(f"🎯 自适应调整: 用户数={self.current_users}, "
f"响应时间={avg_response_time:.1f}ms")
return (self.current_users, 5)
class AdaptiveUser(HttpUser):
wait_time = between(0.1, 0.5)
host = "https://httpbin.org"
@task
def adaptive_request(self):
self.client.get("/get")2. 批量请求优化
import asyncio
import aiohttp
from locust import User, task
class BatchUser(User):
"""批量请求用户"""
wait_time = between(1, 2)
host = "https://httpbin.org"
batch_size = 10 # 批量大小
async def batch_requests(self, urls):
"""批量发送请求"""
async with aiohttp.ClientSession() as session:
tasks = []
for url in urls:
task = session.get(f"{self.host}{url}")
tasks.append(task)
responses = await asyncio.gather(*tasks, return_exceptions=True)
# 处理响应
success_count = 0
for i, response in enumerate(responses):
if isinstance(response, Exception):
print(f"❌ 请求失败: {urls[i]} - {response}")
else:
success_count += 1
response.close()
print(f"✅ 批量请求完成: {success_count}/{len(urls)} 成功")
@task
def batch_test(self):
"""批量测试任务"""
urls = [f"/get?id={i}" for i in range(self.batch_size)]
asyncio.run(self.batch_requests(urls))🎯 性能调优检查清单
客户端优化 ✅
系统级优化 ✅
脚本优化 ✅
监控告警 ✅
🏆 性能优化总结
记住这些优化原则：
- 测量先行：先监控，再优化
- 逐步优化：一次只改一个参数
- 持续监控：优化后要持续观察
- 平衡取舍：性能和稳定性要平衡
性能优化就像调音师调钢琴，需要耐心和技巧。通过这些优化技巧，你的Locust压测将更加高效稳定！🎼
下一步可以学习框架封装和扩展，让你的压测工具更加专业化。