วิธีนำ Reactive Streams ไปใช้กับ Flow API ใน Java 9

Flow API เป็นการรองรับมาตรฐาน Reactive Streams อย่างเป็นทางการตั้งแต่ Java 9 เป็นการรวมของทั้งรูปแบบ Iterator และ Observer Flow API เป็นข้อกำหนดสำหรับการทำงานร่วมกัน (interoperation specification) ไม่ใช่ API สำหรับผู้ใช้ปลายทางแบบ RxJava

อินเทอร์เฟซหลัก 4 ตัวของ Flow API

Subscriber: ผู้สมัครรับข้อมูลจาก Publisher ผ่าน callback
Publisher: ผู้เผยแพร่สตรีมข้อมูลให้ Subscriber ที่ลงทะเบียน
Subscription: ลิงก์ระหว่าง Publisher และ Subscriber ควบคุมการไหลของข้อมูล
Processor: ตัวประมวลผลระหว่าง Publisher และ Subscriber แปลงสตรีมหนึ่งเป็นอีกสตรีมหนึ่ง

ตัวอย่างการใช้งาน SubmissionPublisher

Java มีการใช้งาน Publisher แบบพร้อมใช้คือ SubmissionPublisher นี่คือตัวอย่าง Subscriber ที่ขอข้อมูลทีละ 1 ชิ้น พิมพ์ออกมา แล้วขอต่อ

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

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

    @Override
    public void onSubscribe(Flow.Subscription subscription) {
        this.subscription = subscription;
        this.subscription.request(1);
    }

    @Override
    public void onNext(T item) {
        System.out.println(item);
        subscription.request(1);
    }

    @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();
    }
}

ผลลัพธ์

1 2 3 4 5 6 7 8 9 10 Done

Flow API ให้พื้นฐานสำหรับ reactive programming ใน Java มาตรฐาน ไม่ต้องพึ่งไลบรารีภายนอก