devbox
秋刀鱼在做梦
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

Jmeter Kafka Producer Sampler_jmeter kafka插件

作者：秋刀鱼在做梦 | 2024-07-14 11:08:23

赞

踩

jmeter kafka插件

目录

前言

Jmeter Kafka插件开发之Sampler篇

插件界面如下：

主要有3个区域：

broker地址：Kafka Broker地址

topic名称：Kafka topic

body：Kafka 报文区域

项目说明：

实现方式:

UI代码

业务代码：

pom文件

界面如下：

 测试验证​编辑

前言

之前写了一个Jmeter kafka 插件是基于公司对kafka二次封装写的，这次基于原生kafka写一个插件，如下，废话不多说，中心思想是一样的，可以参考我上篇文章

Jmeter Kafka插件开发之Sampler篇

插件界面如下：

主要有3个区域：

broker地址：Kafka Broker地址

用于填写Kafka broker地址，一般为IP+端口号如：127.0.0.1:1234

topic名称：Kafka topic

用于填写Kafka主题名称，如：test

body：Kafka 报文区域

填写Kafka发送的报文，如：test123

项目说明：

需要建一个gui包和一个sampler包,Jmeter在加载插件的时候会加载gui里面的类，可以参考JMeter源码的包命名方式，以下是参考的项目目录结构：

实现方式:

  • 继承AbstractSamplerGui方法
  • 重写createTestElement()方法：此方法应创建TestElement类的新实例，然后将其传递给modifyTestElement（TestElement） 方法
  • 重写configure()方法：通过调用此方法，可以用测试元素对象的内容初始化新创建的GUI组件
  • 重写modifyTestElement()方法：将GUI元素中的数据移动到TestElement中
  • 主要重写2、3、4这3个方法，其他的包括重写clearGui()用于恢复GUI到初始状态，getStaticLabel()返回插件名称
  • 引入ApacheJMeter_java、ApacheJMeter_core以及需要使用的其他库
  • GUI类主要负责输入数据然后将数据传送到TestElement中，然后sampler类拿到数据进行业务逻辑处理。
     

