【kafka】使用jmeter压测kafka并发性能-蒲公英云

【kafka】使用jmeter压测kafka并发性能

1、idea中开发kafka并发测试需要的扩展包

由于jmeter的jms模块并不支持kafka，因此需要扩展开发
引入jmeter

<dependency>
            <groupId>org.apache.jmeter</groupId>
            <artifactId>ApacheJMeter_core</artifactId>
            <version>5.4.1</version>
            <scope>provided</scope>
        </dependency>
        <dependency>
            <groupId>org.apache.jmeter</groupId>
            <artifactId>ApacheJMeter_java</artifactId>
            <version>5.4.1</version>
            <scope>provided</scope>
        </dependency>

编写向kafka发送消息的核心逻辑，继承AbstractJavaSamplerClient

public class JmeterTest extends AbstractJavaSamplerClient {
    private static final Logger log = LoggerFactory.getLogger(JmeterTest.class);
    private KafkaProducer producer;
    private Integer dataSize;
    private Integer dataNum;
    private String topic;
    // SampleResult类封装了从入口样本返回的各种信息
    private SampleResult results;
    /**
     * 创建生产者实例
    */
    private void init() {
        if (producer == null) {
            // ...
            Map<String, Object> kafkaProperites = ConfigUtil.getProducerParams(openSsl, "jmeter-test");
            producer = new KafkaProducer(kafkaProperites);
        }
    }
    @Override
    public Arguments getDefaultParameters() {
        // 声明定义Arguments类
        Arguments params = new Arguments();
        // 添加一个新参数
        params.addArgument("dataSize", "4608");
        params.addArgument("dataNum", "1");
        params.addArgument("topic", "test_topic_1");
        return params;
    }
    @Override
    public void setupTest(JavaSamplerContext jsc) {
        // 定义SampleResult类
        results = new SampleResult();
        // 以String形式获取指定参数的值，或者如果未指定该值，则返回指定的默认值
        dataSize = jsc.getIntParameter("dataSize", 4608);
        dataNum = jsc.getIntParameter("dataNum", 1);
        topic = jsc.getParameter("topic", "test_topic_1");
    }
    @Override
    public SampleResult runTest(JavaSamplerContext javaSamplerContext) {
        init();
        ProducerRecord<String, byte[]> record = getRecord();
        results.sampleStart();
        for (int i = 0; i < dataNum; i++) {
            producer.send(record);
        }
        results.sampleEnd();
        //由于kafka的错误基本上都是警告，没法通过异常捕获知道，这里直接返回成功了，真正是否成功还得通过日志定位， 待优化
        results.setSuccessful(true);
        results.setResponseData("send " + dataNum + " records to " + topic, StandardCharsets.UTF_8.name());
        producer.close();
        return results;
    }
    // 构造dataSize大小的随机数据
    private ProducerRecord<String, byte[]> getRecord() {
        byte[] payload = null;
        Random random = new Random(0);
        if (dataSize != null) {
            payload = new byte[dataSize];
            for (int i = 0; i < payload.length; ++i)
                payload[i] = (byte) (random.nextInt(26) + 65);
        }
        ProducerRecord<String, byte[]> record = new ProducerRecord<>(topic, payload);
        return record;
    }
}