JMeter插件开发

JMeter 自定义 Java Sampler 开发指南 —— 以 Kafka 消息消费为例

在性能测试中，JMeter 内置的采样器（如 HTTP Request）能满足大部分场景。但当我们需要测试自定义协议（如 Kafka、Socket、私有 RPC）或执行复杂的业务逻辑时，就需要开发自定义 Java Sampler。本文将结合一个从 Kafka 消费消息并校验内容的实际案例，详解开发流程与核心组件。

一、开发前置准备

JMeter 环境：下载并安装 JMeter，版本建议 5.x 以上。
开发工具：IntelliJ IDEA 或 Eclipse，Maven 工程。
依赖包：在项目中引入 ApacheJMeter_core、ApacheJMeter_java，以及业务所需的 Kafka Client 等。
打包部署：将编译后的 Jar 包（及依赖）放入 JMeter 的 lib/ext 目录，重启 JMeter。

二、自定义 Java Sampler 的核心组件

类/接口	作用
`AbstractJavaSamplerClient`	必须继承的抽象类，提供三个核心扩展方法。
`Arguments`	定义在 JMeter GUI 中显示的输入参数（名称、默认值）。
`JavaSamplerContext`	上下文对象，用于获取用户在界面配置的参数值。
`SampleResult`	记录采样结果（成功/失败、响应数据、耗时等）。

三、开发流程（三步法）

步骤1：继承 `AbstractJavaSamplerClient` 并实现三个关键方法

getDefaultParameters()定义采样器在 JMeter 界面中的参数面板。用户可在此填入 Kafka 地址、主题、搜索关键字等。

publicArgumentsgetDefaultParameters(){Arguments params =newArguments();    params.addArgument("kafka_host","192.168.118.168:7483");    params.addArgument("topic","hub-device-property");return params;}

setupTest(JavaSamplerContext ctx)每个线程启动时执行一次（类似 JUnit 的 @Before）。适合做资源初始化，如读取参数、建立连接池。
```
publicvoidsetupTest(JavaSamplerContext ctx){this.kafka_host = ctx.getParameter("kafka_host");this.topic = ctx.getParameter("topic");}
```

runTest(JavaSamplerContext ctx)每次循环都会执行，核心采样逻辑写在这里。需要：

SampleResult sr =newSampleResult();sr.sampleStart();// 业务逻辑...sr.setSuccessful(true);sr.sampleEnd();return sr;

调用 sampleStart() 开始计时
执行业务代码（如消费 Kafka、查询数据库）
设置 setSuccessful(true/false)、setResponseData()
最后调用 sampleEnd() 结束计时

步骤2：编写业务逻辑（本例中的 Kafka 消费）

我们单独封装了一个 KafkaGetter 类（代码未给出，但可推测其内部使用 Kafka Consumer API），核心功能：

根据 timeShift_mins 计算时间范围
拉取 topic 中的消息
返回包含 searchStr 的消息内容

💡 若 Kafka 消费逻辑涉及复杂依赖，建议独立为一个工具类，便于单元测试。

步骤3：打包与调试

使用 maven-assembly-plugin 将依赖一并打包（或放入 JMeter lib 目录）。
在 JMeter 中添加 Java Request 采样器，下拉选择你的类名。
填写参数，添加监听器（查看结果树、聚合报告）运行测试。

四、代码关键点解读（结合例子）

代码片段	作用
`sr.sampleStart()` / `sr.sampleEnd()`	自动计算采样耗时（单位毫秒）
`sr.setSamplerData(...)`	记录请求参数，便于在“查看结果树”中追溯
`sr.setResponseData(res, "utf-8")`	将 Kafka 返回的消息内容展示在“响应数据”中
`sr.setSuccessful(false)`	当消息不包含关键字时标记失败，JMeter 会统计错误率

五、运行效果与断言示例

成功场景：Kafka 中存在包含 p_electric_meter 的消息 → 采样结果为绿色✅。
失败场景：消息中不包含该关键字 → 采样结果为红色❌。
可结合 JMeter 断言：直接在 runTest() 中判断相当于内置断言，也可额外添加响应断言元件。

六、注意事项

线程安全：runTest() 可能被多个线程并发调用，内部业务类应无状态或做好同步。
资源释放：若在 setupTest() 中创建了连接，记得实现 teardownTest() 方法释放资源。
日志输出：使用 SampleResult.setResponseData() 可以避免大量 System.out 影响性能。
参数传递：JavaSamplerContext 中的参数是字符串类型，数值型需要手动转换。

七、扩展建议

支持动态参数：结合 JMeter 变量（${var}）实现参数化。
性能优化：Kafka 消费者可复用，避免每次采样都创建新连接。
错误分类：通过 SampleResult.setResponseCode() 和 setResponseMessage() 区分业务失败和系统异常。

八、总结

通过开发自定义 Java Sampler，JMeter 的能力可以无限延伸至任何 Java 能调用的协议或中间件。本文案例完整演示了如何从 Kafka 拉取数据并校验内容，只需三步：

