Operator hay dùng: map, flatMap, zip...

Bài trước ta tạo được Mono/Flux. Nhưng checkout có nhiều bước: lấy user, kiểm tra, gọi payment, lưu đơn. Không được block() để lấy giá trị ra, nên ta nối các bước bằng operator.

1. Bài toán: checkout nhiều bước

checkout(userId, cartId)
  1. lấy User từ user-service        (Mono<User>)
  2. lấy Cart                        (Mono<Cart>)   -- độc lập với bước 1
  3. user không tồn tại -> 404
  4. gọi payment-service với tổng tiền (Mono<Payment>)
  5. chuyển kết quả thành CheckoutResult

2. Vấn đề

// SAI: block trên event loop
User user = userClient.findById(id).block();
Payment p = paymentClient.pay(user, total).block();

Code chạy kiểu imperative quen tay sẽ block event loop (Reactor còn ném IllegalStateException: block() is not supported trên thread reactor-http-nio). Cần cách nối bước mà không lấy giá trị ra.

3. Giải pháp: operator

4. Code từng bước

Bước 1: Dependency: spring-boot-starter-webflux.

Bước 2: Config WebClient bean:

@Configuration
public class ClientConfig {
    @Bean
    WebClient userWebClient(WebClient.Builder b) {
        return b.baseUrl("http://user-service").build();
    }
    @Bean
    WebClient paymentWebClient(WebClient.Builder b) {
        return b.baseUrl("http://payment-service").build();
    }
}

Bước 3: Code Java

@Service
public class CheckoutService {

    private final UserClient userClient;
    private final CartClient cartClient;
    private final PaymentClient paymentClient;

    // constructor injection...

    public Mono<CheckoutResult> checkout(Long userId, Long cartId) {
        Mono<User> user = userClient.findById(userId)
                .switchIfEmpty(Mono.error(
                    new ResponseStatusException(HttpStatus.NOT_FOUND, "user not found")));

        Mono<Cart> cart = cartClient.findById(cartId);

        return Mono.zip(user, cart)                                  // song song
                .filter(t -> t.getT1().active())                     // user bị khóa -> rỗng
                .switchIfEmpty(Mono.error(
                    new ResponseStatusException(HttpStatus.FORBIDDEN, "user locked")))
                .flatMap(t -> paymentClient.pay(t.getT1().id(), t.getT2().total())) // async
                .map(p -> new CheckoutResult(p.orderId(), p.status()));            // sync
    }
}

// Một client mẫu
public Mono<User> findById(Long id) {
    return userWebClient.get().uri("/users/{id}", id)
            .retrieve()
            .bodyToMono(User.class);   // 404 sẽ là lỗi; tùy API có thể map sang Mono.empty()
}

// Danh sách gợi ý: rỗng thì trả mặc định
public Mono<String> shippingTier(Long userId) {
    return userWebClient.get().uri("/users/{id}/tier", userId)
            .retrieve().bodyToMono(String.class)
            .defaultIfEmpty("STANDARD");
}

Bước 4: Test

curl -X POST localhost:8080/api/checkout -H "Content-Type: application/json" \
  -d '{"userId":1,"cartId":10}'
# user 999 không có:
curl -X POST localhost:8080/api/checkout -H "Content-Type: application/json" \
  -d '{"userId":999,"cartId":10}'   # 404

5. Cách hoạt động

zip:      user  ──────●(300ms)
          cart  ───●(150ms)
          zip   ──────●  (đợi cả hai, tổng ~300ms chứ không phải 450ms)
                      |
flatMap:              └─ pay(...) ─────────●(2s)
map:                                       └─ CheckoutResult

Flux.just(1,2,3).flatMap(id -> callApi(id))    -> 2, 3, 1  (song song, thứ tự tùy ai xong trước)
Flux.just(1,2,3).concatMap(id -> callApi(id))  -> 1, 2, 3  (từng cái một)
Flux.just(1,2,3).flatMapSequential(...)        -> 1, 2, 3  (song song, sắp lại thứ tự)

6. So sánh

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

