JmeterEngine 0.1.0

A jmeter engine

JMeter执行器

基于项目架构封装的JMeter性能测试执行器，支持通过后端接口调用执行性能测试。

功能特性

• 🚀 自动化脚本生成: 根据配置自动生成JMeter测试脚本
• 📊 实时结果收集: 自动解析JMeter执行结果并生成报告
• 🔄 消息队列集成: 与RabbitMQ集成，支持异步任务处理
• 💾 数据库存储: 测试结果自动保存到数据库
• 📈 性能监控: 支持响应时间、吞吐量、错误率等关键指标
• 🎯 灵活配置: 支持多种测试场景和参数配置

项目结构

JmeterEngine/
├── __init__.py                 # 模块初始化
├── jmeter_executor.py         # JMeter执行器核心类
├── jmeter_consumer.py         # 消息队列消费者
├── jmeter_node.py            # JMeter节点主入口
├── jmeter_api.py             # RESTful API接口
├── test_config_example.json  # 测试配置示例
└── README.md                 # 说明文档

核心组件

1. JMeterExecutor

JMeter执行器的核心类，负责：

• 生成JMeter测试脚本
• 执行JMeter测试
• 解析测试结果

2. JMeterTestRunner

测试运行器，集成到项目架构中：

• 接收测试配置
• 执行性能测试
• 返回标准化结果

3. JMeterConsumer

消息队列消费者：

• 监听RabbitMQ消息
• 处理性能测试任务
• 保存测试结果到数据库

4. JMeterNode

JMeter性能测试节点：

• 注册到后端服务
• 启动消息队列监听
• 管理节点生命周期

使用方法

1. 环境准备

确保已安装JMeter并设置环境变量：

export JMETER_HOME=/path/to/jmeter

2. 启动JMeter节点

from JmeterEngine.jmeter_node import main
main()

3. 通过API调用

启动API服务：

from JmeterEngine.jmeter_api import app
app.run(host='0.0.0.0', port=5001)

4. 发送测试任务

通过消息队列发送测试任务：

import json
import pika

# 连接RabbitMQ
connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()

# 发送测试任务
test_config = {
    "name": "API性能测试",
    "threads": 10,
    "ramp_time": 60,
    "duration": 300,
    "requests": [
        {
            "name": "GET请求",
            "method": "GET",
            "domain": "httpbin.org",
            "path": "/get"
        }
    ]
}

message = {
    "env_config": {},
    "test_suite": test_config
}

channel.basic_publish(
    exchange='',
    routing_key='perf_device_jmeter',
    body=json.dumps(message)
)

配置说明

测试配置格式

{
  "name": "测试名称",
  "threads": 10,           // 并发线程数
  "ramp_time": 60,         // 启动时间(秒)
  "duration": 300,         // 持续时间(秒)
  "requests": [            // 请求列表
    {
      "name": "请求名称",
      "method": "GET",     // HTTP方法
      "domain": "api.example.com",
      "path": "/api/test",
      "headers": {         // 请求头
        "Content-Type": "application/json"
      },
      "params": {          // 请求参数
        "key": "value"
      },
      "assertions": [      // 断言
        {
          "type": "response_code",
          "expected_code": 200,
          "message": "响应码应为200"
        }
      ]
    }
  ]
}

支持的HTTP方法

• GET
• POST
• PUT
• DELETE
• PATCH

支持的断言类型

• response_code: 响应码断言
• response_time: 响应时间断言
• response_content: 响应内容断言

API接口

1. 创建测试

POST /api/jmeter/test
Content-Type: application/json

{
  "name": "测试名称",
  "threads": 10,
  "duration": 300,
  "requests": [...]
}

2. 执行测试

POST /api/jmeter/execute
Content-Type: application/json

{
  "test_config": {...}
}

3. 获取测试模板

GET /api/jmeter/templates

4. 获取测试结果

GET /api/jmeter/results/{test_name}

5. 获取执行器状态

GET /api/jmeter/status

6. 健康检查

GET /api/jmeter/health

测试结果

执行完成后会返回以下指标：

{
  "test_name": "API性能测试",
  "status": "completed",
  "start_time": "2024-01-01T10:00:00",
  "end_time": "2024-01-01T10:05:00",
  "duration": 300000,
  "total_samples": 1000,
  "success_samples": 950,
  "failed_samples": 50,
  "error_samples": 0,
  "avg_response_time": 150.5,
  "min_response_time": 50.2,
  "max_response_time": 500.8,
  "throughput": 3.33,
  "error_rate": 5.0
}

集成说明

1. 与现有项目集成

JMeter执行器已集成到项目的性能测试架构中：

• 使用相同的消息队列系统
• 共享数据库配置
• 统一的日志管理
• 一致的结果存储格式

2. 数据库表结构

测试结果会保存到以下表：

• suite_record: 套件执行记录
• task_record: 任务执行记录
• case_record: 用例执行记录

3. 消息队列

JMeter测试任务通过以下队列接收：

• {device_id}_jmeter: JMeter性能测试队列

注意事项

1. JMeter环境: 确保JMeter已正确安装并设置环境变量
2. 权限问题: 确保有权限创建和写入测试脚本和结果文件
3. 资源限制: 根据服务器性能合理设置并发数和持续时间
4. 网络配置: 确保测试目标服务器可访问
5. 日志监控: 定期检查日志文件，及时发现问题

故障排除

常见问题

1. JMeter命令找不到

   • 检查JMETER_HOME环境变量是否正确设置
   • 确认JMeter安装路径正确

2. 测试脚本生成失败

   • 检查测试配置格式是否正确
   • 确认输出目录有写入权限

3. 测试执行失败

   • 检查目标服务器是否可访问
   • 确认网络连接正常
   • 查看JMeter日志文件

4. 结果解析失败

   • 检查结果文件是否存在
   • 确认结果文件格式正确

扩展开发

添加新的断言类型

在JMeterScriptGenerator类中添加新的断言方法：

def _add_custom_assertion(self, parent: ET.Element, assertion: Dict):
    """添加自定义断言"""
    # 实现自定义断言逻辑
    pass

添加新的监听器

在_add_listeners方法中添加新的监听器：

def _add_custom_listener(self, parent: ET.Element):
    """添加自定义监听器"""
    # 实现自定义监听器逻辑
    pass

自定义结果解析

继承JMeterTestResult类并重写解析方法：

class CustomJMeterTestResult(JMeterTestResult):
    def custom_parse_results(self):
        """自定义结果解析逻辑"""
        pass

许可证

本项目遵循MIT许可证。