|
此版本仍在开发中，尚未被视为稳定版本。对于最新的稳定版本，请使用 spring-cloud-stream 4.1.4！ |
反应式 Kafka Binder 中的可观测性
本节介绍如何在反应式 Kafka Binder 中启用基于 Micrometer 的可观测性。
生产者绑定
在生产者绑定中内置了对可观察性的支持。 要启用它，请设置以下属性：
spring.cloud.stream.kafka.binder.enable-observation当此属性设置为true中，您可以观察记录的发布。
两个发布记录都使用StreamBridge和常规Supplier<?>可以观察到豆子。
Consumer Binding
在使用者端启用可观测性比在生产者端启用更复杂。 使用者绑定有两个起点：
-
通过生产者绑定发布数据的主题
-
在 Spring Cloud Stream 之外生成数据的主题
在第一种情况下，应用程序理想情况下希望将可观察性标头向下传递到消费者入站。 在第二种情况下，如果没有启动上游观察，它将开始新的观察。
示例：具有可观测性的函数
@Bean
Function<Flux<ReceiverRecord<byte[], byte[]>>, Flux<Message<String>>> receive(ObservationRegistry observationRegistry) {
return s -> s.flatMap(record -> {
Observation receiverObservation = KafkaReceiverObservation.RECEIVER_OBSERVATION.start(
null,
KafkaReceiverObservation.DefaultKafkaReceiverObservationConvention.INSTANCE,
() -> new KafkaRecordReceiverContext(record, "user.receiver", "localhost:9092"),
observationRegistry
);
return Mono.deferContextual(contextView -> Mono.just(record)
.map(rec -> new String(rec.value()).toLowerCase())
.map(rec -> MessageBuilder.withPayload(rec)
.setHeader(IntegrationMessageHeaderAccessor.REACTOR_CONTEXT, contextView)
.build()))
.doOnTerminate(receiverObservation::stop)
.doOnError(receiverObservation::error)
.contextWrite(context -> context.put(ObservationThreadLocalAccessor.KEY, receiverObservation));
});
}在此示例中：
-
收到记录时，将创建观察。
-
如果有上游观察，它将成为
KafkaRecordReceiverContext. -
一个
Mono是使用 Context Deferred 创建的。 -
当
map作，上下文可以访问正确的观察结果。 -
结果
flatMap作将作为Flux<Message<?>>. -
出站记录将具有与输入绑定相同的可观察性标头。
示例：具有可观察性的使用者
@Bean
Consumer<Flux<ReceiverRecord<?, String>>> receive(ObservationRegistry observationRegistry, @Value("${spring.kafka.bootstrap-servers}") String bootstrap) {
return f -> f.doOnNext(record -> KafkaReceiverObservation.RECEIVER_OBSERVATION.observation(
null,
KafkaReceiverObservation.DefaultKafkaReceiverObservationConvention.INSTANCE,
() -> new KafkaRecordReceiverContext(record, "user.receiver", bootstrap),
observationRegistry).observe(() -> System.out.println(record)))
.subscribe();
}在这种情况下：
-
由于没有输出绑定，因此
doOnNext用于Flux而不是flatMap. -
直接调用
observe启动观察，并在完成后正确关闭观察。