|
对于最新的稳定版本，请使用 spring-cloud-stream 4.3.0！ |
响应式 Kafka Binder 中的可观测性
本节介绍如何在响应式 Kafka 绑定器中启用基于微米的可观测性。
生产者绑定
生产者绑定中内置了对可观测性的支持。 要启用它，请设置以下属性：
spring.cloud.stream.kafka.binder.enable-observation当此属性设置为true，可以观察记录的发布情况。
两个发布记录都使用StreamBridge和常规Supplier<?>可以观察到豆子。
消费者绑定
在消费者端启用可观测性比在生产者端更复杂。 消费者绑定有两个起点：
-
通过生产者绑定发布数据的主题
-
在 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创建上下文延迟。 -
当
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启动观察并在完成后正确关闭它。