夜雨聆风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 | |
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将依赖一并打包（或放入 JMeterlib目录）。在 JMeter 中添加
Java Request采样器，下拉选择你的类名。填写参数，添加监听器（查看结果树、聚合报告）运行测试。
四、代码关键点解读（结合例子）
sr.sampleStart()sr.sampleEnd() | |
sr.setSamplerData(...) | |
sr.setResponseData(res, "utf-8") | |
sr.setSuccessful(false) |
五、运行效果与断言示例
成功场景：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 中显示的默认参数* 用户可以在界面中修改这些参数值*/@Overridepublic 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】初始化/前置处理* 在每个线程启动时执行一次，用于加载参数并建立连接等资源*/@Overridepublic 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 包含状态、响应数据、耗时等信息的对象*/@Overridepublic 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));}*/}