UI代码

  1. package com.jmeter.gui;
  2.  
  3. import com.jmeter.sampler.KafkaSampler;
  4. import org.apache.jmeter.gui.util.JSyntaxTextArea;
  5. import org.apache.jmeter.gui.util.JTextScrollPane;
  6. import org.apache.jmeter.gui.util.VerticalPanel;
  7. import org.apache.jmeter.samplers.gui.AbstractSamplerGui;
  8. import org.apache.jmeter.testelement.TestElement;
  9. import org.apache.jorphan.gui.layout.VerticalLayout;
  10. import org.slf4j.Logger;
  11. import org.slf4j.LoggerFactory;
  12.  
  13. import javax.swing.*;
  14. import java.awt.*;
  15. import java.awt.event.ActionEvent;
  16. import java.awt.event.ActionListener;
  17.  
  18. public class KafkaSamplerUI extends AbstractSamplerGui implements ActionListener {
  19.     private static final long serialVersionUID = 1L;
  20.     private static final Logger log = LoggerFactory.getLogger(KafkaSamplerUI.class);
  21.  
  22.     private JLabel brokerAddressLable;
  23.     private JTextField brokerAddressField;
  24.  
  25. //    private JLabel clusterNameLabel;
  26. //    private JTextField clusterNameField;
  27.  
  28.     private JLabel topicNameLabel;
  29.     private JTextField topicNameField;
  30.  
  31. //    private JLabel chekCodeLabel;
  32. //    private JTextField chekCodeField;
  33.  
  34.  
  35.     private JSyntaxTextArea textMessage = new JSyntaxTextArea(40, 50);
  36.  
  37.     private JLabel textArea = new JLabel();
  38.     private JTextScrollPane textPanel = new JTextScrollPane(textMessage);
  39.  
  40. //    private JLabel kafkaLabel;
  41. //    private JRadioButton readWriteMode;
  42. //    private JRadioButton writeMode;
  43. //    private JRadioButton readMode;
  44.  
  45.  
  46.     Box brokerAddressPanel, topicNamePanel, kafkaPanel;
  47. //    static Box filterPanel, readNumPanel;
  48.  
  49.     JPanel mainPanel = new VerticalPanel();
  50.     JPanel extendPanel = new VerticalPanel();
  51.     static JPanel contentPanel;
  52.  
  53.  
  54.     public KafkaSamplerUI() {
  55.         super();
  56.         this.init();
  57.     }
  58.  
  59.     //创建消息输入框
  60.     private JPanel createContentPanel() {
  61.  
  62.         JPanel ContentPanel = new VerticalPanel();
  63.         JPanel messageContentPanel = new JPanel(new BorderLayout());
  64.         messageContentPanel.add(this.textArea, BorderLayout.NORTH);
  65.         messageContentPanel.add(this.textPanel, BorderLayout.CENTER);
  66.         ContentPanel.add(messageContentPanel);
  67.         ContentPanel.setBorder(BorderFactory.createTitledBorder(BorderFactory.createLineBorder(Color.gray), "写入数据"));
  68.  
  69.         return ContentPanel;
  70.     }
  71.  
  72.     private Box createTextFiledPanle(String name) {
  73.  
  74.         JLabel jLabel = new JLabel(name);
  75.         JTextField jTextField = new JTextField(6);
  76.  
  77.         Box box = Box.createHorizontalBox();
  78.         box.add(jLabel);
  79.         box.add(jTextField);
  80.         return box;
  81.     }
  82.  
  83.  
  84.     private void init() {
  85.         log.info("初始化UI界面");
  86.  
  87.         brokerAddressLable = new JLabel("broker endpoint(s)地址：");
  88.         brokerAddressField = new JTextField(6);
  89.  
  90.  
  91.         topicNameLabel = new JLabel("主题名称：");
  92.         topicNameField = new JTextField(6);
  93.  
  94.  
  95. //        kafkaLabel = new JLabel("kafka：");
  96. //
  97. //        readWriteMode = new JRadioButton("读写模式");
  98. //        writeMode = new JRadioButton("写模式");
  99. //        readMode = new JRadioButton("读模式");
  100.  
  101. //        kafkaPanel = Box.createHorizontalBox();
  102. //
  103. //        kafkaPanel.add(kafkaLabel);
  104. //        kafkaPanel.add(readWriteMode);
  105. //        kafkaPanel.add(writeMode);
  106. //        kafkaPanel.add(readMode);
  107.  
  108.         //设置默认读写方式
  109. //        writeMode.setSelected(true);
  110. //        readWriteMode.setSelected(false);
  111. //        readMode.setSelected(false);
  112.  
  113.         contentPanel = createContentPanel();
  114.         contentPanel.setVisible(true);
  115. //        filterPanel = createTextFiledPanle("筛选条件");
  116. //
  117. //        readNumPanel = createTextFiledPanle("读取条数");
  118.         setLayout(new VerticalLayout(5, VerticalLayout.BOTH, VerticalLayout.TOP));
  119.  
  120.         setBorder(makeBorder());
  121.         add(makeTitlePanel());
  122.  
  123.         brokerAddressPanel = Box.createHorizontalBox();
  124.  
  125.         brokerAddressPanel.add(brokerAddressLable);
  126.         brokerAddressPanel.add(brokerAddressField);
  127.  
  128.  
  129.         topicNamePanel = Box.createHorizontalBox();
  130.  
  131.         topicNamePanel.add(topicNameLabel);
  132.         topicNamePanel.add(topicNameField);
  133.  
  134.  
  135.         mainPanel.add(brokerAddressPanel);
  136.         mainPanel.add(topicNamePanel);
  137. //        mainPanel.add(kafkaPanel);
  138.  
  139.         extendPanel.add(contentPanel);
  140. //        extendPanel.add(filterPanel);
  141. //        extendPanel.add(readNumPanel);
  142.         extendPanel.setVisible(true);
  143.         add(mainPanel, BorderLayout.CENTER);
  144.         add(extendPanel, BorderLayout.CENTER);
  145.  
  146.  
  147.     }
  148.  
  149.     /**
  150.      * 此方法应创建TestElement类的新实例，然后将其传递给modifyTestElement（TestElement） 方法
  151.      *
  152.      * @return
  153.      */
  154.     public TestElement createTestElement() {
  155.         KafkaSampler sampler = new KafkaSampler();
  156.         modifyTestElement((TestElement) sampler);
  157.         return (TestElement) sampler;
  158.     }
  159.  
  160.     @Override
  161.     public void clearGui() {
  162.         super.clearGui();
  163.         brokerAddressField.setText("");
  164.         topicNameField.setText("");
  165.  
  166.  
  167.     }
  168.  
  169.     /**
  170.      * 一定要调用super.configure（e）。这将为您填充一些数据，例如元素的名称。
  171.      * 使用此方法将数据设置到GUI元素中
  172.      * 通过调用此方法，可以用测试元素对象的内容初始化新创建的GUI组件。组件负责查询测试元素对象，以获取要在其GUI中显示的相关信息。
  173.      *
  174.      * @param element
  175.      */
  176.     @Override
  177.     public void configure(TestElement element) {
  178.         super.configure(element);
  179.  
  180.         KafkaSampler sampler = (KafkaSampler) element;
  181.         brokerAddressField.setText(sampler.getBrokerraddress());
  182.         log.info("设置broker集群地址为:" + sampler.getBrokerraddress());
  183.  
  184.  
  185.         topicNameField.setText(sampler.getTopicName());
  186.         log.info("设置集群主题为:" + sampler.getTopicName());
  187.  
  188.  
  189.         textMessage.setText(sampler.getMessage());
  190.         log.info("设置发送消息为:" + sampler.getMessage());
  191.  
  192.     }
  193.  
  194.  
  195.     @Override
  196.     public String getStaticLabel() {
  197.         return "Kafka Sampler";
  198.     }
  199.  
  200.     public String getLabelResource() {
  201.         throw new IllegalStateException("This shouldn't be called");
  202.     }
  203.  
  204.     /**
  205.      * 将GUI元素中的数据移动到TestElement,
  206.      *
  207.      * @param e
  208.      */
  209.  
  210.     public void modifyTestElement(TestElement e) {
  211.         e.clear();
  212.         // 调用super.configureTestElement（e）。这将处理一些默认数据
  213.         configureTestElement(e);
  214.  
  215.         KafkaSampler sampler = new KafkaSampler();
  216.  
  217.         ((KafkaSampler) e).setBrokerAddress(brokerAddressField.getText());
  218. //        log.info("填入的broker集群地址为:" + this.brokerAddressField.getText());
  219.  
  220.  
  221.         ((KafkaSampler) e).setTopicName(topicNameField.getText());
  222. //        log.info("填入的主题名称地址为:" + this.topicNameField.getText());
  223.  
  224.         ((KafkaSampler) e).setMessage(textMessage.getText());
  225. //        log.info("填入消息为:" + this.textMessage.getText());
  226.  
  227.  
  228.         //添加监听
  229. //        readWriteMode.addActionListener(this);
  230. //        readMode.addActionListener(this);
  231. //        writeMode.addActionListener(this);
  232.  
  233.  
  234.     }
  235.  
  236.     public void actionPerformed(ActionEvent e) {
  237. //        if (e.getSource() == readWriteMode) {
  238. //            readWriteMode.setSelected(true);
  239. //            readMode.setSelected(false);
  240. //            writeMode.setSelected(false);
  241. //            log.info("读写模式新增contentPanel");
  242. //
  243. //            contentPanel.setVisible(true);
  244. //
  245. //            log.info("读写模式新增filterPanel");
  246. //
  247. //            filterPanel.setVisible(true);
  248. //
  249. //            log.info("读写模式新增readNumPanel");
  250. //
  251. //            readNumPanel.setVisible(true);
  252. //
  253. //
  254. //            extendPanel.setVisible(true);
  255. //            updateUI();
  256. //            repaint();
  257. //
  258. //        }
  259. //        if (e.getSource() == writeMode) {
  260. //
  261. //            writeMode.setSelected(true);
  262. //            readWriteMode.setSelected(false);
  263. //            readMode.setSelected(false);
  264. //
  265. //            log.info("写模式新增contentPanel");
  266. //            contentPanel.setVisible(true);
  267. //
  268. //            filterPanel.setVisible(false);
  269. //
  270. //            readNumPanel.setVisible(false);
  271. //
  272. //            extendPanel.setVisible(false);
  273. //            extendPanel.setVisible(true);
  274. //
  275. //            updateUI();
  276. //            repaint();
  277. //
  278. //
  279. //        }
  280. //        if (e.getSource() == readMode) {
  281. //            readMode.setSelected(true);
  282. //            writeMode.setSelected(false);
  283. //            readWriteMode.setSelected(false);
  284. //            log.info("读模式新增filterPanel");
  285. //
  286. //            contentPanel.setVisible(false);
  287. //            filterPanel.setVisible(true);
  288. //
  289. //            log.info("读模式新增readNumPanel");
  290. //            readNumPanel.setVisible(true);
  291. //
  292. //            extendPanel.setVisible(true);
  293. //
  294. //            updateUI();
  295. //            repaint();
  296. //
  297. //        }
  298.  
  299.     }
  300. }
  301.  
  302.

