jmeter插件开发—RocketMQ4.9

标签：插件 context producer sr private params RocketMQ4.9 import jmeter

上期是使用java测试了rocketmq中间件的测试，考虑到后期性能测试，准备封装为jmeter插件。参考TesterHome的文章

环境说明

JDK1.8.0，Jmeter 5.4.3，maven构建工具（使用JDK17时遇到过跟Jmeter设置的外观有冲突）

实现代码（pom.xml已贴在上期的文章最后）

以下为全部代码，可以自行调整相关参数，实现了普通send与oneway（单向）的发送方式

package com.qgc.maven;

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.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.SendStatus;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import com.alibaba.fastjson.JSONObject;
import java.io.UnsupportedEncodingException;


public class JmeterMqProducer extends AbstractJavaSamplerClient {
    //这里定义类变量（不能定义为static，会报错）
    private DefaultMQProducer producer;
    private String producerName;
    private String producerGroup;
    private String serverUrl;
    private String topic;
    private String tags;
    private String keys;
    private String body;
    private String delayTime;
    private String timeout;
    private String sendType;
    private long cur_time;
    private byte[] bodyBytes;
    private String orginData;
//    private final Logger log = LogManager.getLogger(JmeterMqProducer.class);

    //这里定义这里作为请求的前置处理
    @Override
    public void setupTest(JavaSamplerContext context) {
        serverUrl = context.getParameter("serverUrl");
        topic = context.getParameter("topic");
        tags = context.getParameter("tags");
        keys = context.getParameter("keys");
        body = context.getParameter("messageBody");
        bodyBytes = body.getBytes();
        producerName = context.getParameter("producerName");
        producerGroup = context.getParameter("producerGroup");
        timeout = context.getParameter("timeout");
        sendType = context.getParameter("sendType");
        delayTime = context.getParameter("delayTime");

        try {
            producer = getProducer(1);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

    }

    //这里自定义了一个Producer的单例方法
    public DefaultMQProducer getProducer(int type) throws InterruptedException {
        if (producer == null){
            producer = new DefaultMQProducer(producerName);
            if (type == 1){
                System.out.println("=======init producer =====");
                cur_time = System.currentTimeMillis();
            }else {
                System.out.println("=======runtest producer =====");
            }
            producer.setNamesrvAddr(serverUrl); //serverUrl
            // 设置超时时间
            producer.setSendMsgTimeout(Integer.parseInt(timeout));
        }
        try {
            producer.start();
        } catch (MQClientException e) {
            System.out.println("启动init！忽略");
        }
        return producer;
    }

    //这里是一个请求结束的后置处理
    @Override
    public void teardownTest(JavaSamplerContext context) {
        producer.shutdown();
    }

    //这里是一个请求主体执行部分
    @Override
    public SampleResult runTest(JavaSamplerContext context) {

        SampleResult sr = new SampleResult();
        sr.sampleStart();
        Message msg = null;
        try {
            msg = new Message(topic, tags, keys, body.getBytes(RemotingHelper.DEFAULT_CHARSET));
        } catch (UnsupportedEncodingException e) {
            e.printStackTrace();
        }

        try {
            JSONObject json = new JSONObject();
            json.put("serverUrl", serverUrl);
            json.put("sendType", sendType);
            json.put("topic", topic);
            json.put("tags", tags);
            json.put("keys", keys);
            json.put("producerName", producerName);
            sr.setRequestHeaders(json.toString());
            sr.setSamplerData(body);
            if ("oneWay".equals(sendType)){
                producer.sendOneway(msg);
                sr.setResponseData("{\"code\": \"200\", \"SendStatus\" : \"success\"}","utf-8");
            }else {
                SendResult sendResult = producer.send(msg);
                sr.setResponseData(sendResult.toString(),"utf-8");
                if(sendResult !=null || sendResult.getSendStatus() == SendStatus.SEND_OK){
                    sr.setResponseData("{\"code\": \"200\", \"SendStatus\" : \"" + sendResult.getSendStatus() + "\",\"MsgId\" : \"" + sendResult.getMsgId() +"\"}","utf-8");
                }
                else {
                    System.err.println(sendResult);
                    sr.setResponseData("{\"code\" : \"500\", \"msg\": \"失败\", \"Error\": \""+ sendResult.toString() +"\"}","utf-8");
                }
            }
        }catch (Exception e){
            e.printStackTrace();
            sr.setResponseData("{\"code\" : \"501\", \"msg\": \"其他失败\", \"Error\": "+ e +"}","utf-8");
            producer.shutdown();

        }
        try {
            sr.setDataType(SampleResult.TEXT);
            sr.setSuccessful(true);
        }catch(Exception e){
            sr.setSuccessful(false);
            e.printStackTrace();
        }
        finally {
            sr.sampleEnd();
        }
        return sr;

    }

    // 给参数填充默认值
    @Override
    public Arguments getDefaultParameters() {
        Arguments params = new Arguments();
        params.addArgument("serverUrl", "18.162.xxx.xx:9876");
        params.addArgument("topic", "qgc_TopicTest");
        params.addArgument("tags", "qgc_TagA");
        params.addArgument("keys", "test_key");
        params.addArgument("messageBody", "test_body");
        params.addArgument("producerName", "test_producerName");
        params.addArgument("producerGroup", "producerGroup");
        params.addArgument("timeout", "6000");
        params.addArgument("sendType", "oneWay");
        params.addArgument("delayTime", "100");

        return params;
    }
}