Spring Kafka监听器性能监控：掌握消息处理耗时

import io.micrometer.core.instrument.MeterRegistry;
import io.micrometer.core.instrument.Timer;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.stereotype.Component;

import java.util.HashMap;
import java.util.List;
import java.util.concurrent.TimeUnit;

@Component
public class MyKafkaConsumer {

    private final MeterRegistry meterRegistry;
    // 定义一个Timer，用于记录消息的实际处理时间
    private final Timer messageProcessingTimer;

    public MyKafkaConsumer(MeterRegistry meterRegistry) {
        this.meterRegistry = meterRegistry;
        // 初始化自定义计时器，可以添加描述和百分位数等配置
        this.messageProcessingTimer = Timer.builder("kafka.listener.message.internal.processing.time")
                .description("Time taken to process messages within the Kafka listener's business logic")
                // 示例：发布P50, P95, P99百分位数，用于更细致的性能分析
                .publishPercentiles(0.5, 0.95, 0.99)
                .register(meterRegistry);
    }

    @KafkaListener(topics = "myTopic", groupId = "myGroup", autoStartup = "true", concurrency = "3")
    public void consumeAssignment(
            @Header(KafkaHeaders.RECEIVED_TOPIC) String topic,
            @Header(required = false, name = KafkaHeaders.BATCH_CONVERTED_HEADERS) List<HashMap<String, byte[]>> headers,
            @Header(required = false, name = KafkaHeaders.RECEIVED_PARTITION_ID) List<Integer> partitions,
            @Payload(required = false) List<String> messages) {

        long startTimeNanos = System.nanoTime(); // 记录消息处理逻辑开始时间

        try {
            // =========================================================
            // 这里是您的实际消息处理逻辑
            // 例如：解析消息、调用外部服务、数据库操作等
            // =========================================================
            System.out.println("Received " + messages.size() + " messages from topic: " + topic + ", partition: " + partitions);
            // 模拟一个耗时操作
            Thread.sleep(50 + (long) (Math.random() * 100));
            // messages.forEach(msg -> System.out.println("Processing: " + msg));

        } catch (Exception e) {
            // 异常处理逻辑
            System.err.println("Error processing messages in topic " + topic + ": " + e.getMessage());
            // 如果需要，可以在这里记录处理失败的自定义指标
            // 例如：meterRegistry.counter("kafka.listener.message.processing.failures", "topic", topic).increment();
        } finally {
            long endTimeNanos = System.nanoTime(); // 记录消息处理逻辑结束时间
            long durationNanos = endTimeNanos - startTimeNanos; // 计算持续时间

            // 记录消息处理时间到自定义计时器
            messageProcessingTimer.record(durationNanos, TimeUnit.NANOSECONDS);

            // 如果需要根据消息的特定属性（如topic, groupId）添加标签，
            // 可以动态构建Timer或使用更通用的Timer.Sample
            // Timer.builder("kafka.listener.message.internal.processing.time")
            //      .tag("topic", topic)
            //      .tag("groupId", "myGroup")
            //      .register(meterRegistry) // 注意：频繁注册新Timer可能影响性能，建议预先定义或使用tagging机制
            //      .record(durationNanos, TimeUnit.NANOSECONDS);
        }
    }
}

import io.micrometer.core.annotation.Timed;
// ... 其他导入

@Component
public class MyKafkaConsumer {
    // ... constructor

    @Timed(value = "kafka.listener.method.execution.time", description = "Time taken for the entire Kafka listener method execution")
    @KafkaListener(topics = "myTopic", groupId = "myGroup", autoStartup = "true", concurrency = "3")
    public void consumeAssignment(
            // ... parameters
    ) {
        // 您的消息处理逻辑
    }
}