跳到主要内容
版本：4.1.1

监控

Doubao 中英对照 Monitoring

监控监听器性能​

从 2.3 版本开始，如果 classpath 上检测到 Micrometer，且应用程序上下文中存在单个 MeterRegistry，监听容器将自动为监听器创建并更新 Micrometer Timer。可以将 ContainerProperty 的 micrometerEnabled 设置为 false 来禁用这些计时器。

维护两个计时器——一个用于对监听器的成功调用，一个用于失败。

这些计时器名为 spring.kafka.listener，并具有以下标签：

  • name：（容器 bean 名称）
  • result：success 或 failure
  • exception：none 或 ListenerExecutionFailedException

你可以使用 ContainerProperties 的 micrometerTags 属性添加额外标签。

从 2.9.8、3.0.6 版本开始，你可以在 ContainerProperties 的 micrometerTagsProvider 中提供一个函数；该函数接收 ConsumerRecord<?, ?> 并返回可基于该记录的标签，并与 micrometerTags 中的任何静态标签合并。

备注

对于并发容器，会为每个线程创建计时器，且 name 标签会加上 -n 后缀，其中 n 为 0 到 concurrency-1。

监控 KafkaTemplate 性能​

从 2.5 版本开始，如果 classpath 上检测到 Micrometer，且应用程序上下文中存在单个 MeterRegistry，模板将自动为发送操作创建并更新 Micrometer Timer。可以将模板的 micrometerEnabled 属性设置为 false 来禁用这些计时器。

维护两个计时器——一个用于对监听器的成功调用，一个用于失败。

这些计时器名为 spring.kafka.template，并具有以下标签：

  • name：（模板 bean 名称）
  • result：success 或 failure
  • exception：none，或失败时的异常类名

你可以使用模板的 micrometerTags 属性添加额外标签。

从 2.9.8、3.0.6 版本开始，你可以提供一个 KafkaTemplate.setMicrometerTagsProvider(Function<ProducerRecord<?, ?>, Map<String, String>>) 属性；该函数接收 ProducerRecord<?, ?> 并返回可基于该记录的标签，并与 micrometerTags 中的任何静态标签合并。

Micrometer 原生指标​

从 2.5 版本开始，该框架提供了 Factory 监听器，以便在创建和关闭生产者与消费者时管理 Micrometer KafkaClientMetrics 实例。

要启用此特性，只需将这些监听器添加到你的生产者和消费者工厂：

@Bean
public ConsumerFactory<String, String> myConsumerFactory() {
    Map<String, Object> configs = consumerConfigs();
    ...
    DefaultKafkaConsumerFactory<String, String> cf = new DefaultKafkaConsumerFactory<>(configs);
    ...
    cf.addListener(new MicrometerConsumerListener<String, String>(meterRegistry(),
            Collections.singletonList(new ImmutableTag("customTag", "customTagValue"))));
    ...
    return cf;
}

@Bean
public ProducerFactory<String, String> myProducerFactory() {
    Map<String, Object> configs = producerConfigs();
    configs.put(ProducerConfig.CLIENT_ID_CONFIG, "myClientId");
    ...
    DefaultKafkaProducerFactory<String, String> pf = new DefaultKafkaProducerFactory<>(configs);
    ...
    pf.addListener(new MicrometerProducerListener<String, String>(meterRegistry(),
            Collections.singletonList(new ImmutableTag("customTag", "customTagValue"))));
    ...
    return pf;
}

传递给监听器的消费者/生产者 id 会以标签名 spring.id 添加到指标的标签中。

double count = this.meterRegistry.get("kafka.producer.node.incoming.byte.total")
                .tag("customTag", "customTagValue")
                .tag("spring.id", "myProducerFactory.myClientId-1")
                .functionCounter()
                .count();

为 StreamsBuilderFactoryBean 提供了类似的监听器——请参阅 KafkaStreams Micrometer 支持。

