Cách triển khai SubmissionPublisher trong Java 9

Java 9 giới thiệu Reactive Streams với 4 interface core: Publisher, Subscriber, Subscription, Processor và 1 class concrete: SubmissionPublisher<T> implements Flow.Publisher<T>, AutoCloseable. Dùng submit() để publish item cho từng subscriber.

Cú pháp

public class SubmissionPublisher<T> extends Object implements Flow.Publisher<T>, AutoCloseable

Ví dụ đầy đủ: Multi-subscriber

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

class MySubscriber<T> implements Subscriber<T> {
    private Subscription subscription;
    private String name;
    
    public MySubscriber(String name) { this.name = name; }
    
    @Override public void onSubscribe(Subscription subscription) {
        this.subscription = subscription;
        System.out.println(name + ": onSubscribe");
        subscription.request(1);
    }
    
    @Override public void onNext(T msg) {
        System.out.println(name + ": " + msg + " received in onNext");
        subscription.request(1);
    }
    
    @Override public void onError(Throwable t) {
        System.out.println(name + ": onError");
        t.printStackTrace();
    }
    
    @Override public void onComplete() {
        System.out.println(name + ": onComplete");
    }
}

public class FlowTest {
    public static void main(String args[]) {
        SubmissionPublisher<String> publisher = new SubmissionPublisher<>();
        
        MySubscriber<String> subscriber = new MySubscriber<>("Mine");
        MySubscriber<String> subscriberYours = new MySubscriber<>("Yours");
        MySubscriber<String> subscriberHis = new MySubscriber<>("His");
        MySubscriber<String> subscriberHers = new MySubscriber<>("Her");
        
        publisher.subscribe(subscriber);
        publisher.subscribe(subscriberYours);
        publisher.subscribe(subscriberHis);
        publisher.subscribe(subscriberHers);
        
        publisher.submit("One");
        publisher.submit("Two");
        publisher.submit("Three");
        publisher.submit("Four");
        publisher.submit("Five");
        
        try { Thread.sleep(1000); } catch(InterruptedException e) { e.printStackTrace(); }
        publisher.close();
    }
}

Kết quả (thứ tự có thể khác do concurrency)

Yours: onSubscribe
His: onSubscribe
Mine: onSubscribe
His: One received in onNext
Yours: One received in onNext
Mine: One received in onNext
Yours: Two received in onNext
His: Two received in onNext
Yours: Three received in onNext
Mine: Two received in onNext
Yours: Four received in onNext
His: Three received in onNext
Yours: Five received in onNext
Mine: Three received in onNext
Her: onSubscribe
His: Four received in onNext
Her: One received in onNext
Mine: Four received in onNext
Her: Two received in onNext
His: Five received in onNext
Her: Three received in onNext
Mine: Five received in onNext
Her: Four received in onNext
Her: Five received in onNext
Yours: onComplete
His: onComplete
Mine: onComplete
Her: onComplete

Đặc điểm SubmissionPublisher

• Buffered: Mặc định buffer 256 items/subscription (configurable qua constructor).
• Backpressure: Subscriber gọi request(n) kiểm soát luồng.
• Thread-safe: submit() an toàn đa luồng.
• AutoCloseable: close() gửi onComplete() cho tất cả subscriber.
• Executor tùy chỉnh: Constructor nhận Executor cho delivery.

Lưu ý

Flow API là spec interoperability – không phải thư viện reactive đầy đủ (như Reactor/RxJava). Dùng làm foundation cho các thư viện đó implement Publisher/Subscriber chuẩn.