Wie implementieren wir die Abonnentenschnittstelle in Java 9?

Java 9 unterstützt die Erstellung von reaktiven Streams durch die Einführung einiger Schnittstellen: Publisher, Subscriber, Subscription und SubmissionPublisher, die den Publisher implementieren Schnittstellenklasse. Jede Schnittstelle kann basierend auf den Prinzipien des Reactive Streaming eine andere Rolle spielen.

Wir können die Schnittstelle Subscriber verwenden, um von publisher veröffentlichte Daten zu abonnieren. Wir müssen die Subscriber -Schnittstelle implementieren und Implementierungen für die abstrakten Methoden bereitstellen.

Methoden der Flow.Subscriber-Schnittstelle:

• onComplete(): Diese Methode wird aufgerufen, wenn das Publisher-Objekt seine Rolle abschließt.
• onError(): Diese Methode wird aufgerufen, wenn der Herausgeber ein Problem hat und benachrichtigt den Abonnenten.
• onNext(): Diese Methode wird aufgerufen, wenn der Herausgeber neue Informationen hat, um alle Abonnenten zu benachrichtigen.
• onSubscribe(): Diese Methode wird aufgerufen, wenn Publisher einen Abonnenten hinzufügt.

Beispiel

import java.util.concurrent.Flow;
import java.util.concurrent.SubmissionPublisher;
import java.util.stream.IntStream;

public class SubscriberImplTest {
   public static class Subscriber implements <strong>Flow.Subscriber<Integer></strong> {
      private <strong>Flow.Subscription</strong> subscription;
      private boolean isDone;
      
     <strong> @Override</strong>
      public void <strong>onSubscribe</strong>(Flow.Subscription subscription) {
         System.out.println("Subscribed");
         this.subscription = subscription;
         this.subscription.request(1);
      }
      <strong>@Override</strong>
      public void <strong>onNext</strong>(Integer item) {
         System.out.println("Processing " + item);
         this.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("Processing done");
         isDone = true;
      }
   }
   public static void main(String args[]) throws InterruptedException {
      <strong>SubmissionPublisher<Integer></strong> publisher = new <strong>SubmissionPublisher<></strong><strong>()</strong>;
      <strong>Subscriber </strong>subscriber = new <strong>Subscriber()</strong>;
      publisher.subscribe(subscriber);
      <strong>IntStream</strong> intData = <strong>IntStream.rangeClosed</strong>(1, 10);
      intData.forEach(<strong>publisher::submit</strong>);
      publisher.<strong>close()</strong>;
      while(!subscriber.isDone) {
         Thread.sleep(10);
      }
      System.out.println("Done");
   }
}

Ausgabe

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

Das obige ist der detaillierte Inhalt vonWie implementieren wir die Abonnentenschnittstelle in Java 9?. Für weitere Informationen folgen Sie bitte anderen verwandten Artikeln auf der PHP chinesischen Website!