可观测性

Spring 通过 Micrometer 提供可观测性支持，Micrometer 定义了一个 Observation 概念，可以在应用程序中同时实现 Metrics 和 Traces。 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 次此类函数的调用。这允许我们包装此类函数以用额外的功能（如 错误处理、重试 和 可观测性）来装饰它。

响应式函数 是初始化函数。它的作用是将用户提供的流处理代码 (Flux) 与绑定器提供的源和目标流连接起来。它在应用程序启动期间只调用一次。一旦流代码与源/目标流连接，我们对实际的流处理就没有可见性或控制权。它掌握在响应式 API 手中。响应式函数还带来了一个额外的变量。鉴于该函数让您可以看到整个流链（而不仅仅是单个事件），那么观测的默认单位应该是什么？ 流链中的单个项？一系列项？如果一段时间后没有消息怎么办？等等……我们想要强调的是，对于响应式函数，我们不能假设任何事情。（有关响应式函数和命令式函数之间差异的更多信息，请参阅 响应式函数）。

因此，就像 重试 和 错误处理 一样，您需要手动处理观测。

幸运的是，您可以通过使用响应式 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));
		});
	}
}

上面的示例模拟了将 Observation 附加到单个消息处理（即命令式函数），因为在这种情况下，观测单位从 Mono.just(..) 开始，最后一个操作将 ObservationRegistry 附加到订阅者。

如果订阅者已经附加了观测，它将用于为 tap 上游的链/片段创建子观测，但是正如我们已经声明的，默认情况下，框架不会将任何观测附加到您返回的流链。