EnglishEnglish العربيةالعربية PortuguêsPortuguês 日本語日本語
404 Not Found

404 Not Found

nginx

Spring WebFlux リアクティブプログラミング

WebFluxはSpringの非同期エンジンです。MonoとFluxを中心に, ノンブロッキングI/Oで数万の同時接続を処理します。

1. 学ぶこと

2. 高並列アーキテクトの実話

(1) ペインポイント: スレッドプールの枯渇

AliceのOrderFlowはセールイベント中にボトルネックに直面しました。500人の同時ユーザーが注文を照会すると, Tomcatのスレッドプール (デフォルト200スレッド)が枯渇し, リクエストがキューに入り, P99レイテンシが3秒に急増しました。スレッドの大部分の時間はI/O (データベース, ネットワーク)の待機に費やされ, CPU使用率はわずか15%でした。Bobはスレッド数を増やすことを提案しましたが, 1,000スレッドを追加すると1GBのメモリが追加消費されます。

(2) WebFluxソリューション

WebFluxは少数のスレッドで高並列を処理します。I/O待機中にスレッドはブロックせず, 他のリクエストの処理を継続できます:

JAVA 
@GetMapping("/{id}")
public Mono<OrderResponse> getOrder(@PathVariable Long id) {
    return orderRepository.findById(id)
        .map(OrderResponse::from)
        .switchIfEmpty(Mono.error(new ResourceNotFoundException("Order", id)));
}

(3) 成果

AliceがWebFluxで照会APIをリファクタリングした後, 4スレッドで2,000の同時リクエストを処理でき, メモリ使用量は70%削減, P99レイテンシは3秒から100msに低下, CPU使用率は60%に向上しました。

3. リアクティブプログラミングの核心概念

(1) MonoとFlux

型 意味 例え 例
Mono<T> 0または1要素 Optionalの非同期版 単一注文の照会
Flux<T> 0〜N要素 Streamの非同期版 注文一覧の照会
Mono<Void> 戻り値なし 完了シグナル 削除操作
100% 
graph LR
    A["Publisher"] --> B["Mono<br/>0または1要素"]
    A --> C["Flux<br/>0〜N要素"]
    B --> D["onNext → onComplete"]
    C --> E["onNext x N → onComplete"]
    B --> F["onError"]
    C --> F

(1) ▶ サンプル:MonoとFluxの基本操作

JAVA 
// Mono: 単一値
Mono<String> mono = Mono.just("Hello OrderFlow");
Mono<String> empty = Mono.empty();
Mono<String> fromCallable = Mono.fromCallable(() -> fetchOrder(1L));

// Flux: 複数値
Flux<Integer> flux = Flux.just(1, 2, 3, 4, 5);
Flux<Integer> range = Flux.range(1, 100);
Flux<Long> interval = Flux.interval(Duration.ofSeconds(1));

出力:

TEXT 
// 実行成功

(2) 主要オペレーター

オペレーター 機能 例
map 要素変換 .map(OrderResponse::from)
flatMap 非同期変換 .flatMap(this::enrichOrder)
filter フィルタリング .filter(order -> "PENDING".equals(order.getStatus()))
switchIfEmpty 空値の代替 .switchIfEmpty(Mono.error(...))
onErrorResume エラー時のフォールバック .onErrorResume(e -> fallbackOrder())
timeout タイムアウト制御 .timeout(Duration.ofSeconds(5))
retry リトライ .retry(3)
subscribeOn 購読スレッドの指定 .subscribeOn(Schedulers.boundedElastic())

(2) ▶ サンプル:オペレーターチェーン

JAVA 
public Mono<OrderResponse> getOrderWithDetails(Long id) {
    return orderRepository.findById(id)
        .flatMap(order ->
            productRepository.findById(order.getProductId())
                .map(product -> OrderResponse.from(order, product)))
        .switchIfEmpty(Mono.error(
            new ResourceNotFoundException("Order", id)))
        .timeout(Duration.ofSeconds(3))
        .onErrorResume(TimeoutException.class,
            e -> Mono.error(new RuntimeException("注文照会がタイムアウトしました")))
        .retry(2);
}

