Publisher, Subscriber, Mono & Flux

Giờ có đủ nền (event loop, non-blocking) để học kiểu dữ liệu reactive. Mục tiêu: hiểu Publisher/Subscriber, rồi Mono/Flux, và vì sao không subscribe thì không có gì chạy.

1. Bài toán: "khi có kết quả payment thì làm tiếp"

Trong checkout, ta muốn nói: "gọi payment; khi có kết quả thì lưu đơn; nếu lỗi thì báo lỗi". Thử bằng Java Core với callback:

public interface PaymentCallback {
    void onSuccess(String result);
    void onError(Exception error);
}

public class PaymentClient {
    private final ExecutorService executor = Executors.newFixedThreadPool(2);

    public void pay(PaymentCallback cb) {
        executor.submit(() -> {
            try {
                Thread.sleep(2000);          // giả lập gọi ngân hàng
                cb.onSuccess("SUCCESS");
            } catch (Exception e) {
                cb.onError(e);
            }
        });
    }
}

START
MAIN CONTINUES
... 2s ...
Payment: SUCCESS

2. Vấn đề của callback tự chế

• Mỗi thư viện tự định nghĩa callback riêng, không chuẩn chung.
• Không có tín hiệu "hết dữ liệu" khi trả về nhiều phần tử.
• Không có cách hủy (cancel), không có cách nói "chỉ gửi tôi 10 cái thôi" (backpressure).
• Lồng nhiều bước thì thành callback hell.

3. Giải pháp: Reactive Streams + Project Reactor

Reactive Streams là chuẩn gồm 4 interface (từ Java 9 cũng có trong java.util.concurrent.Flow):

Publisher<T>     void subscribe(Subscriber<? super T> s)
Subscriber<T>    onSubscribe(Subscription s)
                 onNext(T item)        // 0..N lần
                 onError(Throwable t)  // kết thúc, có lỗi
                 onComplete()          // kết thúc, thành công
Subscription     request(long n)       // xin thêm n phần tử
                 cancel()
Processor<T,R>   vừa Subscriber vừa Publisher

Quy tắc: onNext* rồi tối đa một trong onError / onComplete. Sau tín hiệu kết thúc thì không phát gì nữa.

Project Reactor cài đặt chuẩn này với hai Publisher:

4. Code từng bước

Bước 1: Dependency: spring-boot-starter-webflux (đã có reactor-core). Muốn test thì thêm:

<dependency>
    <groupId>io.projectreactor</groupId>
    <artifactId>reactor-test</artifactId>
    <scope>test</scope>
</dependency>

Bước 2: Tạo Mono / Flux

Mono<String> one   = Mono.just("SUCCESS");
Mono<String> none  = Mono.empty();
Mono<String> err   = Mono.error(new IllegalStateException("bank down"));
Mono<String> lazy  = Mono.fromCallable(() -> "computed");   // chỉ chạy khi subscribe

Flux<Integer> nums = Flux.just(1, 2, 3);
Flux<Order> orders = Flux.fromIterable(orderList);
Flux<Long> ticks   = Flux.interval(Duration.ofSeconds(1)); // vô hạn

Bước 3: Subscribe và thấy tính lazy

Mono<String> payment = Mono.fromCallable(() -> {
    System.out.println("Calling bank...");
    return "SUCCESS";
});

System.out.println("Mono created");       // chưa gọi bank!
payment.subscribe(
    r   -> System.out.println("Success: " + r),   // onNext
    e   -> System.out.println("Error: " + e),     // onError
    ()  -> System.out.println("Completed"));      // onComplete

Mono created
Calling bank...
Success: SUCCESS
Completed

Trong WebFlux, bạn không tự gọi subscribe(). Controller trả về Mono, framework subscribe khi gửi response:

@GetMapping("/api/users/{id}")
public Mono<User> getUser(@PathVariable Long id) {
    return userClient.findById(id);   // WebFlux sẽ subscribe
}

@GetMapping(value = "/api/orders", produces = MediaType.APPLICATION_NDJSON_VALUE)
public Flux<Order> orders() {
    return orderRepository.findAll(); // R2DBC, stream từng dòng
}

Bước 4: Test

@Test
void paymentEmitsSuccess() {
    StepVerifier.create(Mono.just("SUCCESS"))
        .expectNext("SUCCESS")
        .verifyComplete();
}

curl localhost:8080/api/users/1
curl -H "Accept: application/x-ndjson" localhost:8080/api/orders

5. Cách hoạt động

Subscriber                 Publisher (Flux)
    | ---- subscribe() ----> |
    | <--- onSubscribe(s) -- |
    | ---- s.request(3) ---> |
    | <--- onNext(order1) -- |
    | <--- onNext(order2) -- |
    | <--- onNext(order3) -- |
    | <--- onComplete() ---- |

