Reactive Streams

1. Vì sao cần Reactive Streams?

Async/non-blocking giải quyết chuyện “đừng giữ thread khi đang chờ”, nhưng chưa giải quyết chuyện producer phát dữ liệu nhanh hơn consumer xử lý.

Producer 100k msg/s ─────► Consumer 10k msg/s
                 backlog ↑↑↑
                 memory  ↑↑↑

Reactive Streams định nghĩa protocol bất đồng bộ có non-blocking backpressure để consumer biểu đạt demand.

2. Bốn contract cốt lõi

3. Protocol

Subscriber ── subscribe() ──► Publisher
Subscriber ◄─ onSubscribe(s) ─ Publisher
Subscriber ── request(3) ─────► Subscription
Subscriber ◄─ onNext(A)
Subscriber ◄─ onNext(B)
Subscriber ◄─ onNext(C)
Subscriber ── request(2) ─────► Subscription
...
Subscriber ◄─ onComplete() / onError()

Publisher không được tùy ý đẩy vô hạn item nếu demand chưa cho phép. Signal phải tuần tự theo protocol.

4. Demand và Backpressure

request(n) là “tín dụng” consumer cấp cho producer. Demand có thể tăng dần; producer chỉ emit trong phạm vi demand chưa tiêu thụ.

• Buffer: giữ item tạm thời.
• Drop: bỏ item khi consumer chậm.
• Latest: giữ item mới nhất.
• Error: fail khi không thể đáp ứng.

5. Mini Publisher

final class RangePublisher implements Publisher<Integer> {
    private final int start;
    private final int count;

    RangePublisher(int start, int count) {
        this.start = start;
        this.count = count;
    }

    public void subscribe(Subscriber<? super Integer> s) {
        s.onSubscribe(new Subscription() {
            int current = start;
            int emitted = 0;
            boolean cancelled;

            public void request(long n) {
                if (n <= 0) {
                    s.onError(new IllegalArgumentException("n must be > 0"));
                    return;
                }
                long remaining = n;
                while (!cancelled && remaining-- > 0 && emitted < count) {
                    s.onNext(current++);
                    emitted++;
                }
                if (!cancelled && emitted == count) s.onComplete();
            }

            public void cancel() { cancelled = true; }
        });
    }
}

Đây là implementation minh họa; spec thực tế còn nhiều rule về serialization, reentrancy và concurrency.

6. Mini Subscriber

final class BatchSubscriber implements Subscriber<Integer> {
    private Subscription subscription;
    private int received;

    public void onSubscribe(Subscription s) {
        this.subscription = s;
        s.request(3);
    }

    public void onNext(Integer item) {
        System.out.println(item);
        if (++received % 3 == 0) subscription.request(3);
    }

    public void onError(Throwable t) { t.printStackTrace(); }
    public void onComplete() { System.out.println("done"); }
}

7. Terminal signals

onError và onComplete là terminal. Cancellation là consumer chủ động dừng demand, không phải error signal.

8. Sai lầm thường gặp

• Đồng nhất Reactive Streams với “chạy nhiều thread”.
• Nghĩ backpressure = bounded queue duy nhất.
• Emit sau onComplete.
• Gọi request(0) như một cách “pause”.
• Gọi callback đồng thời mà không đảm bảo serial signal.

9. Lab

1. Reject request n ≤ 0.
2. Không emit vượt demand.
3. Không emit sau cancel.
4. Không emit sau completion.
5. Subscriber request theo batch 5.
6. Log total demand và emitted.

10. Bài tập

11. Cầu nối sang Reactor

Reactive Streams contracts
         ↓
Publisher implementation + operators
         ↓
Project Reactor
         ↓
Mono / Flux

Tài liệu và code thực hành

Đọc chapter chi tiết [legacy source: modules/06-reactive-streams.md] · Mở foundation labs [legacy source: labs/foundation/README.md] · Xem Java source [legacy source: examples/README.md]