从 3.3 版本开始，引入了一个 KafkaMetricsSupport 抽象类，用于将 io.micrometer.core.instrument.binder.kafka.KafkaMetrics 绑定管理到针对所提供 Kafka 客户端的 MeterRegistry 中。该类是上述 MicrometerConsumerListener、MicrometerProducerListener 和 KafkaStreamsMicrometerListener 的父类。不过，它可用于任意 Kafka 客户端用例。该类需要被扩展，并且必须调用其 bindClient() 和 unbindClient() API，以将 Kafka 客户端指标与 Micrometer 收集器连接起来。

Micrometer Observation​

从 3.0 版本开始，KafkaTemplate 和监听容器现已支持使用 Micrometer 进行观测（observation）。

在 KafkaTemplate 和 ContainerProperties 上将 observationEnabled 设置为 true 以启用观测；这将禁用 Micrometer 计时器，因为这些计时器现在将随每次观测一起管理。

important

Micrometer Observation 不支持批量监听器；这将启用 Micrometer 计时器

更多信息请参阅 Micrometer Tracing。

要向计时器/追踪添加标签，请分别为模板或监听容器配置自定义的 KafkaTemplateObservationConvention 或 KafkaListenerObservationConvention。

默认实现会为模板观测添加 bean.name 标签，为容器添加 listener.id 标签。

你既可以继承 DefaultKafkaTemplateObservationConvention 或 DefaultKafkaListenerObservationConvention，也可以提供全新的实现。

有关所记录默认观测的详细信息，请参阅 Micrometer Observation 文档。

从 3.0.6 版本开始，你可以基于消费者或生产者记录中的信息，向计时器和追踪添加动态标签。为此，分别向监听容器属性或 KafkaTemplate 添加自定义的 KafkaListenerObservationConvention 和/或 KafkaTemplateObservationConvention。两个观测上下文中的 record 属性分别包含 ConsumerRecord 或 ProducerRecord。

发送方和接收方上下文的 remoteServiceName 属性被设置为 Kafka 的 clusterId 属性；该属性由 KafkaAdmin 获取。如果由于某种原因——或许是缺少管理权限——你无法获取集群 id，那么从 3.1 版本开始，你可以在 KafkaAdmin 上手动设置 clusterId，并将其注入 KafkaTemplate 和监听容器。当它为 null（默认）时，admin 将调用 describeCluster 管理操作从代理获取它。

从 4.1.1 版本开始，一次失败的 describeCluster 尝试会被记住，在 KafkaAdmin 的 clusterIdRetryInterval 经过之前不会重试；该间隔默认为 5 分钟。每次尝试都会创建一个 Admin 实例并阻塞长达 operationTimeout，因此如果没有此间隔，在集群 id 不可用时，每个被观测的操作都会付出这一代价。将 clusterIdRetryInterval 设置为 Duration.ZERO 即可像早期版本那样在每次调用时尝试获取。

批量监听器观测​

使用批量监听器时，默认情况下不会创建观测，即使存在 ObservationRegistry 也是如此。这是因为观测的作用域与线程绑定，而对于批量监听器，观测与记录之间不存在一对一映射。

要在批量监听器中启用按记录的观测，请将容器工厂属性 recordObservationsInBatch 设置为 true。

@Bean
ConcurrentKafkaListenerContainerFactory<?, ?> kafkaListenerContainerFactory(
        ConcurrentKafkaListenerContainerFactoryConfigurer configurer,
        ConsumerFactory<Object, Object> kafkaConsumerFactory) {

    ConcurrentKafkaListenerContainerFactory<Object, Object> factory = new ConcurrentKafkaListenerContainerFactory<>();
    configurer.configure(factory, kafkaConsumerFactory);
    factory.getContainerProperties().setRecordObservationsInBatch(true);
    return factory;
}

当此属性为 true 时，将为批中的每条记录创建一个观测，但该观测不会传播到监听器方法。应用程序随后可以使用观测上下文来跟踪批中每条记录的处理。这使你即使在批处理上下文中也能看到每条记录的处理情况。