Các interface cốt lõi của Reactive Streams trong Java 9

Java 9 giới thiệu Reactive Streams dưới package java.util.concurrent.Flow – hỗ trợ framework publish-subscribe interoperable. Xử lý async stream qua boundary (thread/thread-pool), receiver không bị buộc buffer tùy ý → tránh buffer overflow.

4 Interface cốt lõi (Flow API)

1. Publisher<T>

@FunctionalInterface
public static interface Publisher<T> {
    void subscribe(Subscriber<? super T> subscriber);
}

Phương thức subscribe() đăng ký Subscriber nhận dữ liệu.

2. Subscriber<T>

public static interface Subscriber<T> {
    void onSubscribe(Subscription subscription);
    void onNext(T item);
    void onError(Throwable throwable);
    void onComplete();
}

• onSubscribe: Nhận Subscription, gọi request(n) để xin data.
• onNext: Nhận item, xử lý, gọi request(1) cho item tiếp.
• onError: Xử lý lỗi.
• onComplete: Hoàn thành luồng.

3. Subscription

public static interface Subscription {
    void request(long n);
    void cancel();
}

• request(n): Yêu cầu n item (backpressure).
• cancel(): Hủy subscription, dừng nhận data.

4. Processor<T, R>

public static interface Processor<T, R> extends Subscriber<T>, Publisher<R> { }

Kết hợp Subscriber (nhận T) và Publisher (phát R) – biến đổi luồng.

Mối quan hệ

Publisher --subscribe()--> Subscriber
       <--Subscription-- 
       | request(n) / cancel()

Processor: Subscriber + Publisher (transform T -> R)

Đặc điểm Reactive Streams

• Asynchronous: Non-blocking, event-driven.
• Backpressure: Subscriber kiểm soát tốc độ qua request(n).
• Non-blocking: Không block thread khi chờ data.
• Interoperable: Chuẩn spec → Reactor, RxJava 2+, Akka Streams tuân thủ.

Ví dụ minh họa luồng

// Publisher tạo data
Publisher publisher = new SubmissionPublisher<>();

// Subscriber tiêu thụ
Subscriber subscriber = new Subscriber<>() {
    Subscription sub;
    public void onSubscribe(Subscription s) { sub = s; s.request(1); }
    public void onNext(Integer i) { System.out.println(i); sub.request(1); }
    public void onError(Throwable t) { t.printStackTrace(); }
    public void onComplete() { System.out.println("Done"); }
};

publisher.subscribe(subscriber);
publisher.submit(1);
publisher.submit(2);
publisher.close();

4 interface này là nền tảng cho toàn bộ ecosystem reactive trên JVM.