Locust实战案例分析
大约 12 分钟
Locust实战案例分析
🎯 理论学得再多，不如实战一次！
今天分享几个云计算领域真实项目中的Locust应用案例，重点聚焦云游戏和云手机平台的压测实战。
就像老司机分享驾驶经验一样，我们通过实际案例来学习如何解决云平台压测的各种难题！💡
📱 案例一：云手机平台核心功能压测
背景介绍
某云手机平台需要支持大规模用户同时操作云手机设备，核心功能包括：
- 云手机开机/关机/重启
- APK应用安装/卸载
- 设备状态监控
- 用户会话管理
- 资源调度分配
挑战分析
- 设备资源有限：物理服务器上的云手机实例数量有限
- 状态同步复杂：设备状态变更需要实时同步
- 并发冲突：多用户同时操作同一设备
- 性能要求：开机重启等操作需要快速响应
解决方案
from locust import HttpUser, task, between, events
import random
import time
import json
import uuid
class CloudPhoneUser(HttpUser):
"""云手机平台用户"""
wait_time = between(2, 5)
host = "https://cloudphone-api.example.com"
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self.user_token = None
self.user_id = None
self.assigned_devices = []
self.session_id = None
def on_start(self):
"""用户初始化"""
self.login()
self.get_available_devices()
def login(self):
"""用户登录"""
# 生成测试用户数据
self.user_id = f"test_user_{random.randint(10000, 99999)}"
login_data = {
"username": self.user_id,
"password": "Test123456",
"device_type": "web"
}
# 登录获取token
login_response = self.client.post("/api/auth/login", json=login_data)
if login_response.status_code == 200:
login_result = login_response.json()
self.user_token = login_result["access_token"]
self.session_id = login_result["session_id"]
self.client.headers.update({
"Authorization": f"Bearer {self.user_token}",
"X-Session-ID": self.session_id
})
print(f"🔑 用户登录成功: {self.user_id}")
else:
print(f"❌ 登录失败: {login_response.status_code}")
def get_available_devices(self):
"""获取可用设备列表"""
response = self.client.get("/api/devices/available")
if response.status_code == 200:
devices_data = response.json()
self.assigned_devices = devices_data.get("devices", [])
print(f"📱 获取到 {len(self.assigned_devices)} 台可用设备")
@task(5)
def check_device_status(self):
"""检查设备状态"""
if not self.assigned_devices:
return
device = random.choice(self.assigned_devices)
response = self.client.get(f"/api/devices/{device['device_id']}/status")
if response.status_code == 200:
status_data = response.json()
device_status = status_data.get("status", "unknown")
print(f"📱 设备 {device['device_id']} 状态: {device_status}")
@task(3)
def get_device_info(self):
"""获取设备详细信息"""
if not self.assigned_devices:
return
device = random.choice(self.assigned_devices)
response = self.client.get(f"/api/devices/{device['device_id']}/info")
if response.status_code == 200:
device_info = response.json()
print(f"ℹ️ 设备信息: {device_info.get('model', 'Unknown')} - {device_info.get('android_version', 'Unknown')}")
@task(8)
def device_power_operations(self):
"""设备电源操作（核心业务流程）"""
if not self.assigned_devices:
return
device = random.choice(self.assigned_devices)
device_id = device['device_id']
# 随机选择电源操作
operations = ['power_on', 'power_off', 'reboot']
operation = random.choice(operations)
operation_data = {
"device_id": device_id,
"operation": operation,
"user_id": self.user_id,
"force": False # 是否强制操作
}
# 使用catch_response来处理设备状态冲突
with self.client.post(
f"/api/devices/{device_id}/power",
json=operation_data,
catch_response=True,
name=f"device_{operation}"
) as response:
if response.status_code == 200:
result = response.json()
task_id = result.get("task_id")
print(f"✅ {operation} 操作启动成功: {task_id}")
# 等待操作完成并检查结果
self.wait_for_operation_complete(task_id, operation)
response.success()
elif response.status_code == 409:
# 设备状态冲突，这是正常的业务场景
print(f"⚠️ 设备 {device_id} 状态冲突，无法执行 {operation}")
response.success() # 标记为成功，因为这是预期的业务行为
elif response.status_code == 423:
# 设备被锁定
print(f"🔒 设备 {device_id} 被其他用户占用")
response.success()
else:
response.failure(f"{operation} 操作失败: {response.status_code}")
def wait_for_operation_complete(self, task_id, operation):
"""等待操作完成"""
max_wait_time = 30 # 最大等待30秒
start_time = time.time()
while time.time() - start_time < max_wait_time:
response = self.client.get(f"/api/tasks/{task_id}/status")
if response.status_code == 200:
task_status = response.json()
status = task_status.get("status")
if status == "completed":
print(f"✅ {operation} 操作完成")
return True
elif status == "failed":
print(f"❌ {operation} 操作失败: {task_status.get('error', 'Unknown error')}")
return False
elif status == "running":
time.sleep(2) # 等待2秒后重试
continue
time.sleep(1)
print(f"⏰ {operation} 操作超时")
return False
@task(6)
def install_apk(self):
"""安装APK应用"""
if not self.assigned_devices:
return
device = random.choice(self.assigned_devices)
device_id = device['device_id']
# 模拟常用APK安装
apk_packages = [
{"name": "微信", "package": "com.tencent.mm", "url": "https://example.com/wechat.apk"},
{"name": "抖音", "package": "com.ss.android.ugc.aweme", "url": "https://example.com/douyin.apk"},
{"name": "淘宝", "package": "com.taobao.taobao", "url": "https://example.com/taobao.apk"},
{"name": "王者荣耀", "package": "com.tencent.tmgp.sgame", "url": "https://example.com/wzry.apk"}
]
apk_info = random.choice(apk_packages)
install_data = {
"device_id": device_id,
"package_name": apk_info["package"],
"apk_url": apk_info["url"],
"app_name": apk_info["name"]
}
with self.client.post(
f"/api/devices/{device_id}/install-apk",
json=install_data,
catch_response=True,
name="install_apk"
) as response:
if response.status_code == 200:
result = response.json()
task_id = result.get("task_id")
print(f"📦 开始安装 {apk_info['name']}: {task_id}")
response.success()
elif response.status_code == 400:
print(f"❌ APK安装参数错误: {apk_info['name']}")
response.failure("APK安装参数错误")
else:
response.failure(f"APK安装失败: {response.status_code}")
@task(1)
def check_my_devices(self):
"""查看我的设备列表"""
response = self.client.get("/api/user/devices")
if response.status_code == 200:
devices = response.json()["devices"]
print(f"📱 我的设备: {len(devices)} 台")
# 自定义负载模型 - 模拟云手机平台的使用高峰
from locust import LoadTestShape
class CloudPhoneLoadShape(LoadTestShape):
"""云手机平台负载模型"""
def tick(self):
run_time = self.get_run_time()
if run_time < 120:
# 前2分钟：用户陆续上线
return (100, 10)
elif run_time < 300:
# 2-5分钟：使用高峰期，大量设备操作
return (500, 50)
elif run_time < 600:
# 5-10分钟：持续高负载
return (300, 30)
elif run_time < 900:
# 10-15分钟：流量回落
return (150, 15)
else:
# 停止测试
return None测试结果分析
通过这次云手机平台压测，我们发现了几个关键问题：
- 设备资源竞争：高并发时设备分配出现冲突
- 状态同步延迟：设备状态变更通知存在延迟
- APK安装超时：大文件安装时网络传输超时
优化建议：
- 使用设备池和预分配策略减少竞争
- 优化状态同步机制，使用WebSocket推送
- 增加APK缓存和断点续传功能
🎮 案例二：云游戏平台实时性能压测
背景介绍
某云游戏平台需要支持大量用户同时在线游戏，涉及实时音视频流传输和游戏控制指令处理。
核心挑战
- 实时性要求：游戏画面延迟需要控制在50ms以内
- 带宽压力：高清视频流占用大量带宽
- 并发连接：支持数万用户同时在线
- 资源调度：GPU服务器资源的合理分配
解决方案
from locust import HttpUser, task, between, events
import time
import random
import json
import websocket
import threading
class CloudGameUser(HttpUser):
"""云游戏用户"""
wait_time = between(0.1, 1) # 游戏场景下操作频繁
host = "https://cloudgame-api.example.com"
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self.user_token = None
self.game_session_id = None
self.available_games = []
self.ws_connection = None
self.game_instance_id = None
def on_start(self):
"""用户初始化"""
self.login()
self.load_available_games()
def login(self):
"""用户登录"""
user_id = random.randint(10000, 99999)
login_data = {
"username": f"gamer_{user_id}",
"password": "Game123456",
"platform": "web"
}
response = self.client.post("/api/auth/login", json=login_data)
if response.status_code == 200:
result = response.json()
self.user_token = result["access_token"]
self.client.headers.update({
"Authorization": f"Bearer {self.user_token}"
})
print(f"🎮 玩家登录成功: gamer_{user_id}")
def load_available_games(self):
"""加载可用游戏列表"""
response = self.client.get("/api/games/available")
if response.status_code == 200:
games_data = response.json()
self.available_games = games_data.get("games", [])
print(f"🎯 可用游戏: {len(self.available_games)} 款")
@task(8)
def start_game_session(self):
"""启动游戏会话（核心流程）"""
if not self.available_games:
return
game = random.choice(self.available_games)
# 请求启动游戏实例
start_data = {
"game_id": game["id"],
"quality": random.choice(["720p", "1080p"]),
"region": "cn-east-1"
}
with self.client.post(
"/api/games/start",
json=start_data,
catch_response=True,
name="start_game"
) as response:
if response.status_code == 200:
result = response.json()
self.game_instance_id = result["instance_id"]
self.game_session_id = result["session_id"]
print(f"🚀 游戏启动成功: {game['name']} - {self.game_instance_id}")
# 建立WebSocket连接用于游戏控制
self.establish_game_connection()
response.success()
elif response.status_code == 503:
# 服务器资源不足
print(f"⚠️ 服务器资源不足，无法启动游戏: {game['name']}")
response.success() # 这是预期的业务场景
else:
response.failure(f"游戏启动失败: {response.status_code}")
def establish_game_connection(self):
"""建立游戏控制WebSocket连接"""
if not self.game_session_id:
return
try:
ws_url = f"wss://cloudgame-ws.example.com/game/{self.game_session_id}"
self.ws_connection = websocket.WebSocketApp(
ws_url,
header=[f"Authorization: Bearer {self.user_token}"],
on_open=self.on_ws_open,
on_message=self.on_ws_message,
on_error=self.on_ws_error
)
# 在后台线程运行WebSocket
ws_thread = threading.Thread(target=self.ws_connection.run_forever)
ws_thread.daemon = True
ws_thread.start()
except Exception as e:
print(f"❌ WebSocket连接失败: {e}")
def on_ws_open(self, ws):
"""WebSocket连接建立"""
print(f"🔗 游戏控制连接建立: {self.game_session_id}")
def on_ws_message(self, ws, message):
"""接收游戏画面数据"""
# 模拟处理游戏画面数据
pass
def on_ws_error(self, ws, error):
"""WebSocket错误"""
print(f"❌ 游戏连接错误: {error}")
@task(15)
def send_game_input(self):
"""发送游戏操作指令"""
if not self.ws_connection or not self.game_session_id:
return
# 模拟各种游戏操作
input_actions = [
{"type": "key_press", "key": "W"},
{"type": "key_press", "key": "A"},
{"type": "key_press", "key": "S"},
{"type": "key_press", "key": "D"},
{"type": "mouse_click", "x": random.randint(0, 1920), "y": random.randint(0, 1080)},
{"type": "key_press", "key": "SPACE"}
]
action = random.choice(input_actions)
input_data = {
"session_id": self.game_session_id,
"action": action,
"timestamp": int(time.time() * 1000)
}
try:
self.ws_connection.send(json.dumps(input_data))
except Exception as e:
print(f"❌ 发送游戏指令失败: {e}")
@task(3)
def check_game_status(self):
"""检查游戏状态"""
if not self.game_instance_id:
return
response = self.client.get(f"/api/games/{self.game_instance_id}/status")
if response.status_code == 200:
status_data = response.json()
latency = status_data.get("latency", 0)
fps = status_data.get("fps", 0)
if latency > 100: # 延迟超过100ms
print(f"⚠️ 游戏延迟过高: {latency}ms")
print(f"📊 游戏状态 - 延迟: {latency}ms, FPS: {fps}")
@task(1)
def stop_game_session(self):
"""停止游戏会话"""
if not self.game_instance_id:
return
response = self.client.post(f"/api/games/{self.game_instance_id}/stop")
if response.status_code == 200:
print(f"🛑 游戏会话结束: {self.game_instance_id}")
# 关闭WebSocket连接
if self.ws_connection:
self.ws_connection.close()
self.ws_connection = None
self.game_instance_id = None
self.game_session_id = None
# 云游戏专用负载模型
class CloudGameLoadShape(LoadTestShape):
"""云游戏负载模型"""
def tick(self):
run_time = self.get_run_time()
if run_time < 60:
# 前1分钟：玩家陆续上线
return (50, 5)
elif run_time < 180:
# 1-3分钟：游戏高峰期
return (300, 30)
elif run_time < 600:
# 3-10分钟：持续游戏
return (200, 20)
elif run_time < 900:
# 10-15分钟：玩家逐渐下线
return (100, 10)
else:
return None
# 监控游戏性能指标
game_stats = {
"total_sessions": 0,
"successful_starts": 0,
"high_latency_count": 0,
"connection_failures": 0
}
@events.request_success.add_listener
def on_game_success(request_type, name, response_time, response_length, **kwargs):
if name == "start_game":
game_stats["total_sessions"] += 1
game_stats["successful_starts"] += 1
@events.request_failure.add_listener
def on_game_failure(request_type, name, response_time, response_length, exception, **kwargs):
if name == "start_game":
game_stats["total_sessions"] += 1
game_stats["connection_failures"] += 1
@events.test_stop.add_listener
def on_test_stop(environment, **kwargs):
total = game_stats["total_sessions"]
success = game_stats["successful_starts"]
if total > 0:
success_rate = (success / total) * 100
print(f"\n📊 云游戏统计:")
print(f" 总会话数: {total}")
print(f" 成功启动: {success}")
print(f" 启动成功率: {success_rate:.2f}%")
print(f" 连接失败: {game_stats['connection_failures']}")
print(f" 高延迟次数: {game_stats['high_latency_count']}")📱 案例三：云手机批量管理压测
背景介绍
某云手机管理平台需要支持运营人员批量管理数千台云手机设备，包括批量操作、状态监控和资源调度。
解决方案
from locust import HttpUser, task, between, events
import json
import time
import random
from concurrent.futures import ThreadPoolExecutor
class CloudPhoneManagerUser(HttpUser):
"""云手机管理员用户"""
wait_time = between(1, 3)
host = "https://cloudphone-admin.example.com"
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self.admin_token = None
self.managed_devices = []
self.batch_operations = []
def on_start(self):
"""管理员初始化"""
self.admin_login()
self.load_managed_devices()
def on_stop(self):
"""清理资源"""
self.cleanup_batch_operations()
def admin_login(self):
"""管理员登录"""
admin_data = {
"username": f"admin_{random.randint(1, 100)}",
"password": "Admin123456",
"role": "device_manager"
}
response = self.client.post("/api/admin/login", json=admin_data)
if response.status_code == 200:
result = response.json()
self.admin_token = result["access_token"]
self.client.headers.update({
"Authorization": f"Bearer {self.admin_token}",
"X-Admin-Role": "device_manager"
})
print(f"👨💼 管理员登录成功: {admin_data['username']}")
def load_managed_devices(self):
"""加载管理的设备列表"""
response = self.client.get("/api/admin/devices", params={
"page": 1,
"size": 100,
"status": "all"
})
if response.status_code == 200:
devices_data = response.json()
self.managed_devices = devices_data.get("devices", [])
print(f"📱 管理设备数量: {len(self.managed_devices)}")
def cleanup_batch_operations(self):
"""清理批量操作"""
for operation in self.batch_operations:
if operation.get("status") == "running":
self.client.post(f"/api/admin/batch-operations/{operation['id']}/cancel")
@task(10)
def batch_device_operations(self):
"""批量设备操作（核心功能）"""
if len(self.managed_devices) < 10:
return
# 随机选择一批设备进行操作
batch_size = random.randint(10, 50)
selected_devices = random.sample(self.managed_devices, min(batch_size, len(self.managed_devices)))
# 随机选择批量操作类型
operations = [
{"type": "batch_reboot", "name": "批量重启"},
{"type": "batch_install_apk", "name": "批量安装APK"},
{"type": "batch_uninstall_apk", "name": "批量卸载APK"},
{"type": "batch_clear_data", "name": "批量清理数据"}
]
operation = random.choice(operations)
batch_data = {
"operation_type": operation["type"],
"device_ids": [device["device_id"] for device in selected_devices],
"parameters": self.get_operation_parameters(operation["type"]),
"priority": random.choice(["high", "normal", "low"])
}
with self.client.post(
"/api/admin/batch-operations",
json=batch_data,
catch_response=True,
name=f"batch_{operation['type']}"
) as response:
if response.status_code == 200:
result = response.json()
operation_id = result["operation_id"]
print(f"🔄 {operation['name']} 启动成功: {operation_id} (设备数: {len(selected_devices)})")
# 记录批量操作
self.batch_operations.append({
"id": operation_id,
"type": operation["type"],
"status": "running"
})
response.success()
else:
response.failure(f"批量操作失败: {response.status_code}")
def get_operation_parameters(self, operation_type):
"""获取操作参数"""
if operation_type == "batch_install_apk":
apk_list = [
{"package": "com.tencent.mm", "url": "https://example.com/wechat.apk"},
{"package": "com.ss.android.ugc.aweme", "url": "https://example.com/douyin.apk"},
{"package": "com.taobao.taobao", "url": "https://example.com/taobao.apk"}
]
return {"apk": random.choice(apk_list)}
elif operation_type == "batch_uninstall_apk":
return {"package": random.choice(["com.example.test1", "com.example.test2"])}
elif operation_type == "batch_clear_data":
return {"clear_type": random.choice(["cache", "user_data", "all"])}
return {}
@task(5)
def monitor_batch_operations(self):
"""监控批量操作状态"""
if not self.batch_operations:
return
# 检查正在运行的操作
running_operations = [op for op in self.batch_operations if op["status"] == "running"]
for operation in running_operations[:3]: # 最多检查3个操作
response = self.client.get(f"/api/admin/batch-operations/{operation['id']}/status")
if response.status_code == 200:
status_data = response.json()
current_status = status_data["status"]
progress = status_data.get("progress", 0)
if current_status != "running":
operation["status"] = current_status
print(f"📊 批量操作完成: {operation['id']} - {current_status} (进度: {progress}%)")
@task(3)
def get_device_statistics(self):
"""获取设备统计信息"""
response = self.client.get("/api/admin/statistics/devices")
if response.status_code == 200:
stats = response.json()
print(f"📈 设备统计 - 在线: {stats.get('online', 0)}, 离线: {stats.get('offline', 0)}, 故障: {stats.get('error', 0)}")
@task(2)
def export_device_report(self):
"""导出设备报告"""
export_data = {
"report_type": random.choice(["daily", "weekly", "monthly"]),
"format": random.choice(["excel", "csv", "pdf"]),
"include_details": random.choice([True, False])
}
response = self.client.post("/api/admin/reports/export", json=export_data)
if response.status_code == 200:
result = response.json()
print(f"📄 报告导出任务创建: {result.get('task_id')}")🎯 云计算平台压测实战经验总结
1. 云平台测试策略
- 资源隔离测试：确保不同用户的云手机资源完全隔离
- 弹性扩容测试：验证系统在负载增加时的自动扩容能力
- 故障恢复测试：模拟设备故障时的快速恢复机制
- 多租户并发测试：测试多个租户同时使用时的性能表现
2. 云手机平台常见问题及解决方案
| 问题 | 原因 | 解决方案 |
|---|---|---|
| 设备分配冲突 | 多用户抢占同一设备 | 实现设备预分配和锁定机制 |
| 状态同步延迟 | 设备状态变更通知机制不完善 | 使用消息队列和WebSocket推送 |
| APK安装超时 | 网络带宽限制或文件过大 | 实现断点续传和本地缓存 |
| 批量操作阻塞 | 大批量操作占用过多资源 | 使用任务队列和限流机制 |
| 设备资源泄漏 | 用户会话结束后设备未释放 | 实现会话超时和自动回收 |
3. 云平台性能指标关注点
- 设备利用率：物理服务器上云手机的使用率
- 操作响应时间：开机、重启、安装APK的耗时
- 并发处理能力：同时支持的用户数和设备操作数
- 资源调度效率：设备分配和回收的速度
- 网络延迟：云手机操作的网络传输延迟
- 存储IO性能：APK安装和数据读写的IO表现
4. 云计算压测最佳实践
每次压测后的关键工作：
- 分析资源瓶颈：识别CPU、内存、网络、存储的瓶颈点
- 优化调度算法：改进设备分配和负载均衡策略
- 完善监控体系：建立实时监控和告警机制
- 制定扩容策略：根据压测结果制定自动扩容规则
🚀 云计算压测进阶指南
通过这些云计算平台的实战案例，你应该能够：
- 掌握云平台特性：理解云手机、云游戏等云计算业务的特殊性
- 设计云原生测试：制定适合云平台的压测策略和场景
- 处理复杂并发：解决设备资源竞争、状态同步等云平台特有问题
- 优化资源调度：通过压测数据优化云资源的分配和管理策略
💡 云计算压测核心要点
- 资源弹性：测试系统的自动扩缩容能力
- 多租户隔离：确保不同用户间的完全隔离
- 故障容错：验证单点故障时的快速恢复
- 成本优化：通过压测找到性价比最优的资源配置
记住，云计算压测的目标不仅是验证性能，更是要确保在复杂的云环境下，系统能够稳定、高效、经济地为用户提供服务！☁️
建议结合实际的云手机或云游戏项目，运用这些技巧进行深度压测实践，积累更多云计算领域的测试经验。