可観測性

Spring は、アプリケーションでメトリクスとトレースの両方を可能にする観察コンセプト (英語) を定義する Micrometer (英語) を介したオブザーバビリティのサポートを提供します。

Spring クラウドストリームは、いくつかの抽象化の中で ObservationFunctionAroundWrapper を提供することにより、そのようなサポートを 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>

および利用可能なトレーサブリッジの 1 つ。たとえば、Zipkin Brave (英語) 

<dependency>
	<groupId>io.micrometer</groupId>
	<artifactId>micrometer-tracing-bridge-brave</artifactId>
</dependency>

命令的関数

命令型関数は、Observation レジストリとの対話を処理するために必要なインフラストラクチャを提供する Observation ラッパー ObservationFunctionAroundWrapper でラップされます。このような相互作用は、関数の呼び出しごとに発生します。これは、事実上、オブザベーションが関数の各呼び出しに関連付けられることを意味します (つまり、メッセージごとに 1 つのオブザベーション)。言い換えれば、命令型関数の場合、前述の必要な依存関係が存在する場合、可観測性はそのまま機能します。

リアクティブ機能

リアクティブ関数は本質的に命令型関数とは異なるため、ObservationFunctionAroundWrapper でラップされません。

命令型関数はメッセージハンドラー関数であり、メッセージが存在するたびにフレームワークによって呼び出されます。これは、N 個のメッセージに対してそのような関数が N 回呼び出される一般的なイベントハンドラーのようなものです。これにより、そのような関数をラップして、エラー処理、再試行、もちろん可観測性などの追加機能で装飾することができます。

リアクティブ関数は初期化関数です。そのロールは、ユーザーが提供したストリーム処理コード (Flux) を、バインダーが提供するソースおよびターゲットストリームに接続することです。アプリケーションの起動時に 1 回だけ呼び出されます。ストリームコードがソース / ターゲットストリームに接続されると、実際のストリーム処理を可視化することも制御することもできなくなります。それはリアクティブ API の手にかかっています。リアクティブ機能は追加の変数ももたらします。この関数により (単一のイベントだけでなく) ストリームチェーン 全体を可視化できるという事実を考慮すると、デフォルトの観察単位は何になるべきでしょうか ? ストリームチェーン 内の単一アイテム ? アイテムの範囲 ? しばらく経ってもメッセージがない場合はどうすればよいですか ? など。望んでいたのは、リアクティブ関数では何も想定できないことを強調したいということです。(リアクティブ関数と命令型関数の違いの詳細については、"リアクティブ機能" を参照してください)。

再試行やエラー処理と同様に、監視を手動で処理する必要があります。

ありがたいことに、ObservationRegistry のインスタンスを提供しながら、リアクティブ API の tap 操作を使用してストリームのセグメントを利用することで、これを簡単に行うことができます。このようなセグメントは観測単位を定義します。これは、フラックス内の単一の項目、範囲、ストリーム内で観測したいその他のあらゆるものになります。

@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 の上流の チェーン / セグメント の子オブザベーションの作成に使用されます。ただし、すでに記述されていたように、デフォルトでは、フレームワークはストリームチェーン にオブザベーションをアタッチしません。戻る。