Lỗi giữa chừng:
    | <--- onNext(order1) -- |
    | <--- onError(ex) ----- |   X stream kết thúc

6. So sánh

Cold vs hot (ngắn gọn): cold publisher (mặc định, ví dụ webClient...bodyToMono) chạy lại từ đầu cho mỗi subscriber: subscribe 2 lần = gọi HTTP 2 lần. Hot publisher phát dữ liệu bất kể ai nghe, subscriber đến muộn bỏ lỡ phần trước (ví dụ Sinks.many().multicast(), hoặc flux.share()). Muốn cache kết quả cold thì dùng .cache().

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

• Mono: kết quả đơn (findById, gọi một API). Flux: danh sách lớn, stream, SSE.
• Không trả Mono<List<T>> khi danh sách rất lớn; dùng Flux<T> để stream.
• Không gọi .block() trong code WebFlux để "lấy giá trị ra". Chỉ dùng trong test hoặc app non-reactive.

Đúng rồi, thấy giống là có lý do: cùng tư duy phản ứng khi dữ liệu xuất hiện. Nhưng chỉnh một điểm: Reactive Programming không phải ReactJS. Cái giống thật ra là Promise của JavaScript:

// JavaScript Promise
fetch("/api/payment")
  .then(res => res.json())
  .then(data => console.log("Success", data))
  .catch(err => console.log("Error", err));

// Java CompletableFuture
CompletableFuture.supplyAsync(() -> callBank())
  .thenAccept(r -> System.out.println("Success " + r))
  .exceptionally(e -> { System.out.println("Error " + e); return null; });

// Reactor Mono
payment.subscribe(
  r -> System.out.println("Success " + r),
  e -> System.out.println("Error " + e),
  () -> System.out.println("Completed"));

Gần nhất với Reactor bên JS là RxJS (Angular dùng nhiều), cùng họ ReactiveX.

Bạn đoán đúng hướng: không có gì xảy ra. pay() chỉ trả về một "bản kế hoạch". Không ai subscribe thì HTTP không được gửi.

// SAI: payment không bao giờ được gọi
public Mono<Order> checkout(Order order) {
    paymentClient.pay(order);                 // Mono bị bỏ rơi
    return orderRepository.save(order);
}

// ĐÚNG: nối vào chuỗi
public Mono<Order> checkout(Order order) {
    return paymentClient.pay(order)
            .then(orderRepository.save(order));
}

Ngược lại, gọi .subscribe() tay bên trong controller cũng là bẫy: nó chạy "fire-and-forget", lỗi không về tới client, mất context. Quy tắc: trả Mono/Flux lên trên, để framework subscribe.

Q1. Reactive Streams gồm những gì?

🇻🇳 Tiếng Việt:

Là chuẩn xử lý luồng dữ liệu bất đồng bộ có backpressure, gồm Publisher, Subscriber, Subscription và Processor. Subscriber nhận onSubscribe, rồi onNext nhiều lần, kết thúc bằng onError hoặc onComplete. Qua Subscription nó gọi request(n) để xin dữ liệu hoặc cancel.

🇬🇧 English:

It is a standard for asynchronous stream processing with backpressure, made of Publisher, Subscriber, Subscription and Processor. A Subscriber gets onSubscribe, then onNext zero or more times, and finishes with either onError or onComplete. Through the Subscription it calls request(n) to ask for data, or cancel.

Q2. Mono và Flux khác nhau thế nào?

🇻🇳 Tiếng Việt:

Cả hai là Publisher của Project Reactor. Mono phát 0 hoặc 1 phần tử, dùng cho kết quả đơn như findById. Flux phát 0 đến N phần tử, dùng cho danh sách hoặc stream. Khác CompletableFuture, chúng lazy: chỉ chạy khi có subscribe.

🇬🇧 English:

Both are Project Reactor publishers. Mono emits zero or one item, for single results like findById. Flux emits zero to N items, for lists or streams. Unlike CompletableFuture they are lazy: nothing runs until someone subscribes.

Q3. Cold và hot publisher?

🇻🇳 Tiếng Việt:

Cold publisher bắt đầu lại từ đầu cho mỗi subscriber, ví dụ một lời gọi WebClient: subscribe hai lần là gọi HTTP hai lần. Hot publisher phát bất kể có ai nghe, subscriber đến muộn bỏ lỡ dữ liệu cũ, ví dụ Sinks hoặc share().

🇬🇧 English:

A cold publisher starts over for each subscriber, for example a WebClient call: subscribing twice sends two HTTP requests. A hot publisher emits regardless of subscribers, so late subscribers miss earlier data, for example Sinks or share().

Cách nhớ ngắn gọn

• onNext* rồi (onError | onComplete).
• Mono = 0..1, Flux = 0..N.
• Nothing happens until you subscribe.
• Trả Mono/Flux lên framework, không block(), không subscribe() tay.