出力:

TEXT 
// 実行成功

4. WebFlux REST API

(1) ▶ サンプル:リアクティブコントローラー

JAVA 
@RestController
@RequestMapping("/api/v1/reactive/orders")
public class ReactiveOrderController {

    private final ReactiveOrderService orderService;

    public ReactiveOrderController(ReactiveOrderService orderService) {
        this.orderService = orderService;
    }

    @GetMapping("/{id}")
    public Mono<OrderResponse> getOrder(@PathVariable Long id) {
        return orderService.findById(id);
    }

    @GetMapping
    public Flux<OrderResponse> listOrders(
            @RequestParam(defaultValue = "0") int page,
            @RequestParam(defaultValue = "20") int size) {
        return orderService.findAll(page, size);
    }

    @PostMapping
    public Mono<ResponseEntity<OrderResponse>> createOrder(
            @Valid @RequestBody CreateOrderRequest request) {
        return orderService.create(request)
            .map(order -> ResponseEntity
                .status(HttpStatus.CREATED).body(order));
    }

    @DeleteMapping("/{id}")
    public Mono<ResponseEntity<Void>> cancelOrder(@PathVariable Long id) {
        return orderService.cancel(id)
            .then(Mono.just(ResponseEntity.noContent().<Void>build()));
    }
}

出力:

TEXT 
// 実行成功
次元 WebMVC WebFlux
戻り値の型 同期オブジェクト Mono<T> / Flux<T>
スレッドモデル リクエストごとに1スレッド イベントループ (少数スレッド)
I/Oモデル ブロッキング ノンブロッキング
コンテナ Tomcat / Jetty Netty
並列制限 スレッドプールサイズ ほぼ無制限

5. WebFluxとWebMVCの選択

(1) 技術選定の決定

シナリオ 推奨 理由
CRUD API, 同時接続 < 500 WebMVC シンプルで直感的, 成熟したエコシステム
I/O集約型, 高並列 WebFlux ノンブロッキング, 少数スレッドで高並列対応
リアルタイムストリーミング (SSE/WebSocket) WebFlux ネイティブサポート
大量のブロッキング操作 (JDBC) WebMVC ブロッキング操作はWebFluxでは逆に性能が悪化
混合シナリオ WebMVC + 部分的な非同期 段階的最適化
⚠️ 注意: WebFluxは万能ではありません。コードでブロッキングI/O (JDBCなど)を多用している場合, WebFluxはWebMVCよりも性能が悪化します。WebFluxを選択する場合, 呼び出しチェーン全体がノンブロッキングである必要があります。

(1) ▶ サンプル:Server-Sent Events (SSE)

