对于最新的稳定版本，请使用 Spring for Apache Kafka 3.2.1！

对于最新的稳定版本，请使用 Spring for Apache Kafka 3.2.1！

监视侦听器性能

从版本 2.3 开始，如果在类路径上检测到侦听器，并且应用程序上下文中存在单个侦听器，侦听器容器将自动为侦听器创建和更新 Micrometer。 可以通过将 的 设置为 来禁用计时器。TimerMicrometerMeterRegistryContainerPropertymicrometerEnabledfalse

维护两个计时器 - 一个用于成功调用侦听器，一个用于失败。

计时器已命名，并具有以下标记：spring.kafka.listener

  • name：（容器 Bean 名称）

  • result：或successfailure

  • exception：或noneListenerExecutionFailedException

可以使用 的属性添加其他标记。ContainerPropertiesmicrometerTags

从版本 2.9.8、3.0.6 开始，您可以在 的 中提供函数;该函数接收 and 返回可以基于该记录的标签，并与 中的任何静态标签合并。ContainerPropertiesmicrometerTagsProviderConsumerRecord<?, ?>micrometerTags

使用并发容器时，会为每个线程创建计时器，并且标记的后缀为 其中 n 是 to 。name-n0concurrency-1
使用并发容器时，会为每个线程创建计时器，并且标记的后缀为 其中 n 是 to 。name-n0concurrency-1

监视 KafkaTemplate 性能

从版本 2.5 开始，如果在类路径上检测到 Micrometer +++s，并且应用程序上下文中存在单个操作，则模板将自动创建和更新用于发送操作的 Micrometer +++s。 可以通过将模板的属性设置为 来禁用计时器。TimerMicrometerMeterRegistrymicrometerEnabledfalse

维护两个计时器 - 一个用于成功调用侦听器，一个用于失败。

计时器已命名，并具有以下标记：spring.kafka.template

  • name：（模板 Bean 名称）

  • result：或successfailure

  • exception：或失败的异常类名none

您可以使用模板的属性添加其他标记。micrometerTags

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

千分尺原生指标

从版本 2.5 开始，该框架提供工厂监听器，以便在创建和关闭生产者和使用者时管理 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;
}

传递给侦听器的使用者/生产者将添加到计量的标签中，标签名称为 。idspring.id

获取 Kafka 指标之一的示例
double count = this.meterRegistry.get("kafka.producer.node.incoming.byte.total")
                .tag("customTag", "customTagValue")
                .tag("spring.id", "myProducerFactory.myClientId-1")
                .functionCounter()
                .count();

为 - 请参阅 KafkaStreams 千分尺支持提供了类似的侦听器。StreamsBuilderFactoryBean

千分尺观察

从 3.0 版开始，现在支持将 Micrometer 用于侦听器容器。KafkaTemplate

设置为 on 和 以启用观察;这将禁用千分尺计时器，因为计时器现在将针对每次观测进行管理。observationEnabledtrueKafkaTemplateContainerProperties

千分尺观察不支持批量侦听器;这将使千分尺计时器成为可能

有关详细信息，请参阅千分尺跟踪。

要将标记添加到计时器/跟踪，请分别配置自定义或模板或侦听器容器。KafkaTemplateObservationConventionKafkaListenerObservationConvention

默认实现会添加模板观察任务的标记和容器的标记。bean.namelistener.id

您可以子类化或提供全新的实现。DefaultKafkaTemplateObservationConventionDefaultKafkaListenerObservationConvention

有关记录的默认观测值的详细信息，请参阅千分尺观测文档。

从版本 3.0.6 开始，您可以根据使用者或生产者记录中的信息向计时器和跟踪添加动态标记。 为此，请分别向侦听器容器属性和/或添加自定义和/或。 两个观测点上下文中的属性分别包含 或。KafkaListenerObservationConventionKafkaTemplateObservationConventionKafkaTemplaterecordConsumerRecordProducerRecord

发送方和接收方上下文的属性设置为 Kafka 属性;这是由 . 如果由于某种原因（可能缺乏管理员权限），您无法检索集群 ID，从版本 3.1 开始，您可以设置手册并将其注入 s 和侦听器容器中。 当它为（默认值）时，管理员将调用管理员操作以从代理中检索它。remoteServiceNameclusterIdKafkaAdminclusterIdKafkaAdminKafkaTemplatenulldescribeCluster

千分尺观察不支持批量侦听器;这将使千分尺计时器成为可能