Wie implementiert man reaktives Streaming mithilfe der Flow API in Java 9?

Flow API ist die offizielle Unterstützung für die Reactive Flows-Spezifikation seit Java 9. Es handelt sich um eine Kombination aus den Mustern Iterator und Observer . Flow API ist eine Interop-Spezifikation, keine Endbenutzer-API wie RxJava.

Flow API besteht aus vier grundlegenden Schnittstellen:

• Subscriber: Subscriber abonniert Rückrufe an den Herausgeber.
• Publisher: Publisher veröffentlicht einen Stream von Datenelementen für registrierte Abonnenten.
• Abonnements: Die Verbindung zwischen Herausgebern und Abonnenten.
• Prozessor: Der Prozessor sitzt zwischen Herausgeber und Abonnent und wandelt einen Stream in einen anderen um.

Im folgenden Beispiel erstellen wir einen Basisabonnenten, der ein Datenobjekt anfordert, es ausdruckt und ein anderes Datenobjekt anfordert. Wir können die von Java bereitgestellte Publisher-Implementierung (SubmissionPublisher) verwenden, um unsere Sitzung abzuschließen.

Beispiel

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

class MySubscriber<T>implements <strong>Flow.Subscriber<T></strong> {
   private <strong>Flow.Subscription</strong> subscription;
   <strong>@Override</strong>
   public void <strong>onSubscribe</strong>(Flow.Subscription subscription) {
      this.subscription = subscription;
      this.subscription.request(1);
   }
   <strong>@Override</strong>
   public void <strong>onNext</strong>(T item) {
      System.out.println(item);
      subscription.request(1);
   }
   <strong>@Override</strong>
   public void <strong>onError</strong>(Throwable throwable) {
      throwable.printStackTrace();
   }
   <strong>@Override</strong>
   public void <strong>onComplete</strong>() {
      System.out.println("Done");
   }
}

<strong>// main class</strong>
public class FlowTest {
   public static void main(String args[]) {
      <strong>List<String></strong> items = <strong>List.of</strong>("1", "2", "3", "4", "5", "6", "7", "8", "9", "10");
      <strong>SubmissionPublisher<String></strong> publisher = new SubmissionPublisher<>();
      publisher.<strong>subscribe</strong>(new MySubscriber<>());
      items.forEach(s -> {
         try {
            Thread.sleep(1000);
         } catch (InterruptedException e) {
            e.printStackTrace();
         }
         publisher.<strong>submit</strong>(s);
      });
      publisher.close();
   }
}

Ausgabe

<strong>1
2
3
4
5
6
7
8
9</strong>
<strong>10
Done</strong>

Das obige ist der detaillierte Inhalt vonWie implementiert man reaktives Streaming mithilfe der Flow API in Java 9?. Für weitere Informationen folgen Sie bitte anderen verwandten Artikeln auf der PHP chinesischen Website!