Triển khai Reactive Streams với Flow API trong Java 9

Flow API (package java.util.concurrent.Flow) là hỗ trợ chuẩn cho Reactive Streams kể từ Java 9. Nó kết hợp mô hình Iterator và Observer. Flow API là spec tương tác (interoperation spec), không phải thư viện end-user như RxJava.

4 Interface cốt lõi

• Publisher<T>: Xuất bản luồng dữ liệu cho Subscriber đã đăng ký.
• Subscriber<T>: Đăng ký nhận dữ liệu từ Publisher qua callback.
• Subscription: Liên kết giữa Publisher và Subscriber, điều khiển luồng (request/cancel).
• Processor<T,R>: Ngồi giữa Publisher và Subscriber, biến đổi luồng này thành luồng khác.

Ví dụ cơ bản với SubmissionPublisher

Java cung cấp SubmissionPublisher<T> – triển khai sẵn của Publisher.

import java.util.concurrent.Flow.*;
import java.util.List;
import java.util.concurrent.SubmissionPublisher;

class MySubscriber<T> implements Subscriber<T> {
    private Subscription subscription;

    @Override
    public void onSubscribe(Subscription subscription) {
        this.subscription = subscription;
        this.subscription.request(1); // yêu cầu 1 item
    }

    @Override
    public void onNext(T item) {
        System.out.println(item);
        subscription.request(1); // tiếp tục yêu cầu 1 item
    }

    @Override
    public void onError(Throwable throwable) {
        throwable.printStackTrace();
    }

    @Override
    public void onComplete() {
        System.out.println("Done");
    }
}

public class FlowTest {
    public static void main(String args[]) {
        List<String> items = List.of("1","2","3","4","5","6","7","8","9","10");
        SubmissionPublisher<String> publisher = new SubmissionPublisher<>();
        publisher.subscribe(new MySubscriber<>());

        items.forEach(s -> {
            try { Thread.sleep(1000); } catch (InterruptedException e) { e.printStackTrace(); }
            publisher.submit(s);
        });
        publisher.close();
    }
}

Kết quả

1 2 3 4 5 6 7 8 9 10 Done

Luồng hoạt động

1. Publisher tạo ra, Subscriber đăng ký qua subscribe().
2. Subscriber nhận Subscription trong onSubscribe(), gọi request(n).
3. Publisher đẩy dữ liệu qua submit(item) → Subscriber nhận trong onNext().
4. Subscriber xử lý xong, gọi request(1) để xin item tiếp theo (backpressure).
5. Khi Publisher close() → Subscriber nhận onComplete().

Flow API là nền tảng cho các thư viện reactive (Reactor, RxJava 2+) tương tác chuẩn nhau trên Java 9+.