负载测试kafka消费者

（我正在编辑问题，因为我认为它不够清楚）

如何对我的 kafka 消费者进行负载测试？ 我看过很多关于 apache kafka 负载测试的文章，但没有一篇关于消费者负载测试的文章。 对于前。我在jsr223中编写了jmeter kafka生产者测试计划，将事件发送到kafka主题。 我的kafka spring boot消费者监听这个主题，消费事件并开始处理它。 Jmeter 只向我显示有关该主题的生产者的报告，仅此而已。 有没有更好的方法来加载测试 spring boot kafka 消费者应用程序并获取报告？ 从消耗时间到进程结束？

我相信kafka-consumer-perf-test.sh脚本与每个 Kafka 安装一起使用，它应该足以满足您的用例，请参阅性能测试 Kafka Wiki 页面 https://cwiki.apache.org/confluence/display/KAFKA/Performance+testing更多细节。

该脚本可以使用 JMeter 集成操作系统进程采样器 https://jmeter.apache.org/usermanual/component_reference.html#OS_Process_Sampler

如果您正在谈论使用 JMeter 读取来自损坏的消息 - 任何人都可以完成JSR223测试元件 https://jmeter.apache.org/usermanual/best-practices.html#jsr223

1. Obtain 卡夫卡Java客户端 https://docs.confluent.io/clients-kafka-java/current/overview.html

2. 将库和依赖项放入JMeter 类路径 https://jmeter.apache.org/usermanual/get-started.html#classpath

3. 重新启动 JMeter 以获取库

4. 将实现从 Kafka 主题读取消息的代码放入 JSR223 Sampler，示例片段：

   props.put('bootstrap.servers', '192.168.99.100:9092')
   props.put('group.id', 'foo')
   props.put('enable.auto.commit', 'true')
   props.put('auto.commit.interval.ms', '1000')
   props.put('session.timeout.ms', '30000')
   props.put('key.deserializer',
           'org.apache.kafka.common.serialization.StringDeserializer')
   props.put('value.deserializer',
           'org.apache.kafka.common.serialization.StringDeserializer')
   def consumer = new org.apache.kafka.clients.consumer.KafkaConsumer<String, String>(props)
   def topic = 'sometopic'
   
   consumer.subscribe(Arrays.asList(topic))
   log.info('Subscribed to topic ' + topic)
   
   while (true) {
       def records = consumer.poll(100)
       records.each { record ->
           log.info('Received message: ' + record.value())
       }
   }
   

更多信息：

• Apache Kafka - 如何使用 JMeter 进行负载测试 https://www.blazemeter.com/blog/apache-kafka-how-to-load-test-with-jmeter