业务代码：

  1. package com.jmeter.kafka;
  2.  
  3. public interface  IKafkaProducer {
  4.     public void close();
  5.  
  6.     public void sendString(String topicName, String message);
  7. }
  1. package com.jmeter.kafka;
  2.  
  3. import org.apache.kafka.clients.producer.ProducerRecord;
  4.  
  5. import java.util.Properties;
  6.  
  7. public class ProduceConfig extends Properties{
  8.      Properties kafkaProps = new Properties();
  9.  
  10.  
  11.     public ProduceConfig(String brokerraddress) {
  12.         kafkaProps.put("bootstrap.servers", brokerraddress);
  13.         kafkaProps.put("key.serializer",
  14.                 "org.apache.kafka.common.serialization.StringSerializer");
  15.         kafkaProps.put("value.serializer",
  16.                 "org.apache.kafka.common.serialization.StringSerializer");
  17.  
  18.     }
  19.  
  20.  
  21. }
  1. package com.jmeter.kafka;
  2.  
  3. import org.apache.kafka.clients.producer.KafkaProducer;
  4. import org.apache.kafka.clients.producer.ProducerRecord;
  5.  
  6. public class ProducerPool  extends ProduceConfig implements IKafkaProducer {
  7.     KafkaProducer<String, String> producer;
  8.  
  9.     public ProducerPool(String brokerraddress) {
  10.         super(brokerraddress);
  11.     }
  12.  
  13.     public void close() {
  14.     if (producer!=null){
  15.         producer.close();
  16.     }
  17.     }
  18.  
  19.     public void sendString(String topicName, String message) {
  20.         ProducerRecord<String,String> record = new ProducerRecord<String,String>(topicName,message); // 主题，key,value
  21.  
  22.         producer = new KafkaProducer<String, String>(kafkaProps);
  23.         producer.send(record);
  24.     }
  25.     }
  26.  
  1. package com.jmeter.sampler;
  2.  
  3. import com.jmeter.kafka.IKafkaProducer;
  4. import com.jmeter.kafka.ProducerPool;
  5. import org.apache.jmeter.samplers.AbstractSampler;
  6. import org.apache.jmeter.samplers.Entry;
  7. import org.apache.jmeter.samplers.SampleResult;
  8. import org.apache.jmeter.testelement.TestElement;
  9.  
  10.  
  11. public class KafkaSampler extends AbstractSampler implements TestElement {
  12.  
  13.  
  14.     public static final String BROKERRADDRESS = "KafkaSampler.brokerAddress";
  15.  
  16.     public static final String TOPICNAME = "SFKafkaSampler.topicName";
  17.  
  18.     public static final String MESSAGE = "SFKafkaSampler.message";
  19.  
  20.     public void setBrokerAddress(String clusterAddress) {
  21.  
  22.         setProperty(BROKERRADDRESS, clusterAddress);
  23.  
  24.     }
  25.  
  26.  
  27.     public void setTopicName(String topicName) {
  28.  
  29.         setProperty(TOPICNAME, topicName);
  30.  
  31.     }
  32.  
  33.  
  34.     public void setMessage(String message) {
  35.  
  36.  
  37.         setProperty(MESSAGE, message);
  38.  
  39.     }
  40.  
  41.  
  42.     public String getMessage() {
  43.  
  44.         return getPropertyAsString(MESSAGE);
  45.  
  46.     }
  47.  
  48.  
  49.     public String getBrokerraddress() {
  50.  
  51.         return getPropertyAsString(BROKERRADDRESS);
  52.  
  53.     }
  54.  
  55.  
  56.     public String getTopicName() {
  57.  
  58.         return getPropertyAsString(TOPICNAME);
  59.  
  60.     }
  61.  
  62.  
  63.     public KafkaSampler() {
  64.  
  65.         setName("Kafka Sampler");
  66.  
  67.     }
  68.  
  69.  
  70.     public SampleResult sample(Entry entry) {
  71.  
  72.         SampleResult result = new SampleResult();
  73.  
  74.         result.setSampleLabel(getName());
  75.  
  76.         try {
  77.  
  78.             result.sampleStart();
  79.  
  80.             //写入业务数据
  81.  
  82.  
  83.  
  84.             IKafkaProducer kafkaProducer = new ProducerPool(getBrokerraddress());
  85.  
  86.  
  87.             kafkaProducer.sendString(getTopicName(), getMessage());
  88.  
  89.  
  90.  
  91.  
  92.             result.setSamplerData("请求集群地址为:" + getBrokerraddress() + "\n" + "请求集群主题为:" + getTopicName() + "\n" + "\n" + "请求消息为:" + getMessage());
  93.  
  94.  
  95.             result.setResponseData(getMessage(), "utf-8");
  96.  
  97.  
  98.             result.sampleEnd();
  99.  
  100.             result.setSuccessful(true);
  101.  
  102.             result.setResponseCodeOK();
  103.  
  104.  
  105.         } catch (Exception e) {
  106.  
  107.             result.setSamplerData("请求集群地址为:" + getBrokerraddress() + "\n" + "请求集群主题为:" + getTopicName() + "\n" + "请求消息为:" + getMessage());
  108.  
  109.  
  110.             result.sampleEnd(); // stop stopwatch
  111.  
  112.             result.setSuccessful(false);
  113.  
  114.             result.setResponseMessage("Exception: " + e);
  115.  
  116.             // get stack trace as a String to return as document data
  117.  
  118.             java.io.StringWriter stringWriter = new java.io.StringWriter();
  119.  
  120.             e.printStackTrace(new java.io.PrintWriter(stringWriter));
  121.  
  122.             result.setResponseData(stringWriter.toString(), null);
  123.  
  124.             result.setDataType(org.apache.jmeter.samplers.SampleResult.TEXT);
  125.  
  126.             result.setResponseCode("FAILED");
  127.  
  128.         }
  129.  
  130.         return result;
  131.  
  132.     }
  133.  
  134.  
  135. }
  136.  
  137.  
  138.

