Spring Kafka监听器性能监控：从内置指标到自定义处理时间测量

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-actuator</artifactId>
</dependency>
<dependency>
    <groupId>io.micrometer</groupId>
    <artifactId>micrometer-core</artifactId>
</dependency>
<!-- 如果使用Prometheus -->
<dependency>
    <groupId>io.micrometer</groupId>
    <artifactId>micrometer-registry-prometheus</artifactId>
</dependency>

management.endpoints.web.exposure.include=health,info,metrics,prometheus
management.metrics.tags.application=${spring.application.name}

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 MyKafkaListener {

    private final MeterRegistry meterRegistry;

    // 通过构造函数注入MeterRegistry
    public MyKafkaListener(MeterRegistry meterRegistry) {
        this.meterRegistry = 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 startTime = System.nanoTime(); // 记录业务逻辑开始时间
        String status = "success"; // 默认处理状态为成功

        try {
            // --- 核心业务处理逻辑开始 ---
            if (messages != null && !messages.isEmpty()) {
                System.out.println("开始处理批次消息，共 " + messages.size() + " 条，来自主题: " + topic);
                for (String message : messages) {
                    // 模拟单个消息处理，例如：
                    // 调用外部API、存储到数据库、执行复杂计算等
                    // System.out.println("正在处理消息: " + message);
                    // Thread.sleep(new Random().nextInt(50)); // 模拟耗时操作
                }
            }
            // --- 核心业务处理逻辑结束 ---

        } catch (Exception e) {
            status = "failure"; // 捕获到异常，将状态设为失败
            System.err.println("处理消息时发生错误，主题: " + topic + ", 错误: " + e.getMessage());
            // 同时可以记录一个错误计数器，用于统计失败次数
            meterRegistry.counter("kafka.listener.message.processing.errors",
                    "topic", topic, "groupId", "myGroup").increment();
        } finally {
            long endTime = System.nanoTime(); // 记录业务逻辑结束时间
            long durationNanos = endTime - startTime; // 计算处理耗时（纳秒）

            // 使用MeterRegistry记录处理时间，并添加动态标签
            // 标签可以帮助我们按不同维度（如主题、消费组、处理状态）分析性能
            Timer.builder("kafka.listener.message.processing.time") // 指标名称
                    .description("Kafka监听器中业务逻辑的实际处理时间") // 指标描述
                    .tags("topic", topic, "groupId", "myGroup", "status", status) // 动态标签
                    .register(meterRegistry) // 注册或获取已存在的Timer
                    .record(durationNanos, TimeUnit.NANOSECONDS); // 记录耗时
        }
    }
}

import io.micrometer.core.annotation.Timed;
import io.micrometer.core.instrument.MeterRegistry;
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;

@Component
public class MyKafkaListenerWithTimed {

    private final MeterRegistry meterRegistry;

    public MyKafkaListenerWithTimed(MeterRegistry meterRegistry) {
        this.meterRegistry = meterRegistry;
    }

    @Timed(value = "kafka.listener.method.execution", 
           description = "Kafka监听器方法的整体执行时间", 
           extraTags = {"listener.type", "batch"}, 
           histogram = true) // 启用直方图，可以获取分位数
    @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) {

        // 业务处理逻辑
        if (messages != null && !messages.isEmpty()) {
            System.out.println("通过@Timed注解监控，处理批次消息，共 " + messages.size() + " 条，来自主题: " + topic);
            // ... 实际业务逻辑 ...
        }
        // 注意：@Timed注解默认会捕获方法抛出的异常，并可能通过特定标签（如exception）进行标记
        // 如果需要区分成功/失败，可能需要结合try-catch块手动更新标签，或依赖框架的默认行为
    }
}