Überblick

Dieser Artikel erinnert mich an meine Forschungen zu Reactive Streams und der JDK Flow API.

"Flow API" (java.util.concurrent.Flow) ist eine in JDK 9 (JEP 266) eingeführte API und die Special Interest Group "Reactive Streams". Es entspricht der Spezifikation (Reactive Streams), die von einer Arbeitsgruppe namens `(SIG) erstellt wurde. Zu den JVM-Bibliotheken, die diese Spezifikation unterstützen, gehören Akka Streams (Lightbend, Inc.), ReactiveX / RxJava usw. Ja, Project Reactor (Pivotal Software, Inc.), das in Spring Web Flux verwendet wird, wird ebenfalls unterstützt.

Umgebung

Windows 10 Professional 1909
OpenJDK 13.0.1

Referenz

Über reaktive Streams

** Was sind reaktive Streams **

Das Folgende stammt aus dem Eröffnungssatz von Reactive Streams. (Japanische Übersetzung ist Google Übersetzung.)

Reactive Streams is an initiative to provide a standard for asynchronous stream processing with non-blocking back pressure. This encompasses efforts aimed at runtime environments (JVM and JavaScript) as well as network protocols.

Reactive Streams ist eine Initiative, die einen Standard für die asynchrone Stream-Verarbeitung mit nicht blockierendem Gegendruck bietet. Dies umfasst Arbeiten in Laufzeitumgebungen (JVM und JavaScript) und Netzwerkprotokollen.

Der Satz "** asynchrone Stromverarbeitung mit nicht blockierendem Gegendruck" in diesem Satz beschreibt klar die Eigenschaften von reaktiven Strömen. Die Erklärung der einzelnen Begriffe wird aus dem folgenden Glossar zitiert.

** Was ist nicht blockierend **

Nicht blockierend

Die API macht die Ressource zugänglich, wenn sie verfügbar ist. Andernfalls wird sie sofort zurückgegeben und dem Anrufer mitgeteilt, dass die Ressource derzeit nicht verfügbar ist oder dass der Vorgang gestartet und noch nicht abgeschlossen wurde. Die nicht blockierende API für Ressourcen ermöglicht es Anrufern, andere Arbeiten auszuführen, anstatt zu blockieren und darauf zu warten, dass Ressourcen verfügbar werden.

** Was ist Gegendruck? **

Gegendruck

Überladene Komponenten können nicht ohne Kontrolle katastrophal abstürzen oder Nachrichten verlieren. Wenn der Prozess stecken bleibt und sich einen Absturz nicht leisten kann, sollte die Komponente den vorgelagerten Komponenten mitteilen, dass sie überlastet ist, und die Last reduzieren. Dieser als Gegendruck bezeichnete Mechanismus ist ein wichtiger Rückkopplungsmechanismus, der langsam reagiert, ohne das System unter Überlastung zu stören.

** Was ist asynchron? **

Asynchron

Im Kontext einer reaktiven Deklaration bedeutet dies, dass eine von einem Client an einen Dienst gesendete Anforderung jederzeit nach dem Senden verarbeitet wird. Der Client kann die Ausführung der Anforderungsverarbeitung innerhalb des Zieldienstes nicht direkt beobachten oder synchronisieren.

Reactive Streams Specification for the JVM

Die von SIG erstellten Spezifikationen für JVM wurden ab November 2019 auf Version 1.0.3 aktualisiert.

Version 1.0.0: 30. April 2015
Version 1.0.1: 9. August 2017
Version 1.0.2: 19. Dezember 2017
Version 1.0.3: 23. August 2019

** Leistungen **

Zu den Leistungen von Maven gehören die folgenden, dies sind jedoch Spezifikationen, TCK (Technology Compatibility Kit) und Implementierungsbeispiele, sodass sie nicht direkt in normalen Projekten verwendet werden, sondern Bibliotheken wie Akka Streams, ReactiveX / RxJava und Reactor. Ich denke es wird getan.

<!-- https://mvnrepository.com/artifact/org.reactivestreams/reactive-streams -->
<dependency>
    <groupId>org.reactivestreams</groupId>
    <artifactId>reactive-streams</artifactId>
    <version>1.0.3</version>
</dependency>
<!-- https://mvnrepository.com/artifact/org.reactivestreams/reactive-streams-tck -->
<dependency>
    <groupId>org.reactivestreams</groupId>
    <artifactId>reactive-streams-tck</artifactId>
    <version>1.0.3</version>
    <scope>test</scope>
</dependency>
<!-- https://mvnrepository.com/artifact/org.reactivestreams/reactive-streams-tck-flow -->
<dependency>
    <groupId>org.reactivestreams</groupId>
    <artifactId>reactive-streams-tck-flow</artifactId>
    <version>1.0.3</version>
</dependency>
<!-- https://mvnrepository.com/artifact/org.reactivestreams/reactive-streams-examples -->
<dependency>
    <groupId>org.reactivestreams</groupId>
    <artifactId>reactive-streams-examples</artifactId>
    <version>1.0.3</version>
</dependency>

API Components

Die folgenden vier Schnittstellen sind in der Reactive Streams-Spezifikation für JVM Version 1.0.3 definiert.

Publisher

Publisher ist ein Anbieter von unbegrenzten oder endlich sequenzierten Elementen (dh Veröffentlichung eines Datenstroms), der Elemente veröffentlicht, wenn er eine Anfrage von einem Abonnenten (über Abonnement) erhält.

public interface Publisher<T> {
    public void subscribe(Subscriber<? super T> s);
}

Methode	Erläuterung
subscribe	Eine Factory-Methode, die Publisher auffordert, das Streaming von Daten zu starten. Kann für jedes neue Abonnement mehrmals aufgerufen werden.

Subscriber

Subscriber verbraucht die vom Publisher abonnierten Elemente. Die onXxx-Methode dieser Schnittstelle ist die Rückrufmethode, die dem Signal vom Publisher entspricht.

public interface Subscriber<T> {
    public void onSubscribe(Subscription s);
    public void onNext(T t);
    public void onError(Throwable t);
    public void onComplete();
}

Methode	Erläuterung
onSubscribe	Publisher#Wird nach dem Aufruf von subscribe ausgeführt. Der Abonnent fordert Daten an oder storniert sie, indem er das als Argument erhaltene Abonnement verwendet.
onNext	Subscription#Wird nach Anrufanforderung ausgeführt.
onError	Wird ausgeführt, wenn die Publisher-Datenübertragung fehlschlägt.
onComplete	Wird ausgeführt, wenn die Publisher-Datenübertragung normal abgeschlossen ist.(Einschließlich Stornierung)

Subscription

"Abonnement" ist eine Eins-zu-Eins-Darstellung eines Herausgebers und der Abonnenten, die diesen Herausgeber abonnieren. Der Abonnent fordert den Herausgeber auf, die Daten über die Abonnementmethode zu senden oder zu stornieren.

public interface Subscription {
    public void request(long n);
    public void cancel();
}

Methode	Erläuterung
request	Fordern Sie den Publisher auf, Daten zu senden.
cancel	Fordern Sie Publisher auf, das Senden von Daten zu beenden und Ressourcen zu bereinigen.

Processor

"Prozessor" ist eine Komponente, die sowohl Abonnenten- als auch Publisher-Funktionen bietet. Der Prozessor befindet sich zwischen dem Publisher am Anfang und dem Abonnenten am Ende. Es ist jedoch möglich, mehrere Prozessoren anstelle von nur einem zu verketten.

public interface Processor<T, R> extends Subscriber<T>, Publisher<R> {
}

Der Prozessor ist nicht immer erforderlich. Wenn dies nicht erforderlich ist, arbeiten Publisher und Abonnent direkt wie in der folgenden Abbildung dargestellt.

+-----------+              +------------+
|           | <-subscribe- |            |
| Publisher |              | Subscriber |
|           | <--request-- |            |
+-----------+              +------------+

Die folgende Abbildung zeigt ein Bild, wenn zwei Prozessoren (A, B) angeschlossen und platziert sind. Eine Situation, in der ein Prozessor in der Mitte wie dieser benötigt wird, ist, wenn Sie eine Filterung oder Datenkonvertierung in der Mitte eines Datenstroms durchführen möchten.

+-----------+              +-----------+              +-----------+              +------------+
|           | <-subscribe- |           | <-subscribe- |           | <-subscribe- |            |
| Publisher |              | Processor |              | Processor |              | Subscriber |
|           | <--request-- |    (A)    | <--request-- |    (B)    | <--request-- |            |
+-----------+              +-----------+              +-----------+              +------------+

Implementierungsbeispiel

Eine Beispielimplementierung finden Sie auf GitHub (reaktive-Streams / reaktive-Streams-jvm). Im Folgenden finden Sie ein Demo-Programm, das die AsyncIterablePublisher-Klasse verwendet, eines der Beispiele für die Publisher-Implementierung.

import lombok.extern.slf4j.Slf4j;
import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;
import org.reactivestreams.example.unicast.AsyncIterablePublisher;

import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
import java.util.stream.IntStream;

@Slf4j
public class Demo {

  public static void main(String ... args) {
    List<Integer> elements = IntStream.rangeClosed(1, 20).boxed().collect(Collectors.toList());
    ExecutorService executor = Executors.newFixedThreadPool(3);

    AsyncIterablePublisher<Integer> pub = new AsyncIterablePublisher<>(elements, executor);

    MySub mySub1 = new MySub("sub_1");
    MySub mySub2 = new MySub("sub_2");
    MySub mySub3 = new MySub("sub_3");

    log.info("start");

    // Publisher#Wenn Sie abonnieren anrufen
    //Die onSubscribe-Methode des Abonnenten wird zurückgerufen
    pub.subscribe(mySub1);
    pub.subscribe(mySub2);
    pub.subscribe(mySub3);

    log.info("end");

    try {
      //Warten Sie 30 Sekunden, bis der Vorgang aufgrund der asynchronen Verarbeitung abgeschlossen ist
      TimeUnit.SECONDS.sleep(30);
      executor.shutdown();
    } catch (InterruptedException e) {
      e.printStackTrace();
    }
  }

  static class MySub implements Subscriber<Integer> {
    private final String name;
    private Subscription s;

    public MySub(String name) {
      this.name = name;
    }

    private Long getId() {
      return Thread.currentThread().getId();
    }

    @Override
    public void onSubscribe(Subscription s) {
      log.info("({}) onSubscribe:[{}]", getId(), name);
      this.s = s;
      //Fordern Sie Publisher auf, Daten zu veröffentlichen, wenn das Abonnement abgeschlossen ist
      //Durch Anfordern in der onSubscribe-Methode beginnt die Datenausgabe gleichzeitig mit dem Abschluss des Abonnements.
      s.request(1);
    }

    @Override
    public void onNext(Integer integer) {
      //Die onNext-Methode wird zurückgerufen, wenn Daten von Publisher veröffentlicht werden
      log.info("({}) onNext:[{}] item:{}", getId(), name, integer);

      //Daten innerhalb dieser Methode verarbeiten
      //Führen Sie eine Datenverarbeitung durch

      //Fordern Sie Publisher auf, die folgenden Daten zu veröffentlichen
      s.request(1);

      //Oder Abbrechen
      //s.cancel();
    }

    @Override
    public void onError(Throwable t) {
      //Rückruf, wenn beim Veröffentlichen von Daten für Publisher ein Fehler auftritt
      log.info("onError:[{}]", name);
    }

    @Override
    public void onComplete() {
      //Rückruf, wenn Publisher-Daten veröffentlicht (oder storniert) werden
      log.info("({}) onComplete:[{}]", getId(), name);
    }

  }

}

Ausführungsergebnis

[main] INFO Demo - start
[main] INFO Demo - end
[pool-1-thread-2] INFO Demo - (15) onSubscribe:[sub_2]
[pool-1-thread-2] INFO Demo - (15) onNext:[sub_2] item:1
[pool-1-thread-2] INFO Demo - (15) onNext:[sub_2] item:2
[pool-1-thread-3] INFO Demo - (16) onSubscribe:[sub_3]
[pool-1-thread-1] INFO Demo - (14) onSubscribe:[sub_1]
[pool-1-thread-3] INFO Demo - (16) onNext:[sub_3] item:1
[pool-1-thread-2] INFO Demo - (15) onNext:[sub_2] item:3
[pool-1-thread-1] INFO Demo - (14) onNext:[sub_1] item:1
[pool-1-thread-2] INFO Demo - (15) onNext:[sub_2] item:4
[pool-1-thread-1] INFO Demo - (14) onNext:[sub_1] item:2
[pool-1-thread-3] INFO Demo - (16) onNext:[sub_3] item:2
[pool-1-thread-2] INFO Demo - (15) onNext:[sub_2] item:5
[pool-1-thread-1] INFO Demo - (14) onNext:[sub_1] item:3
[pool-1-thread-3] INFO Demo - (16) onNext:[sub_3] item:3
[pool-1-thread-2] INFO Demo - (15) onNext:[sub_2] item:6
[pool-1-thread-1] INFO Demo - (14) onNext:[sub_1] item:4
[pool-1-thread-3] INFO Demo - (16) onNext:[sub_3] item:4
[pool-1-thread-2] INFO Demo - (15) onNext:[sub_2] item:7
[pool-1-thread-1] INFO Demo - (14) onNext:[sub_1] item:5
[pool-1-thread-3] INFO Demo - (16) onNext:[sub_3] item:5
[pool-1-thread-2] INFO Demo - (15) onNext:[sub_2] item:8
[pool-1-thread-1] INFO Demo - (14) onNext:[sub_1] item:6
[pool-1-thread-3] INFO Demo - (16) onNext:[sub_3] item:6
[pool-1-thread-1] INFO Demo - (14) onNext:[sub_1] item:7
[pool-1-thread-2] INFO Demo - (15) onNext:[sub_2] item:9
[pool-1-thread-1] INFO Demo - (14) onNext:[sub_1] item:8
[pool-1-thread-2] INFO Demo - (15) onNext:[sub_2] item:10
[pool-1-thread-3] INFO Demo - (16) onNext:[sub_3] item:7
[pool-1-thread-2] INFO Demo - (15) onNext:[sub_2] item:11
[pool-1-thread-1] INFO Demo - (14) onNext:[sub_1] item:9
[pool-1-thread-3] INFO Demo - (16) onNext:[sub_3] item:8
[pool-1-thread-2] INFO Demo - (15) onNext:[sub_2] item:12
[pool-1-thread-1] INFO Demo - (14) onNext:[sub_1] item:10
[pool-1-thread-3] INFO Demo - (16) onNext:[sub_3] item:9
[pool-1-thread-2] INFO Demo - (15) onNext:[sub_2] item:13
[pool-1-thread-1] INFO Demo - (14) onNext:[sub_1] item:11
[pool-1-thread-3] INFO Demo - (16) onNext:[sub_3] item:10
[pool-1-thread-2] INFO Demo - (15) onNext:[sub_2] item:14
[pool-1-thread-1] INFO Demo - (14) onNext:[sub_1] item:12
[pool-1-thread-2] INFO Demo - (15) onNext:[sub_2] item:15
[pool-1-thread-1] INFO Demo - (14) onNext:[sub_1] item:13
[pool-1-thread-2] INFO Demo - (15) onNext:[sub_2] item:16
[pool-1-thread-2] INFO Demo - (15) onNext:[sub_2] item:17
[pool-1-thread-2] INFO Demo - (15) onNext:[sub_2] item:18
[pool-1-thread-2] INFO Demo - (15) onNext:[sub_2] item:19
[pool-1-thread-1] INFO Demo - (14) onNext:[sub_1] item:14
[pool-1-thread-2] INFO Demo - (15) onNext:[sub_2] item:20
[pool-1-thread-1] INFO Demo - (14) onNext:[sub_1] item:15
[pool-1-thread-2] INFO Demo - (15) onComplete:[sub_2]
[pool-1-thread-1] INFO Demo - (14) onNext:[sub_1] item:16
[pool-1-thread-3] INFO Demo - (16) onNext:[sub_3] item:11
[pool-1-thread-3] INFO Demo - (16) onNext:[sub_3] item:12
[pool-1-thread-3] INFO Demo - (16) onNext:[sub_3] item:13
[pool-1-thread-1] INFO Demo - (14) onNext:[sub_1] item:17
[pool-1-thread-2] INFO Demo - (15) onNext:[sub_3] item:14
[pool-1-thread-1] INFO Demo - (14) onNext:[sub_1] item:18
[pool-1-thread-2] INFO Demo - (15) onNext:[sub_3] item:15
[pool-1-thread-1] INFO Demo - (14) onNext:[sub_1] item:19
[pool-1-thread-2] INFO Demo - (15) onNext:[sub_3] item:16
[pool-1-thread-2] INFO Demo - (15) onNext:[sub_3] item:17
[pool-1-thread-1] INFO Demo - (14) onNext:[sub_1] item:20
[pool-1-thread-2] INFO Demo - (15) onNext:[sub_3] item:18
[pool-1-thread-1] INFO Demo - (14) onComplete:[sub_1]
[pool-1-thread-2] INFO Demo - (15) onNext:[sub_3] item:19
[pool-1-thread-2] INFO Demo - (15) onNext:[sub_3] item:20
[pool-1-thread-2] INFO Demo - (15) onComplete:[sub_3]

JDK Flow API

java.util.concurrent.Flow

Die Flow-Klasse deklariert vier Schnittstellen, die der Spezifikation Reactive Streams entsprechen. Sie müssen diese Schnittstellen implementieren, wenn Sie Anwendungen entwickeln, die reaktive Streams unterstützen.

public final class Flow {

    @FunctionalInterface
    public static interface Publisher<T> {
        public void subscribe(Subscriber<? super T> subscriber);
    }

    public static interface Subscriber<T> {
        public void onSubscribe(Subscription subscription);
        public void onNext(T item);
        public void onError(Throwable throwable);
        public void onComplete();
    }

    public static interface Subscription {
        public void request(long n);
        public void cancel();
    }

    public static interface Processor<T, R> extends Subscriber<T>, Publisher<R> {
    }

}

SubmissionPublisher<T>

Für Publisher gibt es eine Implementierungsklasse namens SubmissionPublisher \ <T >, die unverändert verwendet oder vererbt werden kann, um die ursprüngliche Verarbeitung zu implementieren.

Konstrukteur

Konstrukteur
SubmissionPublisher()
SubmissionPublisher(Executor executor, int maxBufferCapacity)
SubmissionPublisher(Executor executor, int maxBufferCapacity, BiConsumer<? super Flow.Subscriber<? super T>,? super Throwable> handler)

maxBufferCapacity wird auf eine Potenz von 2 gerundet.

try (SubmissionPublisher<Integer> pub = new SubmissionPublisher<>()) {
  //Kürzung
}

try (SubmissionPublisher<Integer> pub = new SubmissionPublisher<>(ForkJoinPool.commonPool(), 8)) {
  //Kürzung
}

ExecutorService executor = Executors.newFixedThreadPool(3);
try (SubmissionPublisher<Integer> pub = new SubmissionPublisher<>(executor, 8, (subscriber, throwable) -> {
})) {
  //Kürzung
}

Datenausgabe

Die SubmissionPublisher-Klasse verfügt über "Submit" - und "Offer" -Methoden zum Veröffentlichen von Daten.

Datenveröffentlichungsmethode
public int submit(T item)
public int offer(T item, BiPredicate<Flow.Subscriber<? super T>,? super T> onDrop)
public int offer(T item, long timeout, TimeUnit unit, BiPredicate<Flow.Subscriber<? super T>,? super T> onDrop)

submit

Senden Sie Blöcke, bis die Daten gesendet werden können.

int lag = pub.submit(value);

if (lag < 0) {
  //Senden wird nicht gelöscht
} else {
  //Maximale Verzögerungsschätzung(Anzahl der gesendeten, aber noch nicht verbrauchten Artikel)
}

offer

Das Angebot blockiert die Datenübertragung nicht und kann die Verarbeitung ausführen (ob erneut gesendet werden soll oder nicht usw.), wenn die Daten nicht übertragen werden können. In diesem Beispiel werden die Daten ohne erneutes Senden gelöscht.

int lag = offer(item, (subscriber, value) -> {
  subscriber.onError(new RuntimeException("drop item:[" + integer + "]"));
  return false; //Nicht erneut senden
});

if (lag < 0) {
  //Anzahl der Tropfen
} else {
  //Maximale Verzögerungsschätzung(Anzahl der gesendeten, aber noch nicht verbrauchten Artikel)
}

offer

Sie können auch eine Zeitüberschreitung angeben. Wenn es in diesem Beispiel nicht gesendet werden kann, wartet es bis zu 1 Sekunde.

int lag = pub.offer(value, 1, TimeUnit.SECONDS, (subscriber, integer) -> {
  subscriber.onError(new RuntimeException("drop item:[" + integer + "]"));
  return false; //Nicht erneut senden
});

if (lag < 0) {
  //Anzahl der Tropfen
} else {
  //Maximale Verzögerungsschätzung(Anzahl der gesendeten, aber noch nicht verbrauchten Artikel)
}

Implementierungsbeispiel

Unten finden Sie ein Demo-Programm, das die SubmissionPublisher-Klasse verwendet.

import lombok.extern.slf4j.Slf4j;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Flow;
import java.util.concurrent.SubmissionPublisher;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.IntStream;

@Slf4j
public class Demo {

  public static void main(String ... args) {
    log.info("start");

    MySub<Integer> mySub1 = new MySub<>("sub_1");
    MySub<Integer> mySub2 = new MySub<>("sub_2");
    MySub<Integer> mySub3 = new MySub<>("sub_3");

    ExecutorService executor = Executors.newFixedThreadPool(3);

    try (SubmissionPublisher<Integer> pub = new SubmissionPublisher<>(executor, 256)) {

      pub.subscribe(mySub1);
      pub.subscribe(mySub2);
      pub.subscribe(mySub3);

      log.info("NumberOfSubscribers:{}", pub.getNumberOfSubscribers());
      log.info("MaxBufferCapacity:{}", pub.getMaxBufferCapacity());

      IntStream.rangeClosed(1, 100000).forEach(value -> {
        log.info("publish:{} estimateMinimumDemand:{} estimateMaximumLag:{}", value, pub.estimateMinimumDemand(), pub.estimateMaximumLag());

        int lag = pub.offer(value, 1, TimeUnit.SECONDS, (subscriber, integer) -> {
          log.info("publish offer on drop:{}", integer);
          subscriber.onError(new RuntimeException("drop item:[" + integer + "]"));
          return false; //Nicht erneut senden
        });

        if (lag < 0) {
          //Anzahl der Tropfen
          log.info("drops:{}", lag * -1);
        } else {
          //Maximale Verzögerungsschätzung(Anzahl der gesendeten, aber noch nicht verbrauchten Artikel)
          log.info("lag:{}", lag);
        }

      });

    }

    log.info("end");

    try {
      TimeUnit.SECONDS.sleep(10);

      mySub1.result();
      mySub2.result();
      mySub3.result();

      if (!executor.isShutdown()) {
        log.info("shutdown");
        executor.shutdown();
      }

    } catch (InterruptedException e) {
      e.printStackTrace();
    }

  }

  static class MySub<Integer> implements Flow.Subscriber<Integer> {
    private final String name;
    private AtomicInteger success = new AtomicInteger(0);
    private AtomicInteger error = new AtomicInteger(0);
    private Flow.Subscription s;

    public MySub(String name) {
      this.name = name;
    }

    private Long getId() {
      return Thread.currentThread().getId();
    }

    @Override
    public void onSubscribe(Flow.Subscription subscription) {
      log.info("({}) onSubscribe:[{}]", getId(), name);
      this.s = subscription;
      s.request(1);
    }

    @Override
    public void onNext(Integer item) {
      log.info("({}) onNext:[{}] item:{}", getId(), name, item);
      success.incrementAndGet();
      s.request(1);
    }

    @Override
    public void onError(Throwable throwable) {
      log.info("({}) onError:[{}]", getId(), name);
      error.incrementAndGet();
    }

    @Override
    public void onComplete() {
      log.info("({}) onComplete:[{}]", getId(), name);
    }

    public void result() {
      log.info("result:[{}] success:{} error:{}", name, success.get(), error.get());
    }

  }

}

[JAVA] Erste Schritte mit Reactive Streams und der JDK 9 Flow API

Überblick

Über reaktive Streams

Implementierungsbeispiel

Konstrukteur

Datenausgabe

Implementierungsbeispiel