可观测性
Spring 通过 Micrometer 提供了对可观测性的支持，Micrometer 定义了一个观测概念，能够在应用程序中同时启用指标和追踪。
Spring Cloud Stream 在 Spring Cloud Function 层面集成了此类支持，通过提供包括 ObservationFunctionAroundWrapper 在内的多种抽象，该包装器能够包装函数以开箱即用地处理观测数据。
所需依赖
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-actuator</artifactId>
</dependency>
<dependency>
<groupId>io.projectreactor</groupId>
<artifactId>reactor-core-micrometer</artifactId>
</dependency>
以及一个可用的追踪桥接器。例如 Zipkin Brave
<dependency>
<groupId>io.micrometer</groupId>
<artifactId>micrometer-tracing-bridge-brave</artifactId>
</dependency>
命令式函数
命令式函数被封装在观测包装器 ObservationFunctionAroundWrapper 中，该包装器提供了必要的基础设施来处理与观测注册表的交互。这种交互发生在函数的每次调用中，这实际上意味着观测被附加到函数的每次调用上（即每条消息对应一次观测）。换句话说，对于命令式函数，如果前面提到的必要依赖项存在，可观测性将自动生效。
响应式函数
响应式函数本质上不同于命令式函数，因此不会使用 ObservationFunctionAroundWrapper 进行包装。
命令式函数是一种消息处理函数，每次有消息时由框架调用，类似于典型的事件处理程序，对于 N 条消息，此类函数将被调用 N 次。这使我们能够包装此类函数，为其添加额外功能，例如错误处理、重试，当然还有可观测性。
Reactive function 是初始化函数。它的职责是将用户提供的流处理代码（Flux）与 binder 提供的源和目标流连接起来。它仅在应用程序启动时调用一次。一旦流代码与源/目标流连接，我们就无法看到或控制实际的流处理。这完全由 reactive API 掌控。Reactive function 还引入了一个额外的变量。考虑到该函数让你能够看到整个流链（而不仅仅是单个事件），那么默认的观察单位应该是什么？流链中的单个项目？一系列项目？如果经过一段时间后没有消息怎么办？等等……我们想要强调的是，对于 reactive functions，我们不能做任何假设。（有关 reactive 和 imperative functions 之间差异的更多信息，请参阅 Reactive Functions）。
因此，就像处理重试和错误处理一样，你需要手动处理观测。
幸运的是，你可以通过使用响应式API的tap操作来轻松实现这一点，同时提供一个ObservationRegistry实例。这段代码定义了一个观察单元，它可以是Flux中的单个元素、一个范围，或者你希望在流中观察的任何其他内容。
@SpringBootApplication
public class DemoStreamApplication {
Logger logger = LoggerFactory.getLogger(DemoStreamApplication.class);
public static void main(String[] args) {
Hooks.enableAutomaticContextPropagation();
SpringApplication.run(DemoStreamApplication.class, args);
}
@Bean
public Function<Flux<String>, Flux<String>> uppercase(ObservationRegistry registry) {
return flux -> flux.flatMap(item -> {
return Mono.just(item)
.map(value -> value.toUpperCase())
.doOnNext(v -> logger.info(v))
.tap(Micrometer.observation(registry));
});
}
}
上述示例模拟了将观察附加到单个消息处理（即命令式函数）的过程，因为在这种情况下，观察单元从 Mono.just(..) 开始，而最后一个操作将 ObservationRegistry 附加到订阅者。
如果订阅者已经附加了一个观察对象，它将用于为 tap 上游的链/段创建一个子观察对象。然而，正如我们已经说明的，默认情况下，框架不会为你返回的流链附加任何观察对象。