Làm cách nào để triển khai các luồng phản ứng bằng API luồng trong Java 9?

API luồng là hỗ trợ chính thức cho đặc tả luồng phản ứng kể từ Java 9. Nó là sự kết hợp của cả Iterator và Người quan sát các mẫu. API luồng là đặc điểm kỹ thuật tương tác chứ không phải API người dùng cuối như RxJava .

API luồng bao gồm bốn giao diện cơ bản:

Người đăng ký : Người đăng ký đăng ký Nhà xuất bản để được gọi lại.
Nhà xuất bản : Nhà xuất bản xuất bản luồng các mục dữ liệu cho những người đăng ký đã đăng ký.
Đăng ký : Mối liên kết giữa nhà xuất bản và người đăng ký.
Bộ xử lý : Bộ xử lý nằm giữa Nhà xuất bản và Người đăng ký và chuyển đổi luồng này sang luồng khác.

Trong ví dụ dưới đây, chúng tôi đã tạo một người đăng ký cơ bản yêu cầu một đối tượng dữ liệu, in nó và yêu cầu một đối tượng khác. Chúng tôi có thể sử dụng triển khai dành cho nhà xuất bản do Java cung cấp ( SubmissionPublisher ) để hoàn thành phiên của chúng tôi.

Ví dụ

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

Đầu ra