1. 压力测试基础
1.1 压力测试的定义与目的
压力测试（Stress Testing） 是一种通过向系统施加超出正常预期负载的请求，来评估系统在极端条件下的行为、稳定性和恢复能力的测试方法。其核心目标是找到系统的崩溃点和瓶颈，而非验证功能正确性。
核心目的：
- 确定极限容量： 找出系统在性能严重下降或崩溃前能承受的最大并发用户数或吞吐量。
- 验证恢复能力： 测试系统在过载崩溃后能否自动恢复，以及恢复所需的时间。
- 暴露资源竞争： 发现内存泄漏、线程死锁、数据库连接池耗尽等仅在高压下才显现的问题。
- 评估降级策略： 验证熔断、限流、服务降级等容错机制是否按预期生效。
1.2 压力测试与负载测试的区别
压力测试常与负载测试（Load Testing）混淆。两者核心区别在于测试目标和负载模型：负载测试关注系统在预期正常负载下的性能表现；压力测试则关注系统在超预期负载下的行为边界。
| 对比维度 | 负载测试 | 压力测试 |
|---|---|---|
| 核心目标 | 验证系统在预期负载下的响应时间、吞吐量是否达标 | 找出系统能承受的最大负载及崩溃后的恢复能力 |
| 负载模型 | 模拟典型业务高峰期的并发用户数（如 1000 并发） | 持续增加并发直至系统崩溃（如 1000 → 2000 → 5000） |
| 关注指标 | 平均响应时间、95 分位响应时间、吞吐量 | 崩溃点、错误率拐点、恢复时间、资源耗尽阈值 |
| 测试场景 | 平稳负载、阶梯负载、峰值负载 | 突增负载、持续高压、资源耗尽 |
| 预期结果 | 系统正常运行，性能指标在可接受范围内 | 系统可能降级或崩溃，但需优雅恢复 |
实战建议： 在特征工程项目中，建议先执行负载测试验证基线性能，再执行压力测试确定特征处理管线的容量上限。例如，先测试 1000 QPS 下特征提取的延迟，再逐步加压直到特征服务超时或 OOM。
1.3 常见压力测试工具简介
业界主流的压力测试工具各有侧重，选择时需考虑协议支持、可扩展性、分布式能力和与特征工程管线的集成便利性。以下对 JMeter、Locust 和 Gatling 进行对比分析。
| 特性 | JMeter | Locust | Gatling |
|---|---|---|---|
| 开发语言 | Java（GUI 配置为主） | Python（纯代码定义场景） | Scala / Java（DSL 脚本） |
| 协议支持 | HTTP、JDBC、JMS、FTP、TCP 等（插件丰富） | HTTP / HTTPS（原生）；其他协议需扩展 | HTTP、WebSocket、JMS、MQTT |
| 脚本方式 | GUI 录制 + XML 配置 | Python 代码（灵活，适合动态参数） | Scala DSL（代码即配置） |
| 分布式支持 | Master-Slave 模式 | Master-Worker 模式（原生支持） | 集群模式（需额外配置） |
| 实时监控 | 监听器 + 第三方后端（InfluxDB + Grafana） | Web UI 实时图表 + CSV 导出 | 内置图表 + 高级报告生成 |
| 与特征工程集成 | 通过 BeanShell / JSR223 调用特征 API | 直接 Python 调用特征 SDK，集成最自然 | 通过 HTTP 请求调用特征服务 |
| 学习曲线 | 中等（GUI 易上手，高级配置复杂） | 低（Python 基础即可） | 较高（需掌握 Scala DSL） |
1.4 工具选型建议
针对特征工程场景，推荐以下选型策略：
- 特征 API 服务压测： 优先选择 Locust，因其 Python 原生支持可直接复用特征工程团队的 Python 代码，方便构造包含特征计算逻辑的动态请求。
- 全链路压测（含数据库/中间件）： 选择 JMeter，其丰富的 JDBC、JMS 插件可模拟特征存储层的读写压力。
- 高并发 HTTP 压测 + 详细报告： 选择 Gatling，其 Akka 底层架构在高并发下资源消耗更低，且生成的 HTML 报告非常详尽。
注意： 在特征工程压力测试中，不要只关注 HTTP 层面的响应时间。必须同时监控特征计算节点的 CPU、内存、特征缓存命中率、特征存储 IOPS 等指标，否则容易遗漏真正的瓶颈。
1.5 快速上手示例：Locust 压测特征服务
以下是一个使用 Locust 对特征提取 API 进行压力测试的简单示例：
# locustfile.py
from locust import HttpUser, task, between
import json
class FeatureExtractionUser(HttpUser):
wait_time = between(0.5, 2.0) # 模拟用户思考时间
@task
def extract_features(self):
payload = {
"user_id": "test_user_001",
"item_id": "item_12345",
"context": {
"timestamp": 1700000000,
"device": "mobile"
}
}
headers = {"Content-Type": "application/json"}
with self.client.post("/api/v1/features/extract",
data=json.dumps(payload),
headers=headers,
catch_response=True) as response:
if response.status_code != 200:
response.failure(f"Status code: {response.status_code}")
elif response.elapsed.total_seconds() > 0.5:
response.failure(f"Response time too long: {response.elapsed.total_seconds():.2f}s")
运行命令：
# 启动 Locust Web UI（默认 http://localhost:8089）
locust -f locustfile.py --host=http://your-feature-service:8000
# 无界面模式运行（适合 CI/CD）
locust -f locustfile.py --host=http://your-feature-service:8000 \
--headless -u 100 -r 10 --run-time 5m --csv=feature_stress_result
关键参数说明：
-u 100：模拟 100 个并发用户-r 10：每秒启动 10 个用户（爬坡速率）--run-time 5m：测试持续 5 分钟--csv=feature_stress_result：将结果导出为 CSV 文件，便于后续特征工程分析