- Spring
Spring WebFlux入門|Mono・Fluxの基本とMVCとの違い
Spring WebFlux は、少ない数のスレッドで多くのリクエストをさばくための、ノンブロッキング(待たずに次へ進む)な Web フレームワークです。 値は Mono(0〜1件)と Flux(0〜N件)で表し、map・flatMap などの演算子をつないで処理を組み立てます。subscribe されるまで何も動かないこと、イベントループのスレッドで待ってはいけないことの2点が、Spring MVC との一番の違いです。
@RestController
@RequestMapping("/api/users")
public class UserController {
private final WebClient client;
public UserController(WebClient.Builder builder) {
this.client = builder.baseUrl("http://localhost:8081").build();
}
@GetMapping("/{id}")
public Mono<UserView> get(@PathVariable long id) {
Mono<User> user = client.get().uri("/users/{id}", id).retrieve().bodyToMono(User.class);
Mono<Points> points = client.get().uri("/points/{id}", id).retrieve().bodyToMono(Points.class);
return Mono.zip(user, points) // 2つを同時に呼ぶ
.map(t -> new UserView(t.getT1().name(), t.getT2().value()))
.timeout(Duration.ofSeconds(2));
}
}
| 選び方の目安 | 選ぶもの |
|---|---|
| JPA・JDBC など、待つ(ブロッキングの)ライブラリが中心 | Spring MVC(Java 21 以降なら仮想スレッドも) |
| 外部 API を何本も同時に呼ぶ・ストリーミング(SSE)・WebSocket が中心 | Spring WebFlux |
| 既存の MVC アプリで、外部 API の呼び出しだけを効率よくしたい | MVC のまま WebClient(または RestClient) |
| DB も含めて全部をノンブロッキングにできる(R2DBC・リアクティブ版のドライバー) | Spring WebFlux |
確認環境:Spring Boot 4.1.1(Spring Framework 7.0.9、Reactor BOM 2025.0.7 の reactor-core 3.8.7・reactor-netty 1.3.7 は Boot の依存管理の版)。reactor-core 3.8.7 と reactor-test 3.8.7 は GitHub のタグのソースからビルドし、OpenJDK 21.0.10(Linux、CPU 2コア)で、この記事の演算子・エラー処理・スケジューラー・block()・StepVerifier の動きを実行して記録しました。WebFlux・WebClient の設定と既定値は、Spring Boot v4.1.1・Spring Framework v7.0.9・reactor-netty v1.3.7 のソースとリファレンスで確かめています(確認日 2026年10月13日)。
本記事の内容は、ご自由にお使いください。
ご利用の際は、出典として本ページへのリンクを記載いただけますようお願いします。
(記載例)出典:株式会社RJC「Spring WebFlux入門|Mono・Fluxの基本とMVCとの違い」
WebFlux と MVC の違い
| 項目 | Spring MVC | Spring WebFlux |
|---|---|---|
| 書き方 | ふつうの戻り値(User・List<User>) |
Mono<User>・Flux<User> |
| サーバー(Boot の既定) | Tomcat | Reactor Netty |
| リクエストを処理するスレッド | リクエストごとに1本(Tomcat は最大 200) | 少数のイベントループ(CPU の数、最低 4) |
| 待つ処理(DB・外部 API) | そのまま書いてよい | 待たない API を使う。待つなら別スレッドへ |
| DB | JPA・JDBC・MyBatis | R2DBC、リアクティブ版の MongoDB・Redis |
| 外部 API の呼び出し | RestClient(WebClient も使える) |
WebClient |
| テスト | MockMvc |
WebTestClient・StepVerifier |
| Boot 4 のスターター | spring-boot-starter-webmvc |
spring-boot-starter-webflux |
数字はソースで確かめた既定値です。Tomcat の server.tomcat.threads.max は 200(Boot v4.1.1)、Reactor Netty のイベントループのスレッド数は Math.max(CPU の数, 4)(reactor-netty v1.3.7 の LoopResources、reactor.netty.ioWorkerCount で変更)でした。
Spring Framework のリファレンスは、選び方について次のように書いています。
- うまく動いている Spring MVC のアプリは、変える必要はない
- JPA・JDBC などのブロッキングな永続化 API を使うなら、Spring MVC が一般的な構成ではもっともよい選択
- MVC のアプリで外部サービスを呼んでいるなら、まずリアクティブな
WebClientを試す - 大きなチームでは、ノンブロッキング・関数型の書き方へ移る学習の負担が大きいことを考える
両方の依存を入れたら MVC になる
Boot 4.1.1 のソースでは、WebMvcWebApplicationTypeDeducer(順番 10)が WebFluxWebApplicationTypeDeducer(順番 20)より先に判定します。spring-boot-starter-webmvc と spring-boot-starter-webflux を両方入れると、MVC(サーブレット)のアプリとして起動します。 MVC のアプリで WebClient だけ使う、という組み合わせはこれで成り立ちます。WebFlux として起動したいときは spring.main.web-application-type=reactive を書きます。
依存(Boot 4)
// build.gradle.kts(Spring Boot 4.1.1)
dependencies {
implementation("org.springframework.boot:spring-boot-starter-webflux")
implementation("org.springframework.boot:spring-boot-starter-webclient") // WebClient.Builder を使うなら
testImplementation("org.springframework.boot:spring-boot-starter-webflux-test")
}
Boot 4 では、WebClient.Builder の自動設定が spring-boot-webclient に分かれました。 v4.1.1 のソースでは、spring-boot-starter-webflux は spring-boot-webflux・spring-boot-reactor・spring-boot-starter-reactor-netty などを含みますが、WebClientAutoConfiguration は spring-boot-webclient(spring-boot-starter-webclient)にしかありません。WebFlux のスターターだけでは、WebClient.Builder を引数で受け取れません。
| スターター | 入るもの |
|---|---|
spring-boot-starter-webflux |
WebFlux、Reactor Netty(サーバー)、Jackson |
spring-boot-starter-webclient |
WebClient.Builder の自動設定、Reactor Netty(クライアント) |
spring-boot-starter-webflux-test |
WebTestClient、reactor-test(StepVerifier) |
spring-boot-starter-data-r2dbc |
R2DBC でのリアクティブな DB アクセス |
Mono と Flux の基本
subscribe するまで何も動かない
Flux を作っただけでは、中の処理は1行も動きません。subscribe(購読)したときに初めて流れ始めます。ソースからビルドした reactor-core 3.8.7 で動かした結果です。
Flux<Integer> f = Flux.range(1, 3).doOnNext(i -> p("doOnNext " + i)).map(i -> i * 10);
p("Flux を作っただけ");
f.subscribe(i -> p("受け取った " + i), e -> p("エラー " + e), () -> p("完了"));
[ 103ms main] Flux を作っただけ
[ 135ms main] doOnNext 1
[ 137ms main] 受け取った 10
[ 137ms main] doOnNext 2
[ 138ms main] 受け取った 20
[ 138ms main] doOnNext 3
[ 138ms main] 受け取った 30
[ 139ms main] 完了
WebFlux では、コントローラーが返した Mono・Flux をフレームワークが subscribe します。subscribe() はフレームワークに任せ、コントローラーは Mono・Flux を返すだけにします。 コントローラーの中で subscribe() を呼ぶと、応答とは別に勝手に動き出し、エラーも応答に反映されません。
just と defer の違い
Mono<Long> eager = Mono.just(System.nanoTime()); // 作った時点で値が決まる
Mono<Long> lazy = Mono.defer(() -> Mono.just(System.nanoTime())); // subscribe のたびに作る
just 1回目=1401802480928 2回目=1401802480928
defer 1回目=1401834543114 2回目=1401840828745
Mono.just(repository.findById(id)) のように書くと、subscribe を待たずにその場で検索が走ります。遅らせたい処理は Mono.defer・Mono.fromCallable・Mono.fromSupplier で包みます。
よく使う演算子と実行結果
| 演算子 | 書き方 | 実行結果 |
|---|---|---|
filter・map |
Flux.range(1, 10).filter(i -> i % 2 == 0).map(i -> i * i) |
[4, 16, 36, 64, 100] |
skip・take |
Flux.range(1, 10).skip(2).take(3) |
[3, 4, 5] |
reduce |
Flux.range(1, 10).reduce(0, Integer::sum) |
55 |
buffer |
Flux.range(1, 7).buffer(3) |
[[1, 2, 3], [4, 5, 6], [7]] |
zip |
Flux.zip(Flux.just("a","b","c"), Flux.range(1, 2), (s, i) -> s + i) |
[a1, b2](短いほうに合わせる) |
concat |
Flux.concat(Flux.just(1, 2), Flux.just(3)) |
[1, 2, 3] |
distinct |
Flux.just(1, 1, 2, 3, 3).distinct() |
[1, 2, 3] |
collectMap |
Flux.just("apple","banana","cherry").collectMap(s -> s.charAt(0)) |
{a=apple, b=banana, c=cherry} |
Mono.zip |
Mono.zip(Mono.just("user"), Mono.just(42)) |
user:42 |
defaultIfEmpty |
Mono.empty().defaultIfEmpty("なし") |
なし |
switchIfEmpty |
Mono.empty().switchIfEmpty(Mono.fromSupplier(() -> "DBから取得")) |
DBから取得 |
空の block() |
Mono.empty().block() |
null |
map と flatMap と concatMap
map は値を値に変えます。値を Mono・Flux(あとで届く値)に変えるときは flatMap です。map で Mono を返すと Mono<Mono<T>> になり、中の処理は動きません。
flatMap と concatMap は、順番と同時に動く数が違います。100〜300ms 遅れて届く値(3・1・2 の順に開始)で試した結果です。
| 演算子 | 結果の順番 | 10件×200ms の所要時間 |
|---|---|---|
flatMap |
[1, 2, 3](終わった順) |
203ms(全部同時) |
flatMap(…, 2)(同時に2件まで) |
終わった順 | 1004ms |
concatMap |
[3, 1, 2](元の順) |
2004ms(1件ずつ) |
flatMapSequential |
[3, 1, 2](元の順) |
同時に動かし、順番だけ元に戻す |
flatMap の同時実行数の既定は、Queues.SMALL_BUFFER_SIZE(既定 256)です。外部 API を1万件呼ぶような処理では、flatMap(…, 10) のように上限を書いて、相手に一度に送りすぎないようにします。
2つの API を同時に呼ぶ
300ms かかる呼び出し2つを、flatMap で順に呼ぶ場合と Mono.zip で同時に呼ぶ場合です。
user(1).flatMap(u -> points(1).map(pt -> u + ":" + pt)) // 順に
Mono.zip(user(1), points(1)).map(t -> t.getT1() + ":" + t.getT2()) // 同時に
| 書き方 | 所要時間 |
|---|---|
flatMap で順に |
609ms |
Mono.zip で同時に |
302ms |
前の結果を使わない呼び出しは、Mono.zip で同時に呼ぶと、遅いほうの時間で終わります。
エラー処理
Flux.just(1, 2, 0, 4).map(i -> 10 / i) は、3件目で ArithmeticException になります。エラーが起きると、その Flux はそこで終わり、後ろの値(4)は流れません。 演算子ごとの結果です。
| 演算子 | 結果 | 使いどころ |
|---|---|---|
(何もしない)+block() |
ArithmeticException: / by zero が投げられる |
― |
onErrorReturn(-1) |
[10, 5, -1] |
決まった値で終える |
onErrorResume(ArithmeticException.class, e -> Flux.just(-1, -2)) |
[10, 5, -1, -2] |
別の処理に切り替える(キャッシュを返すなど) |
onErrorMap(e -> new IllegalStateException("計算できません", e)) |
IllegalStateException: 計算できません |
例外を業務の例外に変える |
onErrorContinue(...) |
[10, 5, 2](0 を飛ばす) |
使える演算子が限られる。下の書き方を使う |
flatMap の内側で onErrorResume(e -> Mono.empty()) |
[10, 5, 2] |
1件の失敗で全体を止めない |
1件ずつの失敗を飛ばしたいときは、onErrorContinue より、flatMap の内側で処理する書き方がわかりやすく、どの演算子でも同じように動きます。
Flux.just(1, 2, 0, 4)
.flatMap(i -> Mono.fromCallable(() -> 10 / i)
.onErrorResume(e -> Mono.empty())) // 失敗した1件だけ捨てる
retry と timeout
Mono<String> flaky = Mono.fromCallable(() -> callApi()); // 2回失敗して3回目に成功する処理
flaky.retry(3); // すぐに再試行
flaky.retryWhen(Retry.backoff(3, Duration.ofMillis(100))); // 待ち時間を伸ばしながら再試行
[4734ms main ] 呼び出し 1
[4869ms parallel-2] 呼び出し 2
[4975ms parallel-1] 呼び出し 3
[4976ms main ] retryWhen backoff 成功
Retry.backoff は待ち時間を伸ばしながら(ここでは約135ms・106ms、ばらつきあり)再試行し、2回目以降は parallel のスレッドで動きました。回数を使い切ると、次の例外になります。
reactor.core.Exceptions$RetryExhaustedException: Retries exhausted: 2/2
cause=java.lang.IllegalStateException: 一時的な失敗 -7
timeout で時間を区切った結果です。
reactor.core.Exceptions$ReactiveException: java.util.concurrent.TimeoutException: Did not observe any item or terminal signal within 300ms in 'source(MonoDelay)' (and no fallback has been configured)
timeout(Duration.ofMillis(300), Mono.just("キャッシュ")) のように代わりを渡すと、時間切れで キャッシュ を返しました。block() は、チェック例外(TimeoutException など)を ReactiveException で包んで投げます。 元の例外は getCause() で取れます。再試行の考え方は、関連記事「Spring Retry・Resilience4jで再試行|@Retryable設定」で扱っています。
doOnXxx と doFinally
Mono.just("x")
.doOnSubscribe(s -> log("doOnSubscribe"))
.doOnSuccess(v -> log("doOnSuccess " + v))
.doFinally(sig -> log("doFinally " + sig))
.block();
doOnSubscribe
doOnSuccess x
doFinally onComplete
doFinally は成功・エラー・キャンセルのどれでも呼ばれ、引数で理由(onComplete・onError・cancel)がわかります。
スケジューラー:subscribeOn と publishOn
既定の大きさ
| スケジューラー | 用途 | 既定の大きさ(CPU 2コアでの値) |
|---|---|---|
Schedulers.parallel() |
CPU を使う短い処理 | CPU の数(2) |
Schedulers.boundedElastic() |
待つ処理(ブロッキング I/O) | CPU の数×10(20)、待ち行列 100,000 |
Schedulers.single() |
1本のスレッドで順に | 1 |
Schedulers.immediate() |
呼んだスレッドのまま | ― |
Schedulers.DEFAULT_BOUNDED_ELASTIC_SIZE・DEFAULT_BOUNDED_ELASTIC_QUEUESIZE を出力した値です。
どこで動くか
Flux.range(1, 2)
.doOnNext(i -> p("A(発生) " + i))
.publishOn(Schedulers.parallel())
.doOnNext(i -> p("B(publishOn の後) " + i))
.subscribeOn(Schedulers.boundedElastic())
.doOnNext(i -> p("C(最後) " + i))
.blockLast();
[ 332ms boundedElastic-1] A(発生) 1
[ 335ms parallel-1 ] B(publishOn の後) 1
[ 335ms boundedElastic-1] A(発生) 2
[ 336ms parallel-1 ] C(最後) 1
[ 337ms parallel-1 ] B(publishOn の後) 2
[ 339ms parallel-1 ] C(最後) 2
| 演算子 | 影響する範囲 | 書く場所 |
|---|---|---|
subscribeOn |
発生元(fromCallable など)から。書く場所に関係なく効く |
どこでもよい。2つ書くと発生元に近いほうが効く |
publishOn |
書いた場所より後ろの演算子だけ | 切り替えたい位置に書く |
subscribeOn を2回書いた結果は boundedElastic-1 で、発生元に近いほう(先に書いたほう)が効きました。
待つ処理は boundedElastic に逃がす
JDBC やファイル、RestTemplate などの同期の HTTP クライアントのような待つ処理(ブロッキング I/O)を WebFlux で使うときは、boundedElastic に逃がします。
public Mono<User> findUser(long id) {
return Mono.fromCallable(() -> jdbcTemplate.queryForObject(SQL, mapper, id)) // 待つ処理
.subscribeOn(Schedulers.boundedElastic()); // 別のスレッドで
}
「100ms かかる処理」を 100件流した結果です(CPU 2コア)。
| 書き方 | 所要時間 |
|---|---|
ノンブロッキング(Mono.delay(100ms)) |
108ms |
Thread.sleep(100) をそのまま(呼んだスレッドで1件ずつ) |
10,032ms |
Thread.sleep(100) を parallel(2本)で |
5,020ms |
Thread.sleep(100) を boundedElastic(20本)で |
507ms |
待つ処理はスレッドを占有するので、所要時間は「件数÷スレッドの数×100ms」になりました。ノンブロッキングなら、スレッドを占有しないので 100件でも約100ms です。 これが WebFlux が少ないスレッドで多くのリクエストをさばけるしくみです。
parallel やイベントループを埋めると、ほかの処理も止まる
parallel の2本のスレッドを Thread.sleep(1000) で埋めてから、10ms のタイマー(Mono.delay、これも parallel で動く)を動かしました。
10ms のタイマーが返るまで 1001ms
関係のない処理まで1秒待たされました。 WebFlux のイベントループで待つ処理を書くと、同じスレッドを使うほかのリクエストすべてが同じように待たされます。
仮想スレッドで boundedElastic を動かす
reactor-core 3.8.7 は、Java 21 以降で -Dreactor.schedulers.defaultBoundedElasticOnVirtualThreads=true を付けると、boundedElastic が仮想スレッドで動きます。付けて動かした結果です。
[148ms loomBoundedElastic-1] VirtualThread[#18,loomBoundedElastic-1]/runnable@ForkJoinPool-1-worker-1
block() と、よくある誤り
イベントループで block() すると例外
parallel のスレッドの中で block() を呼んだ結果です。
java.lang.IllegalStateException: block()/blockFirst()/blockLast() are blocking, which is not supported in thread parallel-1
Reactor は parallel・single のスレッドと、Reactor Netty のイベントループのスレッド(reactor-netty v1.3.7 の DefaultLoopResources.EventLoop は NonBlocking を実装)を「待ってはいけないスレッド」として扱い、そこで block() すると例外にします。WebFlux のコントローラーやフィルターの中で block() を呼ぶと、このエラーになります。
なお、Mono.just(1).block() のようにすでに値がある Mono は待たずに返るので、例外になりませんでした。テストでは通っても、本番で外部呼び出しが入ると例外になる、ということが起きます。
よくある誤りの一覧
| 誤り | 何が起きるか | 直し方 |
|---|---|---|
コントローラーで block() |
block()/blockFirst()/blockLast() are blocking… |
Mono をそのまま返す。つなぐなら flatMap |
| JDBC・JPA をそのまま呼ぶ | イベントループが止まり、全体が遅くなる | fromCallable(…).subscribeOn(boundedElastic())、または R2DBC |
RestTemplate・RestClient を WebFlux の中で使う |
同上 | WebClient |
map で Mono を返す |
Mono<Mono<T>> になり、中が動かない |
flatMap |
自分で subscribe() を呼ぶ |
応答と関係なく動く。エラーが返らない | 返り値としてつなぐ |
Mono.just(重い処理()) |
subscribe 前にその場で動く |
Mono.fromCallable(…)・Mono.defer(…) |
ThreadLocal で値を渡す |
スレッドが変わると消える | Reactor の Context(下) |
flatMap で外部 API を大量に呼ぶ |
既定で最大 256 件を同時に送る | flatMap(…, 並行数) |
onErrorResume で何も返さず握りつぶす |
失敗に気づけない | ログを出してから代わりを返す |
| 大きな JSON を受け取る | DataBufferLimitException: Exceeded limit on max bytes to buffer : 262144 |
spring.http.codecs.max-in-memory-size(既定 256KB) |
最後の項目の既定値 256KB は Spring Framework v7.0.9 の AbstractDataBufferDecoder、プロパティ名は Boot v4.1.1 の HttpCodecsProperties(spring.http.codecs)で確かめました。Boot 3 の spring.codec.max-in-memory-size から名前が変わっています。
ThreadLocal の代わりに Context
Mono.deferContextual(ctx -> Mono.just("traceId=" + ctx.getOrDefault("traceId", "none")))
.publishOn(Schedulers.parallel())
.map(v -> v + " on " + Thread.currentThread().getName())
.contextWrite(Context.of("traceId", "abc123"))
.block();
traceId=abc123 on parallel-1
ThreadLocal=null
Context の値はスレッドが変わっても届き、ThreadLocal に入れた値は別スレッドでは null でした。Boot 4.1.1 には spring.reactor.context-propagation があります。既定の limited では tap・handle の演算子だけで、auto ではすべての演算子で、Context の値(Micrometer のトレース ID など)を ThreadLocal へ受け渡します(ReactorProperties の説明文)。
WebFlux のコード(完成形)
@RestController で書く
package com.example.demo;
import org.springframework.http.HttpStatus;
import org.springframework.http.MediaType;
import org.springframework.web.bind.annotation.*;
import org.springframework.web.server.ResponseStatusException;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import java.time.Duration;
@RestController
@RequestMapping("/api/products")
public class ProductController {
private final ProductRepository repository; // ReactiveCrudRepository(R2DBC)
public ProductController(ProductRepository repository) {
this.repository = repository;
}
@GetMapping
public Flux<Product> list() {
return repository.findAll();
}
@GetMapping("/{id}")
public Mono<Product> get(@PathVariable long id) {
return repository.findById(id)
.switchIfEmpty(Mono.error(new ResponseStatusException(HttpStatus.NOT_FOUND)));
}
@PostMapping
@ResponseStatus(HttpStatus.CREATED)
public Mono<Product> create(@RequestBody Mono<Product> body) {
return body.flatMap(repository::save);
}
// SSE(Server-Sent Events):1秒ごとに値を送り続ける
@GetMapping(value = "/ticks", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
public Flux<String> ticks() {
return Flux.interval(Duration.ofSeconds(1)).map(i -> "tick " + i).take(10);
}
}
package com.example.demo;
import org.springframework.data.annotation.Id;
import org.springframework.data.repository.reactive.ReactiveCrudRepository;
record Product(@Id Long id, String name, int price) {}
interface ProductRepository extends ReactiveCrudRepository<Product, Long> {}
findById は見つからないとき空の Mono を返し、そのまま返すと 200 で本文が空になります。404 にしたいときは、上のように switchIfEmpty(Mono.error(…)) を書きます。
関数型ルーティング(RouterFunction)で書く
package com.example.demo;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.web.reactive.function.server.*;
import reactor.core.publisher.Mono;
import static org.springframework.web.reactive.function.server.RequestPredicates.accept;
import static org.springframework.http.MediaType.APPLICATION_JSON;
@Configuration
public class ProductRoutes {
@Bean
RouterFunction<ServerResponse> productRouter(ProductHandler handler) {
return RouterFunctions.route()
.path("/v2/products", b -> b
.GET("/{id}", accept(APPLICATION_JSON), handler::get)
.GET(handler::list) // /v2/products
.POST(handler::create))
.filter((req, next) -> next.handle(req)
.doOnNext(res -> System.out.println(req.path() + " -> " + res.statusCode())))
.build();
}
}
@org.springframework.stereotype.Component
class ProductHandler {
private final ProductRepository repository;
ProductHandler(ProductRepository repository) {
this.repository = repository;
}
Mono<ServerResponse> list(ServerRequest req) {
return ServerResponse.ok().body(repository.findAll(), Product.class);
}
Mono<ServerResponse> get(ServerRequest req) {
long id = Long.parseLong(req.pathVariable("id"));
return repository.findById(id)
.flatMap(p -> ServerResponse.ok().bodyValue(p))
.switchIfEmpty(ServerResponse.notFound().build());
}
Mono<ServerResponse> create(ServerRequest req) {
return req.bodyToMono(Product.class)
.flatMap(repository::save)
.flatMap(saved -> ServerResponse.created(
req.uriBuilder().path("/{id}").build(saved.id())).bodyValue(saved));
}
}
| 項目 | @RestController |
RouterFunction |
|---|---|---|
| ルートの定義 | 注釈(@GetMapping) |
コード(route().GET(…)) |
| 引数の受け取り | @PathVariable・@RequestBody が自動で変換 |
req.pathVariable()・req.bodyToMono() を自分で呼ぶ |
| 入力チェック | @Valid が使える |
Validator を自分で呼ぶ |
| 共通処理 | @ControllerAdvice・WebFilter |
.filter(…) をルートごとに書ける |
| 向いている場面 | MVC から移る、チームで書き方をそろえる | 小さなサービス、ルートを一か所で見たい |
WebClient の使い方
package com.example.demo;
import org.springframework.http.HttpStatusCode;
import org.springframework.stereotype.Component;
import org.springframework.web.reactive.function.client.WebClient;
import org.springframework.web.reactive.function.client.WebClientResponseException;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.util.retry.Retry;
import java.time.Duration;
@Component
public class InventoryClient {
private final WebClient client;
public InventoryClient(WebClient.Builder builder) { // Boot が用意する Builder を使う
this.client = builder.baseUrl("https://inventory.example.com").build();
}
public Mono<Stock> stock(String sku) {
return client.get()
.uri("/stocks/{sku}", sku) // 変数で渡す(メトリクスの uri がまとまる)
.retrieve()
.onStatus(HttpStatusCode::is4xxClientError,
res -> res.bodyToMono(String.class).map(body -> new IllegalArgumentException(body)))
.bodyToMono(Stock.class)
.timeout(Duration.ofSeconds(2))
.retryWhen(Retry.backoff(2, Duration.ofMillis(200))
.filter(e -> e instanceof WebClientResponseException.ServiceUnavailable));
}
public Flux<Stock> stocks(Flux<String> skus) {
return skus.flatMap(this::stock, 8); // 同時に8件まで
}
public record Stock(String sku, int quantity) {}
}
| メソッド | 返すもの |
|---|---|
retrieve().bodyToMono(T.class) |
本文(4xx・5xx は WebClientResponseException) |
retrieve().bodyToFlux(T.class) |
本文の配列・ストリーム |
retrieve().toEntity(T.class) |
ステータス・ヘッダー・本文 |
retrieve().toBodilessEntity() |
ステータス・ヘッダーだけ |
exchangeToMono(res -> …) |
応答を自分で読む(ステータスで分岐など) |
MVC のアプリから外部 API を呼ぶだけなら、同期の RestClient も選べます。3つのクライアントの違いは、関連記事「RestClient・WebClient・RestTemplateの違い|移行対応表」で扱っています。
仮想スレッドとの使い分け
Java 21 以降の仮想スレッドは、待つ処理をふつうに書いても、待っている間にスレッドを手放す軽いスレッドです。Spring MVC で spring.threads.virtual.enabled=true にすると、リクエストを仮想スレッドで処理します。
| 観点 | MVC+プラットフォームスレッド | MVC+仮想スレッド | WebFlux |
|---|---|---|---|
| 書き方 | ふつう | ふつう | Mono・Flux |
| 同時に待てる数 | スレッド数(既定 200)まで | 多い | 多い |
| JPA・JDBC | そのまま | そのまま(接続プールの数が上限) | R2DBC か別スレッド |
| ストリーミング・背圧 | 不向き | 不向き | 得意 |
| 外部 API を同時に何本も | スレッドを分ける必要がある | 書き方を工夫すれば可 | zip・flatMap で自然に書ける |
| デバッグ・スタックトレース | 読みやすい | 読みやすい | 読みにくい |
Boot 4.1.1 の WebFlux では、spring.threads.virtual.enabled=true にすると、Mono・Flux を返さない(ふつうの戻り値の)コントローラーのメソッドを、仮想スレッドの applicationTaskExecutor で動かします(WebFluxAutoConfiguration の configureBlockingExecution、Spring Framework 7.0.9 の BlockingExecutionConfigurer)。WebFlux のアプリの一部に待つ処理が残っている場合の逃げ道になります。仮想スレッドの使いどころは、関連記事「Java 仮想スレッド入門|10万タスク実測と使いどころ判断表」で扱っています。
背圧(バックプレッシャー)
背圧は、受け取る側が「あと何件まで受け取れる」と伝え、送る側がそれ以上送らないしくみです。limitRate(4) を付けて、送る側に届いた要求(request)を記録しました。
[ INFO] (main) | request(4)
[ INFO] (main) | request(3)
[ INFO] (main) | request(3)
[ INFO] (main) | request(3)
最初に4件を要求し、そのあとは4件の75%(3件)を受け取るたびに3件ずつ追加で要求しました。DB や外部から大量に読むとき、受け取る側の速さに合わせて読み進められます。
テスト
StepVerifier
reactor-test の StepVerifier で、Mono・Flux が流す値と終わり方を確かめます。
StepVerifier.create(Flux.just(1, 2, 0).map(i -> 10 / i))
.expectNext(10, 5)
.expectError(ArithmeticException.class)
.verify();
期待と違うと、次のメッセージで失敗しました。
expectation "expectNext(3)" failed (expected value: 3; actual value: 2)
時間のかかる処理は仮想時間で確かめます。 1時間待つ Mono.delay を、7ms で検証できました。
StepVerifier.withVirtualTime(() -> Mono.delay(Duration.ofHours(1)).thenReturn("done"))
.expectSubscription()
.expectNoEvent(Duration.ofMinutes(59))
.thenAwait(Duration.ofMinutes(1))
.expectNext("done")
.verifyComplete();
WebTestClient
@WebFluxTest(ProductController.class)
class ProductControllerTest {
@Autowired
WebTestClient client;
@MockitoBean
ProductRepository repository;
@Test
void notFound() {
when(repository.findById(99L)).thenReturn(Mono.empty());
client.get().uri("/api/products/99")
.exchange()
.expectStatus().isNotFound();
}
}
テスト全体の考え方は、関連記事「Spring Bootテスト入門|@WebMvcTestとMockMvcの使い分け」で扱っています。
よくある質問
Q. WebFlux にすると速くなりますか。 A. 1件の処理が速くなるわけではありません。上の測定のとおり、待つ処理が多いときに、少ないスレッドで多くのリクエストを同時に扱えるのが利点です。CPU を使う計算が中心の処理や、JDBC が中心のアプリでは、利点は小さくなります。
Q. MVC のコントローラーから Mono を返してもよいですか。
A. 返せます。Spring Framework のリファレンスにも、MVC のコントローラーからリアクティブ型をそのまま返せると書かれています。WebClient の結果を block() せずに返す使い方ができます。
Q. WebFlux で JPA は使えますか。
A. 動かすことはできますが、JPA は待つ処理なので、boundedElastic に逃がす必要があります。全体をノンブロッキングにするなら、spring-boot-starter-data-r2dbc で R2DBC を使います。
Q. Kotlin のコルーチンでも書けますか。
A. 書けます。WebFlux のコントローラーは suspend fun と Flow を受け付けます。
まとめ
- WebFlux は少ないスレッドで待ち時間の多い処理をさばくための仕組み。速さではなく同時に扱える数が利点
Mono・Fluxはsubscribeするまで動かない。Mono.just(重い処理())はその場で動く- 値を
Monoに変えるときはflatMap。前の結果を使わない呼び出しはMono.zipで同時に(609ms → 302ms) subscribeOnは発生元から、publishOnは書いた場所から後ろのスレッドを変える- 待つ処理は
boundedElasticへ。parallelで 100件×100ms は 5秒、boundedElasticで 0.5秒、ノンブロッキングなら 0.1秒 - イベントループで
block()するとblock()/blockFirst()/blockLast() are blocking… - Boot 4 では
WebClient.Builderにspring-boot-starter-webclientが必要。両方のスターターを入れると MVC で起動する - JPA・JDBC 中心なら MVC(+仮想スレッド)、ストリーミングや多数の外部呼び出しなら WebFlux
参考・出典
確認日はいずれも 2026年10月13日です。
- Spring Framework Reference「Spring WebFlux」:https://docs.spring.io/spring-framework/reference/web/webflux.html
- Spring Framework Reference「WebClient」:https://docs.spring.io/spring-framework/reference/web/webflux-webclient.html
- Spring Framework Reference「Functional Endpoints」:https://docs.spring.io/spring-framework/reference/web/webflux-functional.html
- Spring Framework v7.0.9 ソース(
BlockingExecutionConfigurer・AbstractDataBufferDecoder):https://github.com/spring-projects/spring-framework/tree/v7.0.9 - Spring Boot Reference「Reactive Web Applications」「Calling REST Services」:https://docs.spring.io/spring-boot/reference/web/reactive.html
- Spring Boot v4.1.1 ソース(
WebFluxAutoConfiguration・WebClientAutoConfiguration・WebFluxWebApplicationTypeDeducer・HttpCodecsProperties):https://github.com/spring-projects/spring-boot/tree/v4.1.1 - Reactor 3 Reference Guide:https://projectreactor.io/docs/core/release/reference/
- reactor-core v3.8.7 ソース:https://github.com/reactor/reactor-core/tree/v3.8.7
- reactor-netty v1.3.7 ソース(
LoopResources・DefaultLoopResources):https://github.com/reactor/reactor-netty/tree/v1.3.7
本記事の内容は、ご自由にお使いください。
ご利用の際は、出典として本ページへのリンクを記載いただけますようお願いします。
(記載例)出典:株式会社RJC「Spring WebFlux入門|Mono・Fluxの基本とMVCとの違い」
株式会社RJC ― SI事業・SES事業・AI駆動開発。RJCは一緒に成長を楽しめる会社です。
WE ARE HIRING
RJCで一緒に開発しながら、
成長を楽しみませんか?
RJCは、Web・モバイル・AIを活用した開発プロジェクトで、テックリードやPM・PMOも活躍するシステム開発会社です。会社を知る、待遇を確かめる、話を聞いてみる。気になるところ見てみてください!
- 127日年間休日
- 12時間平均残業時間
- 毎日ガチャ遊びココロも大切にする福利厚生。アマギフなどの賞品ラインナップ!
ほかにも、チケットレストラン、書籍読み放題、2年ごとの慰労報奨(休暇 or 金一封)、11期連続の黒字決算。