JAVA 
@GetMapping(value = "/stream", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
public Flux<OrderEvent> streamOrderEvents() {
    return orderEventPublisher.eventStream()
        .log("order-events");
}

出力:

TEXT 
// 実行成功

6. Spring Data R2DBC

(1) リアクティブデータベースアクセス

(1) ▶ サンプル:R2DBCエンティティとリポジトリ

XML 
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-data-r2dbc</artifactId>
</dependency>
<dependency>
    <groupId>io.r2dbc</groupId>
    <artifactId>r2dbc-mysql</artifactId>
</dependency>

出力:

TEXT 
// 実行成功
JAVA 
@Table("products")
public class Product {

    @Id
    private Long id;
    private String name;
    private BigDecimal price;
    private Integer stock;

    public Product() {}

    public Product(String name, BigDecimal price, Integer stock) {
        this.name = name;
        this.price = price;
        this.stock = stock;
    }
    // getters
}

public interface ReactiveProductRepository
        extends ReactiveCrudRepository<Product, Long> {

    Flux<Product> findByNameContaining(String keyword);

    Flux<Product> findByStockLessThan(Integer threshold);
}
次元 JPA + JDBC R2DBC
ドライバーモデル ブロッキング ノンブロッキング
APIスタイル 同期 リアクティブ (Mono/Flux)
リレーションマッピング サポート (@OneToMany) 未サポート
クエリ言語 JPQL ネイティブSQL / メソッド名派生
トランザクション @Transactional @Transactional (リアクティブ)

(2) ▶ サンプル:リアクティブサービス

JAVA 
@Service
public class ReactiveOrderService {

    private final ReactiveOrderRepository orderRepository;
    private final ReactiveProductRepository productRepository;

    public ReactiveOrderService(ReactiveOrderRepository orderRepository,
                                ReactiveProductRepository productRepository) {
        this.orderRepository = orderRepository;
        this.productRepository = productRepository;
    }

    public Mono<OrderResponse> create(CreateOrderRequest request) {
        return productRepository.findById(request.productId())
            .switchIfEmpty(Mono.error(
                new ResourceNotFoundException("Product", request.productId())))
            .flatMap(product -> {
                if (product.getStock() < request.quantity()) {
                    return Mono.error(new InsufficientStockException(
                        product.getId(), product.getStock(), request.quantity()));
                }
                product.deductStock(request.quantity());
                return productRepository.save(product)
                    .then(orderRepository.save(new Order(product.getId(), request.quantity())));
            })
            .map(OrderResponse::from);
    }

    public Flux<OrderResponse> findAll(int page, int size) {
        return orderRepository.findAll()
            .skip(page * size)
            .take(size)
            .map(OrderResponse::from);
    }

    public Mono<Void> cancel(Long orderId) {
        return orderRepository.findById(orderId)
            .flatMap(order -> {
                order.setStatus("CANCELLED");
                return orderRepository.save(order);
            })
            .then();
    }
}

出力:

TEXT 
// 実行成功

7. 総合サンプル:OrderFlowリアクティブ照会API

JAVA 
// ReactiveProductRepository.java
public interface ReactiveProductRepository
        extends ReactiveCrudRepository<Product, Long> {
    Flux<Product> findByNameContaining(String keyword);
}

// ReactiveOrderRepository.java
public interface ReactiveOrderRepository
        extends ReactiveCrudRepository<Order, Long> {
    Flux<Order> findByStatus(String status);
}

// ReactiveOrderController.java
@RestController
@RequestMapping("/api/v1/reactive/orders")
public class ReactiveOrderController {

    private final ReactiveOrderService orderService;

    public ReactiveOrderController(ReactiveOrderService orderService) {
        this.orderService = orderService;
    }

    @GetMapping("/{id}")
    public Mono<OrderResponse> get(@PathVariable Long id) {
        return orderService.findById(id);
    }

    @GetMapping
    public Flux<OrderResponse> list(
            @RequestParam(defaultValue = "0") int page,
            @RequestParam(defaultValue = "20") int size) {
        return orderService.findAll(page, size);
    }

    @GetMapping("/status/{status}")
    public Flux<OrderResponse> byStatus(@PathVariable String status) {
        return orderService.findByStatus(status);
    }

    @PostMapping
    public Mono<ResponseEntity<OrderResponse>> create(
            @RequestBody CreateOrderRequest req) {
        return orderService.create(req)
            .map(o -> ResponseEntity.status(HttpStatus.CREATED).body(o));
    }

    @DeleteMapping("/{id}")
    public Mono<ResponseEntity<Void>> cancel(@PathVariable Long id) {
        return orderService.cancel(id)
            .thenReturn(ResponseEntity.noContent().<Void>build());
    }
}

// ReactiveOrderService.java
@Service
public class ReactiveOrderService {

    private final ReactiveOrderRepository orderRepo;
    private final ReactiveProductRepository productRepo;

    public ReactiveOrderService(ReactiveOrderRepository orderRepo,
                                ReactiveProductRepository productRepo) {
        this.orderRepo = orderRepo;
        this.productRepo = productRepo;
    }

    public Mono<OrderResponse> findById(Long id) {
        return orderRepo.findById(id)
            .map(OrderResponse::from)
            .switchIfEmpty(Mono.error(new ResourceNotFoundException("Order", id)));
    }

    public Flux<OrderResponse> findAll(int page, int size) {
        return orderRepo.findAll().skip((long) page * size).take(size)
            .map(OrderResponse::from);
    }

    public Flux<OrderResponse> findByStatus(String status) {
        return orderRepo.findByStatus(status).map(OrderResponse::from);
    }

    public Mono<OrderResponse> create(CreateOrderRequest req) {
        return productRepo.findById(req.productId())
            .switchIfEmpty(Mono.error(new ResourceNotFoundException("Product", req.productId())))
            .flatMap(p -> {
                if (p.getStock() < req.quantity()) {
                    return Mono.error(new InsufficientStockException(p.getId(), p.getStock(), req.quantity()));
                }
                p.deductStock(req.quantity());
                return productRepo.save(p)
                    .then(orderRepo.save(new Order(p.getId(), req.quantity())));
            })
            .map(OrderResponse::from);
    }

    public Mono<Void> cancel(Long id) {
        return orderRepo.findById(id)
            .flatMap(o -> { o.setStatus("CANCELLED"); return orderRepo.save(o); })
            .then();
    }
}

❓ よくある質問

Q WebFluxはWebMVCを完全に置き換えられますか?
A お勧めしません。WebMVCのエコシステムはより成熟しており, ほとんどのCRUDプロジェクトではWebMVCの方がシンプルです。WebFluxはI/O集約型, 高並列, リアルタイムストリーミングのシナリオに適しています。両者は共存可能ですが, 混在は避けてください (WebFlux内でブロッキングコードを使用しないでください)。
Q MonoとCompletableFutureの違いは何ですか?
A Monoは遅延評価 (購読時のみ実行)ですが, CompletableFutureは先行評価 (作成直後に実行開始)です。Monoはバックプレッシャーとキャンセルをサポートしますが, CompletableFutureはサポートしません。
Q R2DBCは@OneToManyリレーションマッピングをサポートしていますか?
A いいえ。R2DBCは軽量なリアクティブデータベースドライバーであり, リレーションマッピングを行いません。結合クエリが必要な場合は, flatMapで複数のクエリを手動で組み合わせる必要があります。
Q WebFluxでブロッキングコードを呼び出す方法は?
A Mono.fromCallable() + subscribeOn(Schedulers.boundedElastic())を使用して, ブロッキング操作を弾性スレッドプールにスケジュールします。ただし, 頻繁に使用する場合はWebFluxに適していないことを示しています。
Q WebFluxのエラー処理はどうなりますか?
A 1) オペレーター onErrorResume / onErrorReturn; 2) グローバル @ExceptionHandler (WebFluxもサポート); 3) onErrorMapで例外型を変換。
Q WebFluxのテスト方法は?
A MockMvcの代わりにWebTestClientを使用します。@WebFluxTestはWebFlux層を読み込みます。StepVerifierはMono/Fluxの要素シーケンスをテストするために使用します。

📖 まとめ

📝 練習問題

  1. 基本問題 (難易度:⭐): WebFluxを使って商品照会のリアクティブAPI (GET /api/v1/reactive/products/{id})を実装し, R2DBC経由でH2インメモリデータベースに接続してください。

  2. 応用問題 (難易度:⭐⭐): リアクティブ注文APIを実装し, 在庫確認, 在庫引当, 注文作成の全プロセスをカバーし, flatMapで複数のリアクティブ操作を組み合わせてください。

  3. チャレンジ (難易度:⭐⭐⭐): WebTestClientとStepVerifierを使ってリアクティブAPIテストを書き, SSEリアルタイム注文イベントプッシュインターフェースを実装し, 1,000の同時接続下でのWebMVCとWebFluxのパフォーマンス差を比較してください。

Web-Tutorial.com

Web-Tutorial 技術チーム

複数の開発者によって共同維持されているプログラミングチュートリアルプラットフォーム。各チュートリアルは専門分野の開発者が執筆・レビューしています。正確で信頼性の高いコンテンツを目指しています — 問題を見つけた場合はお知らせください。

100%