WebFlux & WebClient: checkout gọi 3 service song song

OK. Tới đây bạn đã có đủ "nguyên liệu": Mono/Flux, operator, xử lý lỗi, Schedulers. Bài này lắp vào Spring WebFlux (phía server) và WebClient (phía client).

1. Bài toán: checkout trong shop-app

Ở chương Concurrency, checkout gọi 3 API, mỗi cái ~1s:

GET /checkout/{userId}
  |-- user-service     : GET /users/{id}            1s
  |-- order-service    : GET /carts/{userId}        1s
  |-- payment-service  : GET /payment-methods/{id}  1s

Bản MVC tuần tự mất 3s. Bản CompletableFuture còn 1s, nhưng vẫn giữ 3 thread pool đứng chờ I/O.

2. Vấn đề

• RestTemplate / RestClient là blocking: mỗi lời gọi giữ một thread đến khi có response.
• Gọi chúng trong WebFlux là block event loop (bài 4.5.7).
• Lúc tải cao, thread pool cạn trước khi CPU bận.

3. Giải pháp: WebFlux + WebClient

• WebFlux: web framework chạy trên Reactor Netty, handler trả về Mono/Flux. Framework subscribe và ghi response khi có dữ liệu.
• WebClient: HTTP client non-blocking, trả về Mono/Flux. Gửi request xong, thread được trả lại ngay.
• Mono.zip: subscribe nhiều Mono cùng lúc, đợi tất cả xong rồi gộp kết quả.

WebFlux có hai kiểu viết endpoint:

4. Code từng bước

Bước 1: dependency (chỉ webflux, không thêm starter-web)

<dependency>
  <groupId>org.springframework.boot</groupId>
  <artifactId>spring-boot-starter-webflux</artifactId>
</dependency>

Bước 2: cấu hình WebClient có timeout

@Configuration
public class WebClientConfig {
    @Bean
    WebClient userWebClient(WebClient.Builder builder,
                            @Value("${clients.user.base-url}") String baseUrl) {
        HttpClient http = HttpClient.create()
            .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 1_000) // connect timeout
            .responseTimeout(Duration.ofSeconds(2));             // chờ response
        return builder.baseUrl(baseUrl)
            .clientConnector(new ReactorClientHttpConnector(http))
            .build();
    }
    // tương tự cho orderWebClient, paymentWebClient
}

# application.yml
clients:
  user.base-url: http://user-service:8081
  order.base-url: http://order-service:8082
  payment.base-url: http://payment-service:8083

Bước 3: client dùng retrieve

@Component
public class UserClient {
    private final WebClient webClient;
    public UserClient(@Qualifier("userWebClient") WebClient webClient) { this.webClient = webClient; }

    public Mono<User> get(Long id) {
        return webClient.get().uri("/users/{id}", id)
            .retrieve()
            .onStatus(s -> s.value() == 404,
                      r -> Mono.error(new UserNotFoundException(id)))
            .bodyToMono(User.class)
            .timeout(Duration.ofSeconds(2));     // timeout cho cả pipeline
    }
}

Bước 4: service gọi 3 API song song bằng Mono.zip

@Service
@RequiredArgsConstructor
public class CheckoutService {
    private final UserClient userClient;
    private final OrderClient orderClient;
    private final PaymentClient paymentClient;

    public Mono<CheckoutView> view(Long userId) {
        return Mono.zip(
                userClient.get(userId),
                orderClient.getCart(userId),
                paymentClient.methods(userId)
                    .onErrorReturn(PaymentInfo.empty()))   // payment lỗi vẫn hiện trang
            .map(t -> new CheckoutView(t.getT1(), t.getT2(), t.getT3()));
    }
}

Bước 5: controller, hai kiểu

@RestController
@RequiredArgsConstructor
@RequestMapping("/checkout")
public class CheckoutController {
    private final CheckoutService checkoutService;

    @GetMapping("/{userId}")
    public Mono<CheckoutView> view(@PathVariable Long userId) {
        return checkoutService.view(userId);   // KHÔNG gọi .block()
    }
}

@Configuration
public class CheckoutRoutes {
    @Bean
    RouterFunction<ServerResponse> routes(CheckoutService service) {
        return RouterFunctions.route()
            .GET("/v2/checkout/{userId}", req ->
                 service.view(Long.valueOf(req.pathVariable("userId")))
                        .flatMap(v -> ServerResponse.ok().bodyValue(v)))
            .build();
    }
}

Bước 6: SSE, đẩy trạng thái đơn hàng

