Reactive Kafka Binder 中的可观测性

本节描述了如何在 Reactive Kafka Binder 中启用基于 Micrometer 的可观测性。

生产者绑定

生产者绑定内置了可观测性支持。要启用它，请设置以下属性：

spring.cloud.stream.kafka.binder.enable-observation

当此属性设置为 true 时，您可以观察记录的发布。使用 StreamBridge 和常规 Supplier<?> bean 发布记录都可以被观察到。

消费者绑定

在消费者端启用可观测性比在生产者端更复杂。消费者绑定有两个起点：

  1. 一个通过生产者绑定发布数据的主题

  2. 一个在 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));
    });
}

在此示例中：

  1. 当收到一条记录时，会创建一个观察。

  2. 如果存在上游观察，它将成为 KafkaRecordReceiverContext 的一部分。

  3. 创建一个 Mono，其上下文被延迟。

  4. 当调用 map 操作时，上下文可以访问正确的观察。

  5. flatMap 操作的结果以 Flux<Message<?>> 的形式发送回绑定。

  6. 出站记录将具有与入站绑定相同的可观测性头信息。

示例：具有可观测性的消费者

@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();
}

在这种情况下：

  1. 由于没有输出绑定，因此在 Flux 上使用 doOnNext 而不是 flatMap。

  2. 对 observe 的直接调用会启动观察并在完成后正确关闭它。