问题

I use KafkaListener to get a message from kafka, but I don't get the spring_kafka_listener metrics

使用KafkaListener来从Kafka获取消息，但我没有得到spring_kafka_listener指标

registering a metric with the name spring_kaf_ka_listener did not give results, an error message will appear that such a metric has already been registered

使用名称spring_kaf_ka_listener注册指标没有产生结果，将出现已经注册了这样的指标的错误消息

the addition did not give any result
cf.addListener(new MicrometerConsumerListener<>(new SimpleMeterRegistry()));

添加没有产生任何结果
cf.addListener(new MicrometerConsumerListener<>(new SimpleMeterRegistry()));

@Configuration
@EnableKafka
public class KafkaConsumerConfig {

    @Autowired
    private KafkaProperties kafkaProperties;

    @Bean
    public Map<String, Object> consumerConfigs() {
        Map<String, Object> props = new HashMap<>(kafkaProperties.buildConsumerProperties());
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        props.put(ConsumerConfig.GROUP_ID_CONFIG, kafkaProperties.getConsumer().getGroupId());
        return props;
    }

    @Bean
    public ConsumerFactory<String, String> consumerFactory() {
        final StringDeserializer jsonDeserializer = new StringDeserializer();
        ConsumerFactory<String, String> cf = new DefaultKafkaConsumerFactory<>(
            kafkaProperties.buildConsumerProperties(),
            new StringDeserializer(), jsonDeserializer
        );
        //I tried to add it but it didn't give any result
        cf.addListener(new MicrometerConsumerListener<>(new SimpleMeterRegistry()));
        return cf;
    }

    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory() {
        ConcurrentKafkaListenerContainerFactory<String, String> factory =
            new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory());
        factory.setBatchListener(false);

        return factory;
    }

    @Bean
    public DefaultKafkaHeaderMapper headerMapper() {
        return new DefaultKafkaHeaderMapper();
    }
}

@Configuration
@EnableKafka
public class KafkaConsumerConfig {
@Autowired
private KafkaProperties kafkaProperties;
@Bean
public Map<String, Object> consumerConfigs() {
Map<String, Object> props = new HashMap<>(kafkaProperties.buildConsumerProperties());
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
props.put(ConsumerConfig.GROUP_ID_CONFIG, kafkaProperties.getConsumer().getGroupId());
return props;
}
@Bean
public ConsumerFactory<String, String> consumerFactory() {
final StringDeserializer jsonDeserializer = new StringDeserializer();
ConsumerFactory<String,String> cf = new DefaultKafkaConsumerFactory<>(
kafkaProperties.buildConsumerProperties(),
new StringDeserializer(), jsonDeserializer
);
//I tried to add it but it didn't give any result
cf.addListener(new MicrometerConsumerListener<>(new SimpleMeterRegistry()));//
return cf;
}
@Bean
public ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory() {
ConcurrentKafkaListenerContainerFactory<String, String> factory =
new ConcurrentKafkaListenerContainerFactory<>();
factory.setConsumerFactory(consumerFactory());
factory.setBatchListener(false);
return factory;
}
@Bean
public DefaultKafkaHeaderMapper headerMapper() {
return new DefaultKafkaHeaderMapper();
}
}

答案1

得分: 0

消费者监听器启用了本机Kafka指标，spring_kafka_listener指标由micrometerEnabled容器属性控制，默认情况下为true。

https://docs.spring.io/spring-kafka/docs/current/reference/html/#micrometer

只有在类路径上存在Micrometer时才会捕获这些指标。

如果您无法弄清楚，请提供一个MCRE，以便我们看到问题出在哪里。