Java 9에서 Flow API를 사용하여 반응 스트림을 구현하는 방법은 무엇입니까?

흐름 API Java 9 이후 반응 스트림 사양에 대한 공식 지원입니다. Iterator 및 관찰자 패턴. 흐름 API RxJava와 같은 최종 사용자 API가 아닌 상호 운용 사양입니다. .

흐름 API 네 가지 기본 인터페이스로 구성됩니다.

구독자 : 구독자는 콜백을 위해 게시자를 구독합니다.
게시자 : 게시자는 등록된 구독자에게 데이터 항목 스트림을 게시합니다.
구독 : 게시자와 구독자 간의 링크입니다.
프로세서 : 프로세서는 게시자와 구독자 사이에 위치하며 한 스트림을 다른 스트림으로 변환합니다.

아래 예에서 우리는 하나의 데이터 객체를 요청하고 그것을 인쇄하고 하나를 더 요청하는 기본 가입자를 생성했습니다. 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();
   }
}

출력