spring-webflux

Spring WebFlux

完全版リファレンス: Sinks を使った SSE、StepVerifier によるテスト、Context Propagation のパターンについては、advanced.md を参照してください。

クイックスタート

@RestController
@RequestMapping("/api/users")
@RequiredArgsConstructor
public class UserController {

    private final UserService userService;

    @GetMapping("/{id}")
    public Mono<UserResponse> getUser(@PathVariable Long id) {
        return userService.findById(id);
    }

    @GetMapping
    public Flux<UserResponse> getAllUsers() {
        return userService.findAll();
    }

    @PostMapping
    @ResponseStatus(HttpStatus.CREATED)
    public Mono<UserResponse> createUser(@RequestBody Mono<CreateUserRequest> request) {
        return request.flatMap(userService::create);
    }
}

---

Mono と Flux の基本

Mono（要素数 0 または 1）

Mono<String> empty = Mono.empty();
Mono<String> just = Mono.just("Hello");
Mono<String> fromCallable = Mono.fromCallable(() -> expensiveOperation());
Mono<String> defer = Mono.defer(() -> Mono.just(dynamicValue()));
Mono<String> fromOptional = Mono.justOrEmpty(optionalValue);
Mono<String> fromFuture = Mono.fromFuture(completableFuture);

Flux（要素数 0 から N）

Flux<Integer> just = Flux.just(1, 2, 3);
Flux<Integer> fromIterable = Flux.fromIterable(List.of(1, 2, 3));
Flux<Integer> range = Flux.range(1, 10);
Flux<Long> interval = Flux.interval(Duration.ofSeconds(1));

Flux<Integer> generate = Flux.generate(
    () -> 0,
    (state, sink) -> {
        sink.next(state);
        if (state == 10) sink.complete();
        return state + 1;
    }
);

---

リアクティブオペレーター

変換

// map - transform each element
users.map(user -> new UserResponse(user.getId(), user.getName()));

// flatMap - async transformation (parallel)
users.flatMap(user -> orderRepository.findByUserId(user.getId()));

// flatMapSequential - maintains order
users.flatMapSequential(user -> orderRepository.findByUserId(user.getId()));

// concatMap - sequential, one at a time
users.concatMap(user -> orderRepository.findByUserId(user.getId()));

// switchMap - cancels previous when new arrives
searchTerms.switchMap(term -> searchService.search(term));

フィルタリング

// filter
users.filter(user -> user.getStatus() == Status.ACTIVE);

// filterWhen - async filter
users.filterWhen(user -> permissionService.canAccess(user.getId()));

// distinct / distinctUntilChanged
items.distinct();
values.distinctUntilChanged();

// take / skip
users.skip((long) page * size).take(size);

結合

// zip - combine by position
Flux.zip(users, orders, UserWithOrders::new);

// merge - interleave from multiple sources
Flux.merge(source1, source2);

// concat - sequential
Flux.concat(first, second);

// zipWith on Mono
userRepository.findById(userId)
    .zipWith(profileRepository.findByUserId(userId))
    .map(tuple -> new UserWithProfile(tuple.getT1(), tuple.getT2()));

エラー処理

// onErrorReturn - default value on error
userRepository.findById(id).onErrorReturn(new User("default"));

// onErrorResume - fallback Publisher
primaryRepository.findById(id)
    .onErrorResume(e -> fallbackRepository.findById(id));

// onErrorResume with specific type
userRepository.findById(id)
    .onErrorResume(NotFoundException.class, e -> Mono.empty())
    .onErrorResume(TimeoutException.class, e -> cacheRepository.findById(id));

// onErrorMap - transform exception
userRepository.findById(id)
    .onErrorMap(DataAccessException.class,
        e -> new ServiceException("Database error", e));

// retryWhen - advanced retry
userRepository.findById(id)
    .retryWhen(Retry.backoff(3, Duration.ofSeconds(1))
        .filter(e -> e instanceof TransientException));

副作用

userRepository.findById(id)
    .doOnSubscribe(s -> log.info("Subscribed"))
    .doOnNext(user -> log.info("Found user: {}", user.getId()))
    .doOnError(e -> log.error("Error: {}", e.getMessage()))
    .doFinally(signalType -> log.info("Finally: {}", signalType));

---

WebClient

設定

@Configuration
public class WebClientConfig {

    @Bean
    public WebClient webClient(WebClient.Builder builder) {
        HttpClient httpClient = HttpClient.create()
            .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 5000)
            .responseTimeout(Duration.ofSeconds(30))
            .doOnConnected(conn -> conn
                .addHandlerLast(new ReadTimeoutHandler(30, TimeUnit.SECONDS))
                .addHandlerLast(new WriteTimeoutHandler(10, TimeUnit.SECONDS)));

        return builder
            .baseUrl("https://api.example.com")
            .clientConnector(new ReactorClientHttpConnector(httpClient))
            .defaultHeader(HttpHeaders.CONTENT_TYPE, MediaType.APPLICATION_JSON_VALUE)
            .build();
    }
}