@GetMapping(value = "/orders/{id}/status", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
public Flux<OrderStatus> status(@PathVariable Long id) {
    return orderClient.statusStream(id);   // Flux: PENDING, PAID, SHIPPED ...
}

Bước 7: test bằng curl

curl localhost:8080/checkout/42
# ~1s thay vì 3s

curl -N localhost:8080/checkout/orders/7/status
# data:{"status":"PENDING"}
# data:{"status":"PAID"}

5. Cách hoạt động

reactor-http-nio-1
  | nhận GET /checkout/42
  | Mono.zip subscribe 3 nguồn:
  |   gửi GET /users/42      ─┐
  |   gửi GET /carts/42      ─┼─ đăng ký với NIO selector, KHÔNG chờ
  |   gửi GET /payment/42    ─┘
  | (rảnh, xử lý request khác)
  ...
  | response users  đến → lưu T1
  | response carts  đến → lưu T2
  | response payment đến → lưu T3 → đủ 3 → map → ghi JSON
Tổng thời gian ≈ max(1s,1s,1s) = 1s, không thread nào đứng chờ.

6. So sánh

7. Khi nào nên dùng / không nên dùng

• Nên: gateway, BFF, service gọi nhiều API khác, streaming (SSE), số connection đồng thời rất lớn.
• Không nên: app CRUD dùng JPA là chính, team chưa quen reactive. Khi đó MVC + RestClient + virtual threads đơn giản hơn (bài 4.5.10).
• Tuyệt đối không .block() trong controller WebFlux: sẽ ném IllegalStateException trên thread Netty.

Đúng ý bài 4.5.2. WebClient chính là "thư viện call HTTP hỗ trợ async" đó trong Spring. Nối lại với code ở trên:

// ❌ RestTemplate trong WebFlux: 100 request = event loop bị chiếm
User u = restTemplate.getForObject("/users/{id}", User.class, id);

// ✅ WebClient: gửi xong trả thread ngay, response về thì callback chạy tiếp
Mono<User> u = webClient.get().uri("/users/{id}", id).retrieve().bodyToMono(User.class);

Chỉ cần thêm một điểm: "không bị chiếm" chỉ đúng khi cả chuỗi không có chỗ nào blocking. Một .block(), một lời gọi JDBC, hay một Thread.sleep trong map là đủ làm hỏng lợi ích đó. Và vẫn phải có timeout: không tốn thread, nhưng 100 request treo mãi vẫn giữ connection và bộ nhớ.

Đúng, mặc định zip fail-fast: một nguồn báo lỗi thì zip báo lỗi ngay và cancel các nguồn còn lại. Đó là lý do ở Bước 4 mình gắn onErrorReturn(PaymentInfo.empty()) riêng cho payment: dữ liệu "có cũng được" thì cho fallback, dữ liệu bắt buộc (user, cart) thì để lỗi lan ra.

user lỗi     → 503, không có user thì không checkout
payment lỗi  → vẫn trả trang, ẩn phần "phương thức đã lưu"

Ngoài ra có Mono.zipDelayError(...) nếu muốn đợi hết rồi gộp tất cả lỗi. Và zip với một Mono rỗng thì kết quả cũng rỗng, nhớ switchIfEmpty khi cần.

Q1. Spring WebFlux là gì, khác Spring MVC thế nào?

🇻🇳 Tiếng Việt:

WebFlux là web framework reactive, mặc định chạy trên Reactor Netty với vài event loop thread, handler trả Mono hoặc Flux. MVC chạy trên Servlet, mỗi request giữ một thread đến khi xong. WebFlux có hai kiểu endpoint: annotated controller giống MVC và functional routes với RouterFunction. Nó chỉ có lợi khi toàn bộ chuỗi I/O là non-blocking.

🇬🇧 English:

WebFlux is Spring's reactive web framework. By default it runs on Reactor Netty with a few event-loop threads, and handlers return Mono or Flux. Spring MVC runs on the Servlet API and holds one thread per request until it finishes. WebFlux supports annotated controllers like MVC and functional routes with RouterFunction. It only pays off when the whole I/O chain is non-blocking.

Q2. Gọi 3 service song song trong WebFlux thế nào?

🇻🇳 Tiếng Việt:

Dùng WebClient cho từng service, mỗi lời gọi trả Mono có timeout, rồi Mono.zip để subscribe cả ba cùng lúc và gộp kết quả. Tổng thời gian bằng lời gọi chậm nhất. zip là fail-fast, nên với dữ liệu không bắt buộc em thêm onErrorReturn hoặc onErrorResume để có fallback.

🇬🇧 English:

I use WebClient for each service, each call returns a Mono with a timeout, and then Mono.zip subscribes to all three at once and combines the results. The total time equals the slowest call. zip fails fast, so for optional data I add onErrorReturn or onErrorResume as a fallback.

Q3. retrieve() và exchangeToMono() khác nhau ra sao? Cấu hình timeout ở đâu?

🇻🇳 Tiếng Việt:

retrieve đơn giản, lấy body và tự biến 4xx/5xx thành exception, chỉnh bằng onStatus. exchangeToMono cho toàn quyền với status và header, thay cho exchange cũ vốn dễ rò connection. Timeout em đặt hai lớp: connect timeout và responseTimeout trên HttpClient của Reactor Netty, cộng operator timeout trên Mono cho cả pipeline.

🇬🇧 English:

retrieve is the simple path: it extracts the body and turns 4xx and 5xx into exceptions, which I can customize with onStatus. exchangeToMono gives full control over status and headers, and it replaces the old exchange method that could leak connections. For timeouts I use two layers: connect timeout and responseTimeout on the Reactor Netty HttpClient, plus the timeout operator on the Mono for the whole pipeline.

Q4. SSE trong WebFlux làm thế nào?

🇻🇳 Tiếng Việt:

Trả về Flux và đặt produces là text/event-stream. Mỗi phần tử Flux thành một event gửi xuống client trên cùng một connection, hợp với việc theo dõi trạng thái đơn hàng hoặc thông báo realtime một chiều.

🇬🇧 English:

Return a Flux and set produces to text/event-stream. Each element of the Flux becomes an event pushed to the client over the same connection, which suits order status tracking or one-way realtime notifications.

Cách nhớ ngắn gọn

• WebFlux = Netty + handler trả Mono/Flux; không bao giờ .block().
• WebClient = HTTP non-blocking; timeout ở HttpClient + .timeout().
• Song song: Mono.zip, nhanh bằng service chậm nhất, fail-fast.
• retrieve() mặc định; exchangeToMono() khi cần toàn quyền.
• SSE = Flux + text/event-stream.