1. WebFluxのコアリアクティブサーバー基盤
SpringフレームワークのインフラストラクチャはServlet APIを中心に構築されており、密接に結合されています。リアクティブWebに入る前に、まずWebモジュールの設計を見直し、その機能を確認しましょう。
低レベルのServletコンテナは、コンテナ内のすべてのマッピングされたServletの処理を担当します。DispatchServletは、柔軟で高度に設定可能なSpring Webインフラストラクチャと複雑なServlet APIを統合するための統合ポイントとして機能します。HandlerMappingはビジネスロジックとServlet APIを分離します。
Spring MVCの制限事項：
- リクエストライフサイクル全体でノンブロッキング操作が許可されない。ノンブロッキングHTTPクライアントが標準で提供されない。
- WebMVC抽象化は、ノンブロッキングServlet 3.1のすべての機能をサポートしていない。
- ノンServletサーバーでは、Spring Web機能の再利用が柔軟性に欠ける。
そのため、Springチームの過去数年間の主要課題は、アノテーションベースのプログラミングモデルを使用しながら、非同期ノンブロッキングサービスのすべての利点を提供する新しいソリューションを構築することでした。
1.1 リアクティブWebカーネル
SpringフレームワークのインフラストラクチャはServlet APIを中心に構築されており、密接に結合されています。リアクティブWebカーネルでは、まずjavax.servlet.Servlet.serviceメソッドを置き換えるシミュレーションインターフェースとリクエスト処理方法が必要です。
関連するクラスとインターフェースの変更：
Servlet APIのクライアントリクエストとサーバーレスポンスの相互作用方法を強化・カスタマイズします。
/**
* リクエストのカプセル化。
* リクエストボディの取得型はFluxで、リアクティブ能力を持つことを示す。
* DataBufferはバイトバッファの抽象化であり、特定のサーバー実装でのデータ交換を容易にする。
* リクエストボディの他にも、メッセージヘッダー、リクエストパス、Cookie、クエリパラメータなどの情報が提供される。
*/
interface ServerHttpRequest {
// ReactiveHttpInputMessageに存在するメソッド
Flux<DataBuffer> getBody();
// ...
}
/**
* レスポンスのカプセル化。
* writeWithメソッドの引数はPublisherで、リアクティブを提供し、特定のリアクティブライブラリから切り離す。
* 戻り値はMono<Void>で、ネットワークへのデータ送信が非同期プロセスであることを示す。
* つまり、Monoをサブスクライブしたときのみデータ送信プロセスが実行される。
* 受信側は転送プロトコルのフロー制御によりバックプレッシャーをサポートできる。
*/
interface ServerHttpResponse {
// ... 親インターフェースReactiveHttpOutputMessageに対応
Mono<Void> writeWith(Publisher<? extends DataBuffer> body);
// ...
}
/**
* HTTPリクエスト-レスポンスのコンテナ。
* これは高レベルインターフェースで、HTTPインタラクションの他にもフレームワーク関連情報を保存できる。
* 例えば、リクエストの復元されたWebSession情報など。
*/
interface ServerWebExchange {
// ...
ServerHttpRequest getRequest();
ServerHttpResponse getResponse();
// ...
Mono<WebSession> getSession();
// ...
}
上記3つのインターフェースはServlet APIのインターフェースに類似しています。リアクティブインターフェースはインタラクションモデルの観点からほぼ同じメソッドを提供しつつ、すぐに使えるリアクティブ機能を提供します。
リクエストハンドラーとフィルターAPI：
/**
* WebMVCのDispatcherServletに対応
* リクエストハンドラーの検索、ビューリゾルバーによるビュー描画を行うため、handleメソッドは何も返す必要がない。
*
* 戻り値のMono<Void>は非同期処理を提供する。
* 指定時間内にシグナルが出現しなければ、実行をキャンセルできる。
*/
interface WebHandler {
Mono<Void> handle(ServerWebExchange exchange);
}
/**
* フィルターチェーン
*/
interface WebFilterChain {
Mono<Void> filter(ServerWebExchange exchange);
}
/**
* フィルター
*/
interface WebFilter {
Mono<Void> filter(ServerWebExchange exchange, WebFilterChain chain);
}
これらはリアクティブWebが備えるべき基本APIです。これらのインターフェースを異なるサーバーに適応させる必要があります。つまり、ServerHttpRequestとServerHttpResponseと直接インタラクションするコンポーネントです。同時に、ServerWebExchangeの構築、特定のセッションストレージ、ローカライズリゾルバーなどの情報の保存も担当します。
public interface HttpHandler {
Mono<Void> handle(ServerHttpRequest request, ServerHttpResponse response);
}
この適切な抽象化により、サーバーエンジンの詳細が隠蔽され、具体的なサーバーの動作方法はSpring WebFluxユーザーにとって重要ではありません。
1.2 リアクティブWebとMVCフレームワーク
Spring Web MVCモジュールの重要な特徴はアノテーションベースです。そのため、リアクティブWebスタックにも同じ概念を提供する必要があります。WebMVCのインフラストラクチャを再利用し、Flux、Mono、Publisherなどのリアクティブ型で同期通信を置き換えます。Spring Web MVCと同じHandlerMappingとHandlerAdapterチェーンを保持し、Reactorベースのリアクティブインタラクションでリアルタイムコマンドを置き換えます：
interface HandlerMapping {
/*
HandlerExecutionChain getHandler(HttpServletRequest request)
*/
Mono<Object> getHandler(ServerWebExchange exchange);
}
interface HandlerAdapter {
boolean supports(Object handler);
/*
ModelAndView handle(HttpServletRequest request, HttpServletResponse response, Object handler);
*/
Mono<HandlerResult> handle(ServerWebExchange exchange, Object handler);
}
リアクティブHandlerMappingでは、2つのメソッドは全体的に類似していますが、違いはリアクティブ戻り値のMono型がリアクティブをサポートすることです。リアクティブHandlerAdapterインターフェースでは、ServerWebExchangeクラスがリクエストとレスポンスを組み合わせているため、handleメソッドのリアクティブ版はより簡潔です。このメソッドはModelAndViewではなくHandlerResultのMonoを返します。これらの手順に従うことで、実行階層構造を破壊することなくリアクティブインタラクションモデルを得ることができ、既存の設計を維持し、最小限の変更で既存コードを再利用できます。
最終設計：
- 受信リクエストは低レベルサーバーエンジンによって処理されます。サーバーエンジンリストはServletAPIベースのサーバーに限定されません。各サーバーエンジンには独自のリアクティブアダプターがあり、HTTPリクエストとHTTPレスポンスの内部表現をServerHttpRequestとServerHttpResponseにマッピングします。
- HttpHandlerフェーズでは、指定されたServerHttpRequest、ServerHttpResponse、ユーザーセッションおよび関連情報をServerWebExchageインスタンスに組み合わせます。
- WebFilterChainフェーズでは、定義されたWebFilterをチェーンに組み合わせます。その後、WebFilterChainはこのチェーン内の各WebFilterインスタンスのWebFilter#filterメソッドを実行して、受信ServerWebExchangeをフィルタリングします。
- すべてのフィルタ条件を満たせば、WebFilterChainはWebHandlerインスタンスを呼び出します。
- HandlerMappingインスタンスを検索し、最初に適切なものを呼び出します。RouterFunctionMappingまたはRequestMappingHandlerMappingのいずれかになります。RouterFunctionMappingはWebFluxに導入され、純粋な機能リクエスト処理を超えています。
- 以前の機能と同じフェーズで、リアクティブストリームを使用してレスポンスストリームを構築します。
WebFluxモジュールでは、デフォルトのサーバーエンジンはNettyです。Nettyサーバーはデフォルトサーバーとして適しており、リアクティブ領域で広く使用されているためです。このサーバーエンジンは同時にクライアントとサーバーの非同期ノンブロッキングインタラクションを提供します。同時に、サーバーエンジンの選択は柔軟です。WebFluxモジュールはSpring Web MVCモジュールのアーキテクチャに対応しており、理解しやすいです。
1.3 WebFluxベースの純粋関数型Web
純粋関数型Webは主に関数型ルーティングマッピングです。関数型マッピングを通じて、軽量アプリケーションを生成できます。例は以下の通り：
依存関係の追加
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<parent>
<artifactId>spring-boot-starter-parent</artifactId>
<groupId>org.springframework.boot</groupId>
<version>2.7.3</version>
<relativePath/>
</parent>
<groupId>com.jiege</groupId>
<artifactId>jiege-webFlux.4.1.3</artifactId>
<version>1.0-SNAPSHOT</version>
<properties>
<maven.compiler.source>8</maven.compiler.source>
<maven.compiler.target>8</maven.compiler.target>
</properties>
<dependencies>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-webflux</artifactId>
</dependency>
<dependency>
<groupId>org.projectlombok</groupId>
<artifactId>lombok</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.security</groupId>
<artifactId>spring-security-crypto</artifactId>
</dependency>
</dependencies>
</project>
エンティティの作成
@Data
@AllArgsConstructor
@NoArgsConstructor
public class Order {
private String id;
}
ハンドラーの追加
@Service
public class OrderHandler {
private Map<String, Order> orderMap = new HashMap<>();
/**
* 注文作成
* @param serverRequest
* @return
*/
public Mono<ServerResponse> create (ServerRequest serverRequest){
return serverRequest.bodyToMono(Order.class)
.doOnNext(order ->{
orderMap.put(order.getId(),order);
})
.flatMap(order -> ServerResponse.created(URI.create("/order/" + order.getId())).build());
}
/**
* 注文情報取得
* @param request
* @return
*/
public Mono<ServerResponse> get(ServerRequest request){
String id = request.pathVariable("id");
Order order1 = orderMap.get(id);
return Mono.just(order1)
.flatMap(order -> ServerResponse.ok().syncBody(order)).switchIfEmpty(ServerResponse.notFound().build());
}
/**
* 注文リスト取得
* @param request
* @return
*/
public Mono<ServerResponse> list(ServerRequest request){
return Mono.just(orderMap.values().stream().collect(Collectors.toList()))
.flatMap(order -> ServerResponse.ok().syncBody(order))
.switchIfEmpty(ServerResponse.notFound().build());
}
}
ルーティング設定
@SpringBootApplication
public class WebFluxApplication {
public static void main(String[] args) {
SpringApplication.run(WebFluxApplication.class);
}
@Bean
public RouterFunction<ServerResponse> routes(OrderHandler orderHandler){
// 2つのパラメータを含む：1、テスト条件が通過するかどうか。通過すれば第2パラメータで指定されたルート関数にルーティング
return nest(
// リクエストパスが指定されたプレフィックスに一致するか判定
path("/orders"),
// 一致すればこの関数にルーティング
nest(accept(MediaType.APPLICATION_JSON),// リクエストヘッダーフィールドacceptがAPPLICATION_JSONに一致するか判定
// 一致すれば下記ルート関数にルーティング。/orders/{id}をhandlerのgetにルーティング
route(GET("/{id}"),orderHandler::get))
// GETリクエスト/ordersの場合、orderhandler.listにルーティング
.andRoute(method(HttpMethod.GET),orderHandler::list)
// contentTypeが一致し、パスがordersに一致すればこの関数にルーティング
.andNest(contentType(MediaType.APPLICATION_JSON),
// POSTリクエスト/ordersの場合、handlerのcreateメソッドにルーティング
route(POST("/"),orderHandler::create)));
}
}
createメソッドはServerRequest（関数型ルーティングリクエスト型）を受け取ります。ServerRequestはリクエストボディを手動でMonoまたはFluxにマッピングできます。このAPIではリクエストボディがマッピングすべきクラスを指定することも可能です。最後に、WebFluxの関数型追加機能はServerResponseクラスのストリームAPIを使用してレスポンスを構築するAPIを提供します。関数型ルーティング宣言APIの他に、リクエストとレスポンス処理の関数型APIもあります。同時に、関数型Webフレームワークにより、Springインフラストラクチャ全体を起動せずにWebアプリケーションを構築できます。以下はケース：
@Data
public class PassswordDTO {
private String raw;
private String secured;
public PassswordDTO(@JsonProperty("raw") String raw,@JsonProperty("secured") String secured){
this.raw = raw;
this.secured = secured;
}
}
public class StandaloneApplication {
static Logger LOGGER = LoggerFactory.getLogger(StandaloneApplication.class);
public static void main(String[] args) {
long start = System.currentTimeMillis();
// Routesメソッドを呼び出し、RouteFunctionをHttpHandlerに変換
HttpHandler handler = RouterFunctions.toHttpHandler(routes(
// BCrytアルゴリズムで18回ハッシュ、ここでは数秒かかる
new BCryptPasswordEncoder(18)
));
// 内蔵HttpHandlerアダプター
ReactorHttpHandlerAdapter reactorAdapter = new ReactorHttpHandlerAdapter(handler);
// HTTPServerインスタンス作成、ReatorNettyAPIの一部
DisposableServer server = HttpServer.create()
.host("localhost")
.port(8080) // ポート設定
.handle(reactorAdapter) // ハンドラー指定
.bindNow();// bindNow呼び出しでサービス起動
LOGGER.info("started in" + (System.currentTimeMillis() - start) + "ms");
// アプリケーションをアクティブ状態に保ち、スレッドをブロックし、サーバー処理イベントを監視
server.onDispose().block();
}
private static RouterFunction<?> routes(BCryptPasswordEncoder passwordEncoder) {
return
// リクエストPOSTかつパスがpasswordに一致
route(POST("password"),
reqeust -> reqeust
.bodyToMono(PassswordDTO.class)
.doOnNext(System.out::println)
// BCryptPasswordEncoderで暗号化済みの生パスワードをチェック、パスワード暗号化
.map(p -> passwordEncoder.matches(p.getRaw(),p.getSecured()))
// 秘密が一致すればOK
// そうでなければEXPECTATION_FAILED
.flatMap(isMatched -> isMatched ? ServerResponse.ok().build():
ServerResponse.status(HttpStatus.EXPECTATION_FAILED).build()));
}
}
関数型ルーティング宣言に切り替えることで：
- すべてのルーティング設定を一箇所で維持でき、受信リクエストに対してリアクティブ手法で処理できる。
- 受信リクエストパラメータ、パス変数、リクエストの他の重要コンポーネントへのアクセスにおいて、関数型ルーティングの柔軟性はアノテーションベースの通常手法とほぼ同等。
- 関数型ルーティングはSpringフレームワークインフラストラクチャ全体の実行を避けられ、ルーティング設定面でも同様に柔軟で、アプリケーションの起動をより高速にできる。
1.4 WebClientベースのノンブロッキングクロスサービス通信
本質的に、WebClientは古いRestTemplateのリアクティブ代替です。WebClientには関数型APIがあり、Project Reactor型（FluxやMonoなど）への組み込みマッピングを提供します。以下の例：
// 作成時にベースURIを指定
WebClient.create("http://localhost/api")
// リクエストメソッド：GETを指定
.get()
// 相対URIを指定し、URI変数を展開
// メッセージヘッダー、Cookie、リクエストボディも指定可能。
.uri("/users/{id}", userId)
// 結果の処理方法を指定
.retrieve()
// レスポンスボディを逆シリアル化
.bodyToMono(User.class)
// その他の操作
.map(...)
// サブスクライブで非同期実行をトリガーし、リモート呼び出しを行う。ここではサブスクライブの副作用のみ使用。
.subscribe();
WebClientはリアクティブストリーム仕様に記述された動作に従います。subscribeメソッドを通じてのみ、WebClientは接続を確立し、リモートサーバーへのデータ送信を開始します。
パスワードチェックサービスへの呼び出しを構築し、WebClient APIを使用してレスポンスステータスをカスタム方法で処理します：
1.5 リアクティブテンプレートエンジン
Spring 5.xとWebFluxモジュールはApache Velocityを含む多くの技術のサポートを放棄しました。Spring WebFluxはWeb MVCと同じビューレンダリング技術を持っています。以下の例はビュー描画指定の一般的な方法を示しています：
@RequestMapping("/")
public String index() {
// 文字列を論理ビュー名として返す
return "index";
}
テンプレートレンダリングプロセスでリアクティブ手法をどうサポートするか？
大規模音楽プレイリストレンダリングのケースを検討:
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-freemarker</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-webflux</artifactId>
</dependency>
<dependency>
<groupId>org.projectlombok</groupId>
<artifactId>lombok</artifactId>
<optional>true</optional>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>io.projectreactor</groupId>
<artifactId>reactor-test</artifactId>
<scope>test</scope>
</dependency>
@RequestMapping("/play-list-view-ftl")
public Mono<String> getPlaylist(final Model model) {
List<Song> songs = new ArrayList<>();
// 通常は非同期でデータソースからデータを取得
Song song = null;
for (int i = 0; i < 5; i++) {
song = new Song("曲目" + i, "張三" + i, "1001" + i, "アルバム1" + (i %
3));
songs.add(song);
}
// リアクティブストリームに変換
final Flux<Song> playlistStream = Flux.fromIterable(songs);
// データ描画
return playlistStream
// コレクションに収集、ここは非同期ではない
.collectList()
// モデルに配置
.doOnNext(list -> model.addAttribute("playList", list))
// 論理ビュー名を返す
.then(Mono.just("/freemarker/play-list-view"));
}
@Getter
@AllArgsConstructor
public class Song {
private String id;
private String name;
private String artist;
private String album;
}
上記例のように、非同期でビュー名を返すためにリアクティブ型Monoを使用しています。さらに、テンプレートには指定されたSongリストで埋められるプレースホルダーdataSourceがあります。
コンテキストデータを提供する一般的な方法はModelを定義し、必要な属性を配置することです。
FreeMarkerはデータのリアクティブ描画とノンブロッキング描画をサポートしていません。すべての曲をリストに収集し、収集したデータをすべてModelに配置する必要があります。
src/main/resources/templates/freemarker/play-list-view.ftl：
<html>
<body>
<head>
<meta charset="UTF-8"/>
<title>曲目リスト-freemarker</title>
</head>
<table border="1">
<thead>
</thead>
<tbody>
<#list playList as e>
<tr>
<td>${e.id}</td>
<td>${e.name}</td>
<td>${e.artist}</td>
<td>${e.album}</td>
</tr>
</#list>
</tbody>
</table>
</body>
</html>
src/main/java/com/webflux/demo/config/WebConfig.java：
@Configuration
@EnableWebFlux
public class MyConfig implements WebFluxConfigurer {
@Override
public void configureViewResolvers(ViewResolverRegistry registry) {
// FreeMakrerViewResolverテンプレートを登録
registry.freeMarker();
}
@Bean
public FreeMarkerConfigurer freeMarkerConfigurer() {
// 設定子を設定
final FreeMarkerConfigurer configurer = new FreeMarkerConfigurer();
// テンプレートパスを設定
configurer.setTemplateLoaderPath("classpath:/templates");
return configurer;
}
}
1.6 Spring Web FluxとSpring Web MVCの比較
Spring MVCとWebFluxのどちらを使うべきか？
Spring MVCとSpring WebFluxは分離したものではありません。どちらも開発の選択肢を拡張します。両者の設計目標は相互の継続性と一貫性であり、一緒に使用してそれぞれの利点を活かすことができます。以下の図は両者の関係と違いを示しています：
具体的な使用方法は以下を考慮：
- 既存プロジェクトがSpring MVCベースで問題なければ変更しない。命令型プログラミングの開発、読解、デバッグが最も簡単。利用可能なライブラリも多いが、大部分はブロッキング式。
- プロジェクトの技術スタックがノンブロッキングなら、WebFluxを使用して環境と同じモデルで実行できる。WebFluxはサーバーオプション（Netty、Tomcat、Jetty、UndertowおよびServlet 3.1以上のコンテナ）、プログラミングモデルオプション（アノテーションベースのコントローラーと関数型webエンドポイント）、リアクティブライブラリオプション（Reactor、RxJavaおよびその他）を提供。
- Java8 lambdaまたはKotlinの利点を活かし、軽量で関数型webフレームワークを使いたい場合、Spring WebFlux関数型webエンドポイントプログラミングモデルを使用。Spring WebFluxは小規模アプリケーションや複雑な要件のないマイクロサービスに最適。
- マイクロサービスアーキテクチャでは、Spring WebFluxとSpring MVCを同時に使用したり、Spring WebFluxを関数型エンドポイントとして使用できる。同じアノテーションプログラミングモデルに基づいているため、適切な場面で正しいツールを使用するのが便利。
- アプリケーション評価の簡単な方法は依存関係をチェックすること。ブロッキング式永続化API（JPA、JDBC）またはブロッキング式ネットワークAPIを使用している場合、Spring MVCが最良の選択。技術的にはReactorとRxJavaも分離スレッドでブロッキング操作をサポートできるが、ノンブロッキングweb技術スタックの全利点を発揮できない。
- Spring MVCアプリケーションがリモートサービスを呼び出す必要がある場合、リアクティブなWebClientを使用。Spring MVCコントローラーメソッドが直接リアクティブ型（Reactor、RxJavaまたはその他）データを返せる。各リモート呼び出しの遅延が大きければ大きいほど、リモート呼び出し間の依存関係が大きければ大きいほど、リアクティブモデルの利点が顕著。もちろん、Spring MVCコントローラーは他のリアクティブコンポーネントを呼び出せる。
- 開発チームが大規模なら、ノンブロッキング、関数型、宣言型プログラミングモデルへの急な学習曲線を考慮する必要がある。ベストプラクティスはまず部分的にリアクティブWebClientを使って転換すること。それから小さなモジュールで使用し、リアクティブモデルがもたらす利点を評価する。通常プロジェクト全体でリアクティブモデルに完全に転換する必要はない。リアクティブプログラミングの利点が不明確なら、まずノンブロッキングI/Oの動作フロー（例えばシングルスレッドNode.jsの並行性）と効果を学ぶこと。
2. WebFluxの応用
2.1 マイクロサービスベースシステム
WebFluxの最初の応用はマイクロサービスシステムです。マイクロサービスシステムの最も顕著な特徴は大量のI/O通信です。I/Oの存在、特にブロッキング式I/Oは、システム全体の遅延とスループットを低下させます。
Spring Cloud Gateway
Spring Cloud GatewayはSpring公式がSpring 5.0、Spring Boot 2.0およびProject Reactorなどの技術を使用して開発したゲートウェイで、マイクロサービスアーキテクチャにシンプルで効果的な統一APIルーティング管理方式を提供することを目的としています。
Spring Cloud Gateway は Spring Cloud エコシステムのゲートウェイとして、目標は Netflix ZUULの代替です。統一路由方式を提供するだけでなく、Filterチェーン方式でゲートウェイ基本機能（セキュリティ、監視/トレース、レート制限など）を提供します。
Spring Cloud Gateway 機能特性
- Spring Framework 5、Project ReactorおよびSpring Boot 2.0に基づく
- ダイナミックルーティング
- PredicatesおよびFiltersが特定ルーティングに作用
- Hystrixサーキットブレーカーとの統合
- Spring Cloud DiscoveryClientとの統合
- 簡単に書けるPredicatesおよびFilters
- レート制限
- パス書き換え
Spring Cloud Gateway 工程フロー
クライアントがSpring Cloud Gatewayにリクエストを送信。その後Gateway Handler Mappingでリクエストに一致するルーティングを見つけ、それをGateway Web Handlerに送信。Handlerは指定されたフィルターチェーンを通じてリクエストを実際のサービスに送信し、ビジネスロジックを実行して戻る。
フィルター間を点線で分離するのは、フィルターがプロキシリクエスト送信前（pre）または後（post）にビジネスロジックを実行する可能性があるためです。
2.2 大ファイルアップロード
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-webflux</artifactId>
</dependency>
src/main/resources/static/index.html：
<html lang="en">
<head>
<meta charset="UTF-8">
<title>ファイルアップロード</title>
</head>
<body>
<form action="/single" method="post" enctype="multipart/form-data">
<input type="file" name="file">
<input type="submit" value="単一ファイルアップロード">
</form>
<hr>
<form action="/multi" method="post" enctype="multipart/form-data">
<input type="file" name="file">
<input type="file" name="file">
<input type="file" name="file">
<input type="file" name="file">
<input type="submit" value="複数ファイルアップロード">
</form>
</body>
</html>
@RestController
public class FileController {
@RequestMapping("/single")
public Mono<String> singleFile(@RequestPart("file")Mono<FilePart> file){
return file.map(filePart -> {
Path tempFile = null;
try {
tempFile = Files.createTempFile("file-", filePart.filename());
} catch (IOException e) {
e.printStackTrace();
}
System.out.println("ファイルパス：" + tempFile.toAbsolutePath());
//非同期ファイルchannel
AsynchronousFileChannel channel = null;
try {
//指定ファイルの書き込み操作channelを開く
channel = AsynchronousFileChannel.open(tempFile, StandardOpenOption.WRITE);
} catch (IOException e) {
e.printStackTrace();
}
DataBufferUtils.write(filePart.content(),channel,0)
.doOnNext(System.out::println)
.doOnComplete(()->{
System.out.println("ファイルコピー完了");
}).subscribe();
// ファイル情報をカプセル化
return tempFile;
}).map(tmp -> tmp.toFile())
.flatMap(fileSingle -> file.map(FilePart::filename));
}
@RequestMapping("/multi")
public Mono<List<String>> multiFiles(@RequestPart("file") Flux<FilePart> filePartFlux){
return filePartFlux.map(filePart -> {
Path tempFile = null;
try {
tempFile = Files.createTempFile("mfile-", filePart.filename());
} catch (IOException e) {
e.printStackTrace();
}
System.out.println("ファイルパス：" + tempFile.toAbsolutePath());
// 下層はゼロコピーを使用
filePart.transferTo(tempFile.toFile());
return tempFile;
}).map(file -> file.toFile())
.flatMap(fileSingle ->filePartFlux.map(FilePart::filename)).collectList();
}
}
3. Spring WebFluxデータベースアクセス
リアクティブリレーショナルデータベース接続（Reactive Relational Database Connectivity、R2DBC）は完全にリアクティブなデータベースAPIを探索するイニシアチブです。Spring DataチームはR2DBCイニシアチブをリードし、それを使用してリアクティブアプリケーション内のリアクティブデータアクセス環境でアイデアを調査・検証します。R2DBCはSpring OnePlatform 2018会議で公開され、バックプレッシャーをサポートするリアクティブデータベースアクセスAPIを定義することが目標です。Spring DataチームはリアクティブNoSQL永続化で先進的な経験を積んでおり、真のリアクティブ言語レベルデータアクセスAPIを実現するビジョンを提示しました。R2DBCプロジェクトには以下の部分が含まれます。
- R2DBCサービスプロバイダインターフェース（Service Provider Interface、SPI）はドライバー実装のミニマルAPIを定義し、ドライバー実装者が遵守すべきAPIを徹底的に削減します。SPIはアプリケーションコードで直接使用するには適さず、専用クライアントライブラリが必要です。
- R2DBCクライアントは人間工学的APIとヘルパークラスを提供し、ユーザーリクエストをSPIレベルに変換します。R2DBCクライアントのR2DBC SPIに対する役割は、JdbiライブラリのJDBCに対する役割と同じです。
3.1 Spring Data R2DBCとMySQLの統合
依存関係の導入
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-data-r2dbc</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-webflux</artifactId>
</dependency>
<dependency>
<groupId>dev.miku</groupId>
<artifactId>r2dbc-mysql</artifactId>
<scope>runtime</scope>
</dependency>
<dependency>
<groupId>mysql</groupId>
<artifactId>mysql-connector-java</artifactId>
<scope>runtime</scope>
</dependency>
<dependency>
<groupId>org.projectlombok</groupId>
<artifactId>lombok</artifactId>
<optional>true</optional>
</dependency>
データベースspring_r2dbcを作成し、以下のSQLを実行：
CREATE DATABASE `spring_r2dbc`;
CREATE TABLE `student` (
`id` int(11) AUTO_INCREMENT,
`code` varchar(50) NOT NULL,
`name` varchar(50) NOT NULL,
`gender` char(1) NOT NULL,
`birthday` date NOT NULL,
`address` varchar(300) NULL,
`remark` varchar(1000) NULL,
`active` tinyint NOT NULL DEFAULT 1,
`createdAt` datetime(0) NOT NULL DEFAULT CURRENT_TIMESTAMP(0),
`createdBy` varchar(50) NOT NULL,
`updatedAt` datetime(0) NOT NULL DEFAULT CURRENT_TIMESTAMP(0) ON UPDATE
CURRENT_TIMESTAMP(0),
`updatedBy` varchar(50) NOT NULL,
PRIMARY KEY (`id`),
UNIQUE INDEX `idx_main`(`code`)
);
データ追加
INSERT INTO student ( CODE, NAME, gender, birthday, address, createdBy, updatedBy )
VALUES
( 'NO.0001', 'Tom', 'M', '2001-03-05', 'LiMing', 'TEST', 'TEST' ),
( 'NO.0002', 'Ted', 'M', '2002-06-12', 'zhangYu', 'TEST', 'TEST' ),
( 'NO.0003', 'Mary', 'F', '2003-9-12', 'zhaoGou', 'TEST', 'TEST' );
データソース設定
設定ファイルapplication.ymlを修正し、以下の設定を追加：
spring:
r2dbc:
url: r2dbc:mysql://localhost:3306/spring_r2dbc?serverTimezone=Asia/Shanghai
username: root
password: 123xxxxxx
@RestController
@RequestMapping("/api/students")
public class StudentController {
private final StudentRepository studentRepository;
public StudentController(StudentRepository studentRepository) {
this.studentRepository = studentRepository;
}
@GetMapping
public Flux<Student> index() {
return studentRepository.findAll();
}
}
エンティティクラス作成
コードは以下の通り：
@Data
public class Student {
@Id
private Long id;
private String code;
private String name;
private String gender;
private LocalDate birthday;
private String address;
private String remark;
private boolean active;
private LocalDateTime createdAt;
private String createdBy;
private LocalDateTime updatedAt;
private String updatedBy;
}
リポジトリクラス作成
Spring Data R2DBCはSpring Data JPAの概念を基本的に踏襲しているが、機能的にはJPAほど強力ではない。
コードは以下の通り：
@Repository
public interface StudentRepository extends ReactiveCrudRepository<Student, Long> {
}
コントローラー作成
コードは以下の通り：
import reactor.core.publisher.Flux;
@RestController
@RequestMapping("/api/students")
public class StudentController {
private final StudentRepository studentRepository;
public StudentController (StudentRepository studentRepository){
this.studentRepository = studentRepository;
}
@GetMapping
public Flux<Student> index(){
return studentRepository.findAll();
}
}
起動およびアクセス
http://localhost:8080/studentsを使用してインターフェースにアクセス。
3.2 Spring Web FluxとRedisの統合
- 依存関係導入：
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-data-redis-reactive</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-webflux</artifactId>
</dependency>
<dependency>
<groupId>org.projectlombok</groupId>
<artifactId>lombok</artifactId>
<optional>true</optional>
</dependency>
- yml設定
spring:
redis:
host: xxx.xxx.xxx.xxx
port: 6379
timeout: 5000
- config設定
@Configuration
@EnableCaching
public class MyRedisConfig extends CachingConfigurerSupport {
@Bean
public ReactiveRedisTemplate<String, Object> reactiveRedisTemplate(ReactiveRedisConnectionFactory factory) {
RedisSerializer serializer = new StringRedisSerializer();
GenericJackson2JsonRedisSerializer valueSerializer = new GenericJackson2JsonRedisSerializer();
RedisSerializationContext<String, Object> context = RedisSerializationContext.newSerializationContext()
.key(serializer)
.value(valueSerializer)
.hashKey(serializer)
.hashValue(valueSerializer).build();
ReactiveRedisTemplate<String, Object> reactiveRedisTemplate = new ReactiveRedisTemplate<String, Object>(factory, context);
return reactiveRedisTemplate;
}
}
- エンティティ
@Data
public class City implements Serializable {
private static final long serialVersionUID = -1L;
// 都市番号
@Id
private Long id;
// 省番号
private Long provinceId;
// 都市名
private String cityName;
// 説明
private String description;
}
- controller
@RestController
@RequestMapping("/city")
public class ReactiveCityController {
@Autowired
private ReactiveRedisTemplate reactiveRedisTemplate;
@GetMapping(value = "/{id}")
public Mono<City> findCityById(@PathVariable("id") Long id) {
String key = "city_" + id;
ReactiveValueOperations<String, City> operations = reactiveRedisTemplate.opsForValue();
Mono<City> city = operations.get(key);
return city;
}
@PostMapping
public Mono<City> saveCity(@RequestBody City city) {
return reactiveRedisTemplate.opsForValue()
.getAndSet("city_" + city.getId(), city);
}
@DeleteMapping(value = "/{id}")
public Mono<Long> deleteCity(@PathVariable("id") Long id) {
String key = "city_" + id;
return reactiveRedisTemplate.delete(key);
}
}
4. StepVerifierを使用したリアクティブストリームテスト
テスト目的のために、Reactorは追加のreactor-testモジュールを提供し、Publisherの検証フローを構築するためのストリームAPIを提供するStepVerifierを含みます。
4.1 StepVerifierの要点
Publisherの検証には主に2つの方法があります。1つ目はStepVerifier.<T>create(Publisher<T> source)です。この技術で構築されたテストは以下のようになります：
StepVerifier
.create(Flux.just("foo", "bar"))
.expectSubscription()
.expectNext("foo")
.expectNext("bar")
.expectComplete()
.verify();
この例では、Publisherは2つの特定要素を生成し、後続操作は特定要素が最終サブスクライバーに渡されたかどうかを検証します。このクラスが提供するビルダーテクニックは検証プロセス中のイベント発生順序を定義できます。前述のコードによれば、最初に発生するイベントはサブスクリプション関連イベントで、それに続くイベントはfooとbar文字列でなければなりません。最後に、StepVerifier#expectCompletionは終了シグナルの存在を定義します。この例では、Subscriber#onCompleteの呼び出しまたは指定されたFluxの正常完了でなければなりません。検証を実行する、つまり作成ストリームをサブスクライブするには、.verify()メソッドを呼び出す必要があります。verify()はブロッキング呼び出しで、ストリームが予想されるすべてのイベントを発行するまでブロッキングします。この単純な技術を使用することで、カウント可能な要素とイベントでPublisherを検証できます。しかし、大量の要素でフローを検証するのは困難です。発行要素数が特定数に達したかをチェックする場合、.expectNextCount()を使用できます。以下のコード：
StepVerifier
// .create(Flux.range(0, 101))
.create(Flux.range(0, 100))
// .create(Flux.range(0, 99))
.expectSubscription()
.expectNext(0) // 次の要素が指定と等しいことを期待
.expectNextCount(98) // 前の期待またはサブスクリプション開始から、指定数の要素を受信することを期待
.expectNext(99) // 次の要素が指定と等しいことを期待
.expectComplete() // onCompleteシグナルを受信することを期待
.verify(); // ブロッキング検証
expectNextCount()メソッドは一部の問題を解決しますが、特定の状況では発行要素数のチェックだけでは不十分です。例えば、特定ルールで要素をフィルタリングまたは選択するコードを検証する場合、発行されたすべての項目がフィルタリングルールに一致するかをチェックすることが非常に重要です。このため、StepVerifierはJava Hamcrestなどのツールを提供し、発行されたデータとその検証を即座に記録します。以下のコード：
Publisher<Wallet> usersWallets = findAllUsersWallets();
StepVerifier.create(usersWallets)
.expectSubscription()
.recordWith(ArrayList::new)
.expectNextCount(1)
.consumeRecordedWith(
wallets -> assertThat(
wallets,
everyItem(hasProperty("owner", equalTo("admin")))
)
)
.expectComplete()
.verify();
前述の例とは異なり、各期待は1つの要素または指定数要素の検証のみをカバーしますが、.consumeRecordedWith()は指定されたPublisherが発行するすべての要素を検証できます。.consumeRecordedWith()は.recordWith()が指定された場合にのみ有効であることに注意してください。逆に、記録を保存するコレクションクラスを慎重に定義する必要があります。マルチスレッドPublisherの場合、イベント記録用のコレクション型は並行アクセスをサポートする必要があるため、これらの状況ではArrayList::newよりもConcurrentLinkedQueue::newの方が良いでしょう。なぜなら、ArrayListと比べてConcurrentLinkedQueueはスレッドセーフだからです。これ以外にも機能が似たメソッドがあります。例えば、次の要素の期待定義は以下のコードの通り：
StepVerifier
.create(Flux.just("alpha-foo", "betta-bar"))
.expectSubscription()
.expectNextMatches(e -> e.startsWith("alpha"))
.expectNextMatches(e -> e.startsWith("betta"))
.expectComplete()
.verify();
expectNextMatches()と.expectNext()の唯一の違いは、前者がカスタムマッチャーPredicateを定義できることで、後者より柔軟であることです。これは、.expectNext()が要素間の比較に基づいており、この比較は要素の.equals()メソッドを使用するためです。同様に、.assertNext()と.consumeNextWith()はカスタムアサーションの作成を可能にします。.assertNext()は.consumeNextWith()の別名であることに注意してください。.expectNextMatches()と.assertNext()の違いは、前者がtrueまたはfalseを返すPredicateを受け取り、後者が例外をスローする可能性のあるConsumerを受け取り、消費者がスローしたすべてのAssertionErrorをキャッチして.verify()メソッドを通じてスローすることです。以下のコードのように：
StepVerifier
.create(findUsersUSDWallet())
.expectSubscription()
.assertNext(wallet -> assertThat(wallet, hasProperty("currency", equalTo("USD"))))
.expectComplete().verify();
最後に、カバーされていないエラー状況が残っていますが、これは正常システムライフサイクルの一部です。エラーシグナルをチェックするAPIメソッドは多くありませんが、最も簡単なのは引数を持たない.expectError()メソッドです。以下のコードのように：
StepVerifier.create(Flux.error(new RuntimeException("Error")))
.expectError()
.verify();
特定の状況では、特定エラー型のテストが極めて重要です。例えば、ログイン中にユーザーが誤った資格情報を入力した場合、セキュリティサービスはBadCredentialsException.classを発行する必要があります。発行されたエラーを検証するために、`expectError(Class<? extends Throwable>)`を使用できます。以下のコードのように：
StepVerifier.create(securityService.login("admin", "wrong"))
.expectSubscription()
.expectError(BadCredentialsException.class)
.verify();
また、発行されたThrowableと直接インタラクションできる`expectErrorMatches()`と`consumeErrorWith()`という拡張機能も使用できます。
4.2 StepVerifierを使用した高度なテスト
パブリッシャーテストの最初のステップは境界なしパブリッシャーの検証です。レスポンシブストリーム仕様によれば、無限ストリームはSubscriber#onComplete()メソッドを決して呼び出さないことを意味します。StepVerifierは完了シグナルを無期限に待機するため、テストは殺されるまでブロックされます。この問題を解決するために、StepVerifierはキャンセルAPIを提供し、特定の期待が満たされたときにソースのサブスクリプションをキャンセルできます。以下のコードのように:
Flux<String> webSocketPublisher = ...;
StepVerifier.create(websocketPublisher)
.expectSubscription()
.expectNext("Connected")
.expectNext("Price: $12.00")
.thenCancel() // サブスクリプションをキャンセル
.verify();
上記コードは、ConnectedおよびPrice:$ 12.00メッセージを受信した後、WebSocketの切断またはサブスクリプションのキャンセルを行うことを示しています。システム検証プロセスの別の重要な段階はパブリッシャーのバックプレッシャー動作のチェックです。例えば、WebSocketを通じて外部システムとインタラクションすると、プッシュオンリーのパブリッシャーが生成されます。このような動作を防ぐ簡単な方法は.onBackpressureBuffer()オペレーターを使用してダウンストリームを保護することです。選択されたバックプレッシャー戦略でシステムが期待通りに動作するかをチェックするには、ユーザー要求を手動で制御する必要があります。このため、StepVerifierは.thenRequest()メソッドを提供し、ユーザー要求を制御できます。これは以下のコードで説明されます：
Flux<String> websocketPublisher = ...;
Class<Exception> expectedErrorClass =
reactor.core.Exceptions.failWithOverflow().getClass();
StepVerifier.create(websocketPublisher.onBackpressureBuffer(5), 0) // バックプレッシャー制御を使用
.expectSubscription()
.thenRequest(1)
.expectNext("Connected")
.thenRequest(1)
.expectNext("Price: $12.00")
.expectError(expectedErrorClass)
前述の例では、初期サブスクライバーのリクエストを第2引数として受け取るStepVerifier.create()メソッドのオーバーロードを使用しています。単一引数メソッドのオーバーロードでは、デフォルト要求はLong.MAX_VALUE、つまり無限要求です。
Spring 5.xとWebFluxモジュールはApache Velocityを含む多くの技術のサポートを放棄しました。Spring WebFluxはWeb MVCと同じビューレンダリング技術を持っています。以下の例はビュー描画指定の一般的な方法を示しています：
@RequestMapping("/")
public String index() {
// 文字列を論理ビュー名として返す
return "index";
}
テンプレートレンダリングプロセスでリアクティブ手法をどうサポートするか？
大規模音楽プレイリストレンダリングのケースを検討:
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-freemarker</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-webflux</artifactId>
</dependency>
<dependency>
<groupId>org.projectlombok</groupId>
<artifactId>lombok</artifactId>
<optional>true</optional>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>io.projectreactor</groupId>
<artifactId>reactor-test</artifactId>
<scope>test</scope>
</dependency>
@RequestMapping("/play-list-view-ftl")
public Mono<String> getPlaylist(final Model model) {
List<Song> songs = new ArrayList<>();
// 通常は非同期でデータソースからデータを取得
Song song = null;
for (int i = 0; i < 5; i++) {
song = new Song("曲目" + i, "張三" + i, "1001" + i, "アルバム1" + (i %
3));
songs.add(song);
}
// リアクティブストリームに変換
final Flux<Song> playlistStream = Flux.fromIterable(songs);
// データ描画
return playlistStream
// コレクションに収集、ここは非同期ではない
.collectList()
// モデルに配置
.doOnNext(list -> model.addAttribute("playList", list))
// 論理ビュー名を返す
.then(Mono.just("/freemarker/play-list-view"));
}
@Getter
@AllArgsConstructor
public class Song {
private String id;
private String name;
private String artist;
private String album;
}
上記例のように、非同期でビュー名を返すためにリアクティブ型Monoを使用しています。さらに、テンプレートには指定されたSongリストで埋められるプレースホルダーdataSourceがあります。
コンテキストデータを提供する一般的な方法はModelを定義し、必要な属性を配置することです。
FreeMarkerはデータのリアクティブ描画とノンブロッキング描画をサポートしていません。すべての曲をリストに収集し、収集したデータをすべてModelに配置する必要があります。
src/main/resources/templates/freemarker/play-list-view.ftl：
<html>
<body>
<head>
<meta charset="UTF-8"/>
<title>曲目リスト-freemarker</title>
</head>
<table border="1">
<thead>
</thead>
<tbody>
<#list playList as e>
<tr>
<td>${e.id}</td>
<td>${e.name}</td>
<td>${e.artist}</td>
<td>${e.album}</td>
</tr>
</#list>
</tbody>
</table>
</body>
</html>
src/main/java/com/webflux/demo/config/WebConfig.java：
@Configuration
@EnableWebFlux
public class MyConfig implements WebFluxConfigurer {
@Override
public void configureViewResolvers(ViewResolverRegistry registry) {
// FreeMakrerViewResolverテンプレートを登録
registry.freeMarker();
}
@Bean
public FreeMarkerConfigurer freeMarkerConfigurer() {
// 設定子を設定
final FreeMarkerConfigurer configurer = new FreeMarkerConfigurer();
// テンプレートパスを設定
configurer.setTemplateLoaderPath("classpath:/templates");
return configurer;
}
}