扩展JMeter能力——自定义取样器

本文介绍如何开发一个自定义的取样器元件，来实现向Kafka发送消息，也就是实现一个自定义的Java请求。

新建Maven项目

如何创建Maven项目可以参考，修改pom.xml指定JDK版本、添加依赖，其中开发自定义Java请求依赖ApacheJMeter_core、ApacheJMeter_java，发送Kafka消息依赖kafka-clients。

<?xml version="1.0" encoding="UTF-8"?> <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"> <modelVersion>4.0.0</modelVersion> <groupId>org.xin.jmeter</groupId> <artifactId>jmeter-plugins-kafka</artifactId> <version>1.0.0</version> <properties> <maven.compiler.source>1.8</maven.compiler.source> <maven.compiler.target>1.8</maven.compiler.target> <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding> <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding> <java.version>1.8</java.version> <jmeter.version>5.4.3</jmeter.version> </properties> <dependencies> <dependency> <groupId>org.apache.jmeter</groupId> <artifactId>ApacheJMeter_core</artifactId> <version>${jmeter.version}</version> </dependency> <dependency> <groupId>org.apache.jmeter</groupId> <artifactId>ApacheJMeter_java</artifactId> <version>${jmeter.version}</version> </dependency> <dependency> <groupId>org.apache.kafka</groupId> <artifactId>kafka-clients</artifactId> <version>2.8.1</version> </dependency> </dependencies> </project>自定义Java请求元件的代码结构

自定义Java请求需要继承字AbstractJavaSamplerClient这个类，共包含四个固定的方法：

getDefaultParameters：该方法用来设置请求的参数，本例至少需要设置brokers、topic、message三个参数。setupTest：类似于测试用例结构中的初始化操作，可以把读取参数值、初始化Kafka producer放到这里。runTest：类似于测试用例结构中的执行部分，这里就是真正的业务执行的地方，向Kafka发送消息的代码就写在这里。teardownTest：类似于测试用例结构中的环境清理部分，关闭Kafka producer的代码写在这里。完整代码及注释package org.xin.jmeter; import java.util.Properties; import org.apache.kafka.clients.producer.KafkaProducer; import org.apache.kafka.clients.producer.ProducerConfig; import org.apache.kafka.clients.producer.ProducerRecord; import org.apache.kafka.common.serialization.StringSerializer; 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.apache.log4j.LogManager; import org.apache.log4j.Logger; public SamplerKafkaSendMsg extends AbstractJavaSamplerClient { private static Logger logger = LogManager.getLogger(SamplerKafkaSendMsg.class.getName()); // brokers、topic、msg三个变量用来存储参数的输入值 private String brokers; private String topic; private String msg; private Properties props = new Properties(); KafkaProducer<String, String> producer; /* 设置请求参数 * brokers: kafka的服务地址，一般为ip:9092 * topic: 发送消息的topic名称 * message：要发送的消息内容 */ @Override public Arguments getDefaultParameters() { Arguments params = new Arguments(); // 添加三个参数broker,topic,message接受用户输入 params.addArgument("brokers", "示例：x.x.x.x:9092"); params.addArgument("topic", ""); params.addArgument("message", ""); return params; } @Override public void setupTest(JavaSamplerContext arg0) { // 获取请求参数brokers，topic设置的值 brokers = arg0.getParameter("brokers"); topic = arg0.getParameter("topic"); // 根据brokers参数初始化Kafka producer props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, brokers); props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class); props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class); this.producer = new KafkaProducer<>(props); } @Override public SampleResult runTest(JavaSamplerContext arg0) { // 读取message参数的值,并创建record msg = arg0.getParameter("message"); ProducerRecord<String, String> record = new ProducerRecord<>(topic, msg); // 请求消息设置为输入的消息 SampleResult result = new SampleResult(); result.setSamplerData(msg); try { // 请求开始启动 result.sampleStart(); // 发送kafka消息 this.producer.send(record); // 发送kafka消息不需要等待回应，直接设置为“消息发送成功！” result.setResponseData("消息发送成功!", "utf-8"); result.setDataType(SampleResult.TEXT); // 请求状态标识为成功，也就是查看结果树中Java请求为绿色 result.setSuccessful(true); } catch (Throwable e) { // 抛出异常情况直接设置“消息发送失败！” result.setResponseData("消息发送失败!", "utf-8"); // 请求状态标识为失败，也就是查看结果树中Java请求为红色 result.setSuccessful(false); e.printStackTrace(); } finally { // 请求结束 result.sampleEnd(); } return result; } @Override public void teardownTest(JavaSamplerContext arg0) { // 关闭Kafka producer this.producer.close(); } }打包

执行mvn clean install进行打包，target目录下会生成对应的jar包