最新の安定バージョンについては、spring-cloud-stream 4.2.1 を使用してください。

可観測性

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