如何用Python构建JMX并借助JMeter进行高效的压力测试？

本文共计1434个文字，预计阅读时间需要6分钟。

JMeter支持通过命令行调用Python脚本，Python脚本同样可以通过系统命令调用JMeter执行压力测试。首先需要安装JMeter，从官方下载安装，解压后配置环境路径。可以使用软连接建立快捷方式，或在命令行中直接指定JMeter的执行路径。

JMeter可以通过os命令调用Python脚本，Python同样可以通过系统命令调用JMeter执行压测

Python调用JMeter

首先要安装JMeter，官方下载地址

解压并配置配置环境路径或建立软连，使得在命令输入jmeter便可以执行，如

unzip apache-jmeter-5.3.zip
mv apache-jmeter-5.3 /usr/loca/jmeter
ln -s /usr/local/jmeter/bin/jmeter /usr/bin/jmeter
ln -s /usr/local/jmeter/bin/jmeter-server /usr/bin/jmeter-server

打开JMeter并设计一个测试计划保存为testplan.jmx

使用Python调用JMeter压测并生成报告

Python中可以使用os.system()或supprocess.Popen()调用系统命令，前者实时显示在屏幕上，后者可以获取到屏幕输出信息。
使用Python调用JMeter运行及生成报告的命令如下。

import subprocess jmx_file = 'testplan.jmx' # jmx文件路径 result_file = 'result.jtl' # log_file = 'run.log' report_dir = 'report' run_cmd = f'jmeter -n -t {jmx_file} -l {result_file} -j {log_file}' # 无界面运行JMeter压测命令 report_cmd = f'jmeter -g {result_file} -o {report_dir}' # 生成HTML报告命令 # 不需要获取屏幕输出是，可以使用os.system() # os.system(run_cmd) # os.system(report_cmd) # 需要获取屏幕输出是，可以使用subprocess.Popen() p1 = subprocess.Popen(run_cmd, shell=True, stdout=subprocess.PIPE) print(p1.stdout.read().decode('utf-8')) p2 = subprocess.Popen(report_cmd, shell=True, stdout=subprocess.PIPE) print(p2.stdout.read().decode('utf-8'))

组装jmx

每一测试计划为一个jmx文件，jmx实际上是xml格式的，包含一些JMeter自定义的格式规范。
常用的组件有：

• : 测试计划
• : 线程组
• : CSV数据文件
• : HTTP请求
• : HTTP请求头管理器
• : Cookies管理器
• : DNS缓存管理器
• : 监听器(包括查看结果树、聚合报告等)
• ： 响应断言
• <io.github.ningyu.jmeter.plugin.dubbo.sample.DubboSample></io.github.ningyu.jmeter.plugin.dubbo.sample.DubboSample>: 三方Dubbo请求插件
  

Dubbo插件jmeter-plugins-dubbo下载链接

jmx中，如果一个组件有子组件，格式为

