Spring boot 2.3.4-在/acture/Prometheus中不可见Kafka度量

Question

我有一个Spring Boot应用程序（版本2.3.4)，我正在使用@kafkalistener来消费记录。我还使用执行器和千分尺（版本1.5.5）来进行度量。

问题是我看不到/acture/prometheus中的Kafka度量。我正在使用以下依赖项：

'org.springframework.boot' version '2.3.4.RELEASE'
implementation group: 'org.springframework.kafka', name: 'spring-kafka', version: '2.5.10.RELEASE'
implementation group: 'org.apache.kafka', name: 'kafka-clients', version: '2.5.1'

management:
  server:
    port: 9091
  endpoints:
    web:
      exposure:
        include: "*"
  endpoint:
    health:
      probes:
        enabled: true

spring:
  jmx:
    enabled: true

请注意，当我使用默认KafkaTemplate时，这些度量是可见的，但当试图创建自定义KafkaTemplate时，这些度量会消失：

@SpringBootApplication
public class Application {
    public static void main(String[] args) {
        SpringApplication.run(Application.class, args);
    }

    @Bean
    public ProducerFactory<String, String> customProducerFactory() {
        Map<String, Object> configProps = new HashMap<>();
        configProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "127.0.0.1:9092");
        configProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, Serdes.String().serializer().getClass().getName());
        configProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, Serdes.String().serializer().getClass().getName());
        return new DefaultKafkaProducerFactory<>(configProps);
    }

    @Bean
    public KafkaTemplate<String, String> customProducer() {
        return new KafkaTemplate<>(customProducerFactory());
    }

    @KafkaListener(id = "test", topics = "test_topic")
    public void listen(String in) {
        System.out.println(in);
    }

    @Bean
    public NewTopic topic() {
        return TopicBuilder.name("test_topic").partitions(1).replicas(1).build();
    }


    @Bean
    public ApplicationRunner runner(KafkaTemplate<String, String> template) {
        return args -> {
            template.send("test_topic", "foo");
        };
    }
}

Answer 1

解决方案是在自定义Kafkatemplate中添加一个侦听器：

@Bean
    public ProducerFactory<String, String> customProducerFactory() {
        Map<String, Object> configProps = new HashMap<>();
        configProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "127.0.0.1:9092");
        configProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, Serdes.String().serializer().getClass().getName());
        configProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, Serdes.String().serializer().getClass().getName());
        DefaultKafkaProducerFactory<String, String> producerFactory = new DefaultKafkaProducerFactory<>(configProps);
        producerFactory.addListener(new MicrometerProducerListener<>(meterRegistry));
    }

@Bean
public KafkaTemplate<String, String> customProducer() {
    return new KafkaTemplate<>(customProducerFactory());
}