pom文件

  1. <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
  2.   xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
  3.   <modelVersion>4.0.0</modelVersion>
  4.  
  5.   <groupId>org.example</groupId>
  6.   <artifactId>JmeterKafkaSampler</artifactId>
  7.   <version>1.0-SNAPSHOT</version>
  8.   <packaging>jar</packaging>
  9.  
  10.   <name>JmeterKafkaSampler</name>
  11.   <url>http://maven.apache.org</url>
  12.  
  13.   <properties>
  14.     <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
  15.   </properties>
  16.  
  17.   <dependencies>
  18.     <!-- https://mvnrepository.com/artifact/org.apache.jmeter/ApacheJMeter_java -->
  19.     <dependency>
  20.       <groupId>org.apache.jmeter</groupId>
  21.       <artifactId>ApacheJMeter_java</artifactId>
  22.       <version>5.5</version>
  23.     </dependency>
  24.     <!-- https://mvnrepository.com/artifact/org.apache.jmeter/ApacheJMeter_core -->
  25.     <dependency>
  26.       <groupId>org.apache.jmeter</groupId>
  27.       <artifactId>ApacheJMeter_core</artifactId>
  28.       <version>5.5</version>
  29.     </dependency>
  30.     <dependency>
  31.       <groupId>org.apache.kafka</groupId>
  32.       <artifactId>kafka-clients</artifactId>
  33.       <version>3.2.1</version>
  34.     </dependency>
  35.     <dependency>
  36.       <groupId>junit</groupId>
  37.       <artifactId>junit</artifactId>
  38.       <version>4.12</version>
  39.       <scope>test</scope>
  40.     </dependency>
  41.   </dependencies>
  42.   <build>
  43.  
  44.   <plugins>
  45.     <plugin>
  46.  
  47.       <artifactId> maven-assembly-plugin </artifactId>
  48.  
  49.       <configuration>
  50.  
  51.         <descriptorRefs>
  52.  
  53.           <descriptorRef>jar-with-dependencies</descriptorRef>
  54.  
  55.         </descriptorRefs>
  56.  
  57.         <archive>
  58.  
  59.           <manifest>
  60.  
  61.             <mainClass>com.xx.util.GenerateSortCode</mainClass>
  62.  
  63.           </manifest>
  64.  
  65.         </archive>
  66.  
  67.       </configuration>
  68.  
  69.       <executions>
  70.  
  71.         <execution>
  72.  
  73.           <id>make-assembly</id>
  74.  
  75.           <phase>package</phase>
  76.  
  77.           <goals>
  78.  
  79.             <goal>single</goal>
  80.  
  81.           </goals>
  82.  
  83.         </execution>
  84.  
  85.       </executions>
  86.  
  87.     </plugin>
  88.   </plugins>
  89.   </build>
  90.  
  91. </project>

然后用maven打完包就可以放到Jmeter/lib/ext目录下，重启JMeter后新建Sampler可以查看到自己定义的Kafka Sampler如下图:

界面如下：

 测试验证

经测试数据是成功写入到对应的kafka主题中了，到此就完成了Kafka插件的写入功能开发

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/秋刀鱼在做梦/article/detail/824522
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号