<ThreadGroup 组件基本属性> ...线程组配置 </ThreadGroup> <hashTree> ...内部子组件 </hashTree> ··· 如果不包含子组件，则后面接一个<hashTree/> 单标签直接结束，例如： ```xml <CSVDataSet> ... CSV数据组件配置 </CSVDataSet> <hashTree/>

详细的格式可以自己使用JMeter创建一个测试计划，使用文本文件打开jmx文件查看。

使用Python组装jmx文件的方式有两种，一种是固定模板的数据渲染，一种类似JMeter的逐个组件添加，第一种比较简单。
通过分析jmx文件中的变量，我们使用jinja2模板语法，将其中的变量进行参数化，对需要判断和循环的变量设置if和for循环。

Jinja2中文文档

假设我们的测试计划结构为：

测试计划 DNS缓存管理器 Cookies管理器 CSV文件(多个) ... 聚合报告 线程组(多个) CSV文件(多个) HTTP请求(或Dubbo请求) HTTP请求头管理器 CSV文件(多个) 响应断言 察看结果树

将jmx中的关键数据抽取并组合，我使用的完整数据格式如下：

data.yaml

test_plan_name: 测试计划 comments: 测试计划描述 hosts: - name: las.secoo.com address: 112.126.120.128 cookies: clear_each_iteration: 'true' csv_files: - {'name': '数据文件1', 'path': 'data.csv', 'varnames': 'a,b', 'delimiter': ','} - {'name': '数据文件2', 'path': 'data.csv', 'varnames': 'c,d', 'delimiter': ','} thread_groups: - thread_group_name: 线程组1 comments: 线程组1描述 enabled: 'true' num_threads: 50 loops: -1 ramp_time: 0 scheduler: 'true' duration: 30 delay: '' </stringProp> <stringProp name="FIELD_DUBBO_ADDRESS">{{dubbo_sample.registry.address}}</stringProp> <stringProp name="FIELD_DUBBO_TIMEOUT">{{dubbo_sample.dubbo.timeout}}</stringProp> <stringProp name="FIELD_DUBBO_VERSION"></stringProp> <stringProp name="FIELD_DUBBO_RETRIES">{{dubbo_sample.dubbo.retries}}</stringProp> <stringProp name="FIELD_DUBBO_GROUP">{{dubbo_sample.dubbo.group}}</stringProp> <stringProp name="FIELD_DUBBO_CONNECTIONS">{{dubbo_sample.dubbo.connections}}</stringProp> <stringProp name="FIELD_DUBBO_LOADBALANCE">{{dubbo_sample.dubbo.load_balance}}</stringProp> <stringProp name="FIELD_DUBBO_ASYNC">sync</stringProp> <stringProp name="FIELD_DUBBO_CLUSTER">{{dubbo_sample.dubbo.cluster}}</stringProp> <stringProp name="FIELD_DUBBO_INTERFACE">{{dubbo_sample.dubbo.service}}</stringProp> <stringProp name="FIELD_DUBBO_METHOD">{{dubbo_sample.dubbo.method}}</stringProp> <intProp name="FIELD_DUBBO_METHOD_ARGS_SIZE">1</intProp>{% for param in dubbo_sample.dubbo.params %} <stringProp name="FIELD_DUBBO_METHOD_ARGS_PARAM_TYPE1">{{param.type}}</stringProp> <stringProp name="FIELD_DUBBO_METHOD_ARGS_PARAM_VALUE1">{{param.value}}</stringProp>{% endfor %} <intProp name="FIELD_DUBBO_ATTACHMENT_ARGS_SIZE">0</intProp> <stringProp name="FIELD_DUBBO_CONFIG_CENTER_PROTOCOL"></stringProp> <stringProp name="FIELD_DUBBO_CONFIG_CENTER_GROUP"></stringProp> <stringProp name="FIELD_DUBBO_CONFIG_CENTER_NAMESPACE"></stringProp> <stringProp name="FIELD_DUBBO_CONFIG_CENTER_USER_NAME"></stringProp> <stringProp name="FIELD_DUBBO_CONFIG_CENTER_PASSWORD"></stringProp> <stringProp name="FIELD_DUBBO_CONFIG_CENTER_ADDRESS"></stringProp> <stringProp name="FIELD_DUBBO_CONFIG_CENTER_TIMEOUT"></stringProp> <stringProp name="FIELD_DUBBO_REGISTRY_USER_NAME"></stringProp> <stringProp name="FIELD_DUBBO_REGISTRY_PASSWORD"></stringProp> <stringProp name="FIELD_DUBBO_REGISTRY_TIMEOUT"></stringProp> </io.github.ningyu.jmeter.plugin.dubbo.sample.DubboSample> <hashTree>{% if dubbo_sample.dubbo.headers %} <HeaderManager guiclass="HeaderPanel" testclass="HeaderManager" testname="HTTP信息头管理器" enabled="true"> <collectionProp name="HeaderManager.headers">{% for name, value in dubbo_sample.dubbo.headers.items() %} <elementProp name="" elementType="Header"> <stringProp name="Header.name">{{name}}</stringProp> <stringProp name="Header.value">{{value}}</stringProp> </elementProp>{% endfor %} </collectionProp> </HeaderManager> <hashTree/>{% endif %} {% if dubbo_sample.validate %} {% for assertion in dubbo_sample.validate %} <ResponseAssertion guiclass="AssertionGui" testclass="ResponseAssertion" testname="响应断言" enabled="true"> <collectionProp name="Asserion.test_strings">{% if assertion.strings %}{% for string in assertion.strings %} <stringProp name="97">{{string}}</stringProp>{% endfor %}{% endif %} </collectionProp> <stringProp name="Assertion.custom_message"></stringProp> <stringProp name="Assertion.test_field">Assertion.{{assertion.test_field}}</stringProp> <boolProp name="Assertion.assume_success">false</boolProp> <intProp name="Assertion.test_type">{{assertion.test_type}}</intProp> </ResponseAssertion>{% endfor %} {% endif %} <hashTree/>{% endfor %}{% endif %} {% endfor %} </hashTree> </hashTree> </hashTree> </jmeterTestPlan>

组装出类似data.yaml格式的数据，并使用jinja2渲染模板即可得到完整的jmx文件

pip install pyyaml jinja2

import yaml import jinja2 # 组装或读取数据 with open('data.yaml', encoding='utf-8') as f: data = yaml.safe_load(f) # 读取模板 with open('tpl.xml', encoding='utf-8') as f: tpl = f.read() # 渲染模板生成jmx jmx = jinja2.Template(tpl).render(data) with open(jmx_file, 'w', encoding='utf-8') as f: f.write(jmx)

后计

在实际项目中，还涉及数据文件的拷贝，节点环境的部署，脚本的分发，报告的下载等等，可以使用paramiko或者fabric或ansible完成，压测节点数据分发的服务管理。

以上就是本文的全部内容，希望对大家的学习有所帮助，也希望大家多多支持易盾网络。