继承 AbstractJavaSamplerClient
实现三个核心方法
打包部署到 JMeter

代码：

package com.xxxsmart.testUtil;import org.apache.jmeter.config.Arguments;import org.apache.jmeter.protocol.java.sampler.AbstractJavaSamplerClient;import org.apache.jmeter.protocol.java.sampler.JavaSamplerContext;import org.apache.jmeter.samplers.SampleResult;import org.slf4j.Logger;import org.slf4j.LoggerFactory;import java.io.Serializable;/** * JMeter 自定义 Java Sampler - Kafka 消息消费者 * 用于从 Kafka 指定主题拉取消息，并检查是否包含目标字符串 */public class KafkaGetterJmeter extends AbstractJavaSamplerClient implements Serializable {    // 日志记录器（注：原代码中使用了 SleepTest 的类名，实际应使用当前类名更合理）    private static final Logger LOG = LoggerFactory.getLogger(org.apache.jmeter.protocol.java.test.SleepTest.class);    // 从 JMeter 参数面板中获取的配置项    private String kafka_host;      // Kafka 服务器地址    private String topic;           // 订阅的主题    private String searchStr;       // 需要在消息中搜索的关键字    private String timeShift_mins;  // 时间偏移（分钟），用于筛选消息时间范围    private String name;            // （已注释）可扩展的认证用户名    private String passwd;          // （已注释）可扩展的认证密码    /**     * 构造函数 - 在创建采样器实例时调用     */    public KafkaGetterJmeter() {        LOG.debug(whoAmI() + "\tConstruct");    }    /**     * 辅助方法：返回当前线程和对象哈希码，便于调试并发问题     */    private String whoAmI() {        return Thread.currentThread().toString() +                "@" +                Integer.toHexString(hashCode());    }    /**     * 【核心方法1】定义 JMeter GUI 中显示的默认参数     * 用户可以在界面中修改这些参数值     */    @Override    public Arguments getDefaultParameters() {        Arguments params = new Arguments();        params.addArgument("kafka_host", "192.168.118.168:7483");        params.addArgument("topic", "xxx-device-property");        params.addArgument("searchStr", "p_xxx_meter");        params.addArgument("timeShift_mins", "1");        // 可扩展的认证参数（目前未启用）        // params.addArgument("name", "username");        // params.addArgument("passwd", "password");        return params;    }    /**     * 【核心方法2】初始化/前置处理     * 在每个线程启动时执行一次，用于加载参数并建立连接等资源     */    @Override    public void setupTest(JavaSamplerContext arg0) {        // 从上下文对象中读取用户在 JMeter 界面填写的参数值        this.kafka_host = arg0.getParameter("kafka_host");        this.topic = arg0.getParameter("topic");        this.searchStr = arg0.getParameter("searchStr");        this.timeShift_mins = arg0.getParameter("timeShift_mins");        // this.name = arg0.getParameter("name");        // this.passwd = arg0.getParameter("passwd");    }    /**     * 【核心方法3】实际执行采样逻辑     * 每个样本（循环/迭代）都会调用一次     * @return SampleResult 包含状态、响应数据、耗时等信息的对象     */    @Override    public SampleResult runTest(JavaSamplerContext arg0) {        SampleResult sr = new SampleResult();   // 创建结果对象        // 标记采样开始（用于计算响应时间）        sr.sampleStart();        // 创建 Kafka 消费者助手类（实际业务逻辑在此类中实现）        KafkaGetter kr = new KafkaGetter(this.kafka_host, this.topic, this.searchStr, this.timeShift_mins);        // 记录请求数据（会在 JMeter 的“查看结果树”中显示）        sr.setSamplerData("prepare get message from \nhost: " + this.kafka_host +                           ".\ntopic: " + this.topic +                           ".\nsearchStr: " + this.searchStr +                           ".\ntimeBefore: " + this.timeShift_mins);        try {            String res = null;            res = kr.recieve();   // 实际从 Kafka 拉取消息（推测内部已实现消费逻辑）            System.out.println("结果in jmeter返回:" + res.toString());   // 控制台输出，方便调试            // 判断返回的消息内容是否包含目标字符串            if (res.contains(this.searchStr)) {                // 采样成功                sr.setSuccessful(true);                sr.setResponseData(res, "utf-8");   // 设置响应数据            } else {                // 采样失败（例如消息未找到目标字符串）                sr.setSuccessful(false);                sr.setResponseData(res, "utf-8");            }        } catch (Exception e) {            e.printStackTrace();            sr.setSuccessful(false);            sr.setResponseData(e.toString(), "utf-8");        } finally {            // 标记采样结束（自动计算耗时）            sr.sampleEnd();        }        return sr;    }    // 以下 main 方法用于本地调试（已注释，可取消注释独立测试采样器逻辑）    /*    public static void main(String[] args) {        KafkaRecieverJmeter krj = new KafkaRecieverJmeter();        Arguments args1 = krj.getDefaultParameters();        JavaSamplerContext context = new JavaSamplerContext(args1);        krj.setupTest(context);        System.out.println(krj.runTest(context));    }    */}