• map cho việc nhẹ, đồng bộ. Không gọi I/O trong map.
• flatMap khi gọi API/DB. Với Flux lớn, giới hạn song song: flatMap(fn, 8) để không dồn payment-service.
• concatMap khi thứ tự quan trọng (trừ tiền từng giao dịch theo thứ tự).
• zip khi các bước độc lập. Một nguồn rỗng thì cả zip rỗng.

Bạn hiểu gần đúng, chỉ cần chỉnh một điểm: điều quan trọng là hàm trả về gì.

Mono<User> user = ...;

user.map(u -> paymentClient.pay(u))      // Mono<Mono<Payment>>  -- lồng nhau!
user.flatMap(u -> paymentClient.pay(u))  // Mono<Payment>        -- "làm phẳng" và subscribe vào Mono bên trong

Với map, Mono bên trong không bao giờ được subscribe, nên payment không được gọi (nhớ bài trước: không subscribe thì không chạy). flatMap subscribe vào nó và đợi kết quả. Giống thenApply vs thenCompose của CompletableFuture, hay Optional.map vs flatMap.

Đúng, đó là bẫy kinh điển. defaultIfEmpty(value) dùng khi có sẵn một giá trị. switchIfEmpty(publisher) dùng khi phải chạy một Publisher khác. Nhưng tham số của Java được tính trước khi gọi method, nên createGuestUser() chạy (phần code eager trong nó) dù không cần:

// Bẫy: createGuestUser() luôn được gọi để tạo tham số
.switchIfEmpty(createGuestUser())

// Đúng: trì hoãn tới khi thật sự rỗng
.switchIfEmpty(Mono.defer(() -> createGuestUser()))

Nếu createGuestUser() chỉ trả về Mono lazy (không làm gì trước khi subscribe) thì không sao, nhưng bọc Mono.defer là thói quen an toàn.

Q1. map và flatMap khác nhau thế nào?

🇻🇳 Tiếng Việt:

map biến đổi đồng bộ, hàm trả về giá trị thường. flatMap dùng khi hàm trả về Publisher, tức một bước async như gọi API hay DB; nó subscribe vào Publisher bên trong và làm phẳng kết quả. Dùng map cho bước async sẽ ra Mono lồng Mono và bước đó không chạy.

🇬🇧 English:

map is a synchronous transformation; the function returns a plain value. flatMap is used when the function returns a Publisher, an async step like an API or database call; it subscribes to the inner publisher and flattens the result. Using map for an async step gives a nested Mono and the call never runs.

Q2. flatMap và concatMap trên Flux khác nhau thế nào?

🇻🇳 Tiếng Việt:

flatMap subscribe nhiều Publisher bên trong cùng lúc nên nhanh nhưng không giữ thứ tự; có thể giới hạn concurrency bằng tham số. concatMap chạy từng cái một, giữ thứ tự nhưng chậm hơn. flatMapSequential chạy song song và trả theo thứ tự gốc.

🇬🇧 English:

flatMap subscribes to several inner publishers at once, so it is fast but does not preserve order; concurrency can be limited with a parameter. concatMap processes one at a time, preserving order but slower. flatMapSequential runs concurrently and emits in the original order.

Q3. Gọi hai API độc lập rồi gộp kết quả thế nào?

🇻🇳 Tiếng Việt:

Dùng Mono.zip để subscribe cả hai song song và nhận Tuple khi cả hai xong; tổng thời gian bằng cái chậm nhất. Lưu ý nếu một nguồn rỗng thì zip rỗng, nếu một nguồn lỗi thì zip lỗi.

🇬🇧 English:

Use Mono.zip to subscribe to both in parallel and get a Tuple when both complete; total time equals the slowest call. If one source is empty the zip is empty, and if one errors the zip errors.

Cách nhớ ngắn gọn

• Trả giá trị → map. Trả Mono/Flux → flatMap.
• Cần thứ tự → concatMap. Độc lập → zip.
• Rỗng: giá trị sẵn → defaultIfEmpty; Publisher khác → switchIfEmpty(Mono.defer(...)).
• Không block() để lấy giá trị giữa chừng.