如何捕捉流经MQ -> App -> Kafka -> App -> Kafka的消息处理时间

Question

我的系统管道是"IBM队列-> Springboot app (消息格式化程序) --> Kafka topic-1 (消息队列) --> Springboot app (消息转换器)-> --> Kafka Top-2(消息队列)“。

我的要求是测试IBM中生成的消息的端到端处理时间，并通过管道进行处理。我有一个带有JMeter的JMeter负载测试脚本，它生成通过管道泵出的大量消息。目前还没有实现Dynatrace监控，但是这里使用了用于Kafka监控的汇合控制中心。

您能否建议一些可能的选项来度量Springboot应用程序处理的消息的个人和端到端处理时间(通过DevOps Jenkins管道部署消息格式化程序和消息转换器应用程序)？

Answer 1

理论上，您可以使用来自JMeter的卡夫卡客户代码从JSR223测试元件本身获取卡夫卡主题中挂起的消息的数量。

示例代码：

KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
consumer.subscribe(Collections.singletonList("test"));
while(true) {
    ConsumerRecords<String, String> records = consumer.poll(100);
    for (ConsumerRecord<String, String> record : records) {
        System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());

        // after each message, query the number of messages of the topic
        Set<TopicPartition> partitions = consumer.assignment();
        Map<TopicPartition, Long> offsets = consumer.endOffsets(partitions);
        for(TopicPartition partition : offsets.keySet()) {
            System.out.printf("partition %s is at %d\n", partition.topic(), offsets.get(partition));
        }
    }
}

这些值可以使用样本变量存储到速记中，并作为自定义图表绘制到JMeter HTML报表仪表板中。