使用方法

@Service
@RequiredArgsConstructor
public class ExternalApiService {

    private final WebClient webClient;

    public Mono<UserDto> getUser(Long id) {
        return webClient.get()
            .uri("/users/{id}", id)
            .retrieve()
            .bodyToMono(UserDto.class);
    }

    public Mono<UserDto> getUserSafe(Long id) {
        return webClient.get()
            .uri("/users/{id}", id)
            .retrieve()
            .onStatus(HttpStatusCode::is4xxClientError, response ->
                response.bodyToMono(ErrorResponse.class)
                    .flatMap(error -> Mono.error(new ClientException(error.getMessage()))))
            .onStatus(HttpStatusCode::is5xxServerError, response ->
                Mono.error(new ServerException("Server error")))
            .bodyToMono(UserDto.class);
    }

    // Parallel calls
    public Mono<AggregatedData> getAggregatedData(Long userId) {
        return Mono.zip(
            getUser(userId),
            getOrders(userId).collectList(),
            getProfile(userId)
        ).map(tuple -> new AggregatedData(tuple.getT1(), tuple.getT2(), tuple.getT3()));
    }
}

---

関数型エンドポイント

@Configuration
public class RouterConfig {

    @Bean
    public RouterFunction<ServerResponse> userRoutes(UserHandler handler) {
        return RouterFunctions.route()
            .path("/api/users", builder -> builder
                .GET("", handler::getAll)
                .GET("/{id}", handler::getById)
                .POST("", handler::create)
                .PUT("/{id}", handler::update)
                .DELETE("/{id}", handler::delete)
            )
            .build();
    }
}

@Component
@RequiredArgsConstructor
public class UserHandler {

    private final UserService userService;

    public Mono<ServerResponse> getById(ServerRequest request) {
        Long id = Long.parseLong(request.pathVariable("id"));
        return userService.findById(id)
            .flatMap(user -> ServerResponse.ok()
                .contentType(MediaType.APPLICATION_JSON)
                .bodyValue(user))
            .switchIfEmpty(ServerResponse.notFound().build());
    }

    public Mono<ServerResponse> create(ServerRequest request) {
        return request.bodyToMono(CreateUserRequest.class)
            .flatMap(userService::create)
            .flatMap(user -> ServerResponse.created(
                    URI.create("/api/users/" + user.getId()))
                .bodyValue(user));
    }
}

---

ベストプラクティス

推奨事項	非推奨事項
チェーン全体をリアクティブに保つ	ハンドラー内で .block() を使う
適切なオペレーターを使う（flatMap と concatMap の使い分け）	ブロッキング処理とリアクティブ処理を混在させる
onError* オペレーターでエラーを処理する	エラーを無視する
テストに StepVerifier を使う	.block() でテストする
MDC／セキュリティのために Context を伝播させる	ThreadLocal を使う

本番環境向けチェックリスト

• WebClient にタイムアウトを設定
• エラー処理を完備
• 一時的なエラーに対する再試行ロジック
• バックプレッシャー戦略を定義
• ロギングのための Context 伝播
• リアクティブメトリクス
• StepVerifier でテスト

このスキルを使用しない場合

• 従来型のブロッキングアプリ - spring-web スキルを使用
• シンプルな CRUD API - spring-rest スキルを使用
• CPU バウンドのワークロード - ここではリアクティブ処理は役に立ちません
• リアクティブ処理に不慣れなチーム - 学習の難易度が高いです

アンチパターン

アンチパターン	問題	解決策
何も実行されない	Publisher がサブスクライブされていない	subscribe を確実に行うか、コントローラーから返す
ブロッキング呼び出し	リアクティブチェーン内の .block()	block を避け、オペレーターを使う
Context が失われる	MDC が伝播されない	Context 伝播を使う
メモリリーク	バックプレッシャーのない無限 Flux	バックプレッシャーオペレーターを使う
Cold と Hot の混同	サブスクライブのたびに Publisher が再作成される	.share() または .cache() を使う

トラブルシューティングのクイックガイド

問題	診断	修正方法
Mono／Flux が完了しない	subscribe の不足を確認	コントローラーから返す
Context を利用できない	伝播を確認	Hooks.enableAutomaticContextPropagation() を使う
バックプレッシャーのオーバーフロー	バッファーサイズを確認	onBackpressure* オペレーターを使う
テストがタイムアウトする	StepVerifier の使い方を確認	遅延には仮想時間を使う
メモリ使用量が増え続ける	リークを確認	.limitRate() または .take() を使う

リファレンスドキュメント

• Spring WebFlux
• Project Reactor

関連スキル

• spring-r2dbc - リアクティブなデータベースアクセス用
• spring-web - MVC との比較用
• spring-websocket - リアクティブな WebSocket 用