จะใช้งานสตรีมปฏิกิริยาโดยใช้ Flow API ใน Java 9 ได้อย่างไร

Flow API เป็นการสนับสนุนอย่างเป็นทางการสำหรับข้อกำหนดรีแอกทีฟสตรีมตั้งแต่ Java 9 เป็นการผสมผสานระหว่าง Iterator และ ผู้สังเกตการณ์ รูปแบบ Flow API เป็นข้อกำหนดการทำงานร่วมกันและไม่ใช่ API ของผู้ใช้ปลายทาง เช่น RxJava .

โฟลว์ API ประกอบด้วยอินเทอร์เฟซพื้นฐานสี่ส่วน:

สมาชิก : สมาชิกสมัครรับข้อมูลผู้จัดพิมพ์เพื่อโทรกลับ
ผู้เผยแพร่ : ผู้จัดพิมพ์เผยแพร่สตรีมรายการข้อมูลไปยังสมาชิกที่ลงทะเบียน
สมัครสมาชิก : การเชื่อมโยงระหว่างผู้จัดพิมพ์และผู้สมัครสมาชิก
โปรเซสเซอร์ : โปรเซสเซอร์ตั้งอยู่ระหว่าง Publisher และ Subscriber และแปลงสตรีมหนึ่งเป็นสตรีมอื่น

ในตัวอย่างด้านล่าง เราได้สร้างสมาชิกพื้นฐานที่ขอออบเจ็กต์ข้อมูลหนึ่งรายการ พิมพ์ออบเจ็กต์และขออีกรายการหนึ่ง เราสามารถใช้การนำไปใช้ของผู้เผยแพร่โดย Java (SubmissionPublisher ) เพื่อให้เซสชั่นของเราสมบูรณ์

ตัวอย่าง

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");
   }
}

// main class
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();
   }
}

ผลลัพธ์