Spring boot Kafka 指标配置

Question

我们有一个使用 spring-kafka 的 spring boot 微服务。有几个 Kafka 侦听器和生产者，Kafka 相关指标已成功呈现在执行器端点。但是当添加另一个带有自定义指标的 maven 依赖项时，spring boot 提供的 Kafka 相关指标被删除。我怀疑这与引入新的 MeterRegistry bean 有关，但是使用与 spring-boot-actuator-autoconfigure spring-boot 模块中的 NoOpMeterRegistryConfiguration 相同的 bean 定义覆盖它没有帮助。我也试过调试 KafkaMetricsAutoConfiguration，看起来所有 bean 都已成功实例化。有更多 spring-boot-actuator-autoconfigure 经验的人可以帮助我理解 Kafka 相关指标的默认配置吗？

新的maven依赖中的spring配置很简单，只定义了ProducerFactory和MeterRegistry bean

@Bean
@Primary
public ProducerFactory<Object, Object> producerFactory(MessageCounter messageCounter) {
    Map<String, Object> configs = this.kafkaProperties.buildProducerProperties();
    configs.put("interceptor.classes", Collections.singletonList(RecordSuccessfullySentInterceptor.class));
    configs.put("message.counter.bean", messageCounter);
    DefaultKafkaProducerFactory<Object, Object> producerFactory = new DefaultKafkaProducerFactory(configs);
    producerFactory.setTransactionIdPrefix(this.transactionId);
    return producerFactory;
}

@Bean
public MeterRegistry meterRegistry() {
    return new PrometheusMeterRegistry(PrometheusConfig.DEFAULT);
}

@Bean({"retryProducerFactory"})
public ProducerFactory<Object, Object> retryProducerFactory(MessageCounter messageCounter) {
    Map<String, Object> configs = this.kafkaProperties.buildProducerProperties();
    configs.put("interceptor.classes", Collections.singletonList(RecordSuccessfullySentInterceptor.class));
    configs.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
    configs.put("message.counter.bean", messageCounter);
    DefaultKafkaProducerFactory<Object, Object> producerFactory = new DefaultKafkaProducerFactory(configs);
    producerFactory.setTransactionIdPrefix(this.transactionId);
    return producerFactory;
}

尽管如此，一些瞬态依赖可能会扰乱 Kafka 相关指标的 spring boot 自动配置。

Answer 1

当查看 的用法时，例如，

DefaultKafkaProducerFactoryCustomizer

在

KafkaMetricsAutoConfiguration

中定义了哪些 bean，您会发现它仅在一个地方使用 -

KafkaAutoConfiguration

.

在这里仔细看看：

@ConditionalOnMissingBean(ProducerFactory.class)
public ProducerFactory<?, ?> kafkaProducerFactory(
        ObjectProvider<DefaultKafkaProducerFactoryCustomizer> customizers) {...}

我们看到这个东西只有在

@ConditionalOnMissingBean(ProducerFactory.class)

的时候才会被调用。这意味着如果您自己定义一个

ProducerFactory

，则不会调用它并且

ProducerFactory

不会使用度量侦听器（例如

MicrometerProducerListener

）进行自定义。

在这种情况下，简单的事情就是添加

factory.addListener(new MicrometerProducerListener<>(meterRegistry));

在实例化工厂时。消费者也是如此。