Sollen Produzenten und Konsumenten entkoppelt werden, so gibt es eine Reihe von Möglichkeiten. Nehmen wir zum Beispiel einen Iterator, der eine einfache API definiert, sodass sich auf immer die gleiche Art und Weise Daten von einem Produzenten abholen lassen. Oder nehmen wir einen Beobachter (Observer/Listener): ein Produzent verschickt seine Daten an Interessenten. Oder nehmen wir ein Bussystem: Publisher (Sender) und Subscriber (Empfänger) senden Datenpakete über einen Bus. All diese Design-Pattern sind für bestimmte Anwendungsfälle gut, haben jedoch alle eine Schwäche: es entstehen oft Blockaden und die Gefahr der Überflutung mit Nachrichten besteht.

Neu in Java 9 sind Schnittstellen eingezogen, die vorher unter http://www.reactive-streams.org/ diskutiert und als Defacto-Standard gelten. Ziel ist die Standardisierung von Programmen, die „reaktiv“ programmiert sind. Grundidee ist, dass es Publisher und Subscriber gibt, die einen asynchron Datenstrom austauschen aber nicht blockieren und mit „Gegendruck“ (engl. back pressure) arbeiten. Vergleichen wir das mit dem bekannten Bussystem, dann wird das Problem schnell klar; es kann passieren, dass ein Publisher viel zu schnell Ereignisse produziert, und die Subscriber nicht mitkommen, wo sollen dann die Daten hin? Oder, wenn der Publisher nichts sendet, dann wartet der Subscriber blockierend. Genau diese Probleme möchte das reaktive Modell vermeiden, in dem zwischen Publisher und Subscriber liegt ein Vermittler eingeschaltet wird, der über Gegendruck steuert, dass der Subscriber Daten anfordert und der Publisher dann so viel sendet, wie nicht-blockierend verarbeitet werden kann. Das realisiert eine Flusskontrolle, sodass wir auch von der Flow-API sprechen.

Die API ist untergebraucht in einer finalen Klasse java.util.concurrent.Flow, die vier statische innere Schnittstellen deklariert:

@FunctionalInterface  

public static interface Flow.Publisher<T> { 

    public void    subscribe(Flow.Subscriber<? super T> subscriber); 

}  

 

public static interface Flow.Subscriber<T> { 

    public void    onSubscribe(Flow.Subscription subscription); 

    public void    onNext(T item) ; 

    public void    onError(Throwable throwable) ; 

    public void    onComplete() ; 

}  

 

public static interface Flow.Subscription { 

    public void    request(long n); 

    public void    cancel() ; 

}  

 

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

}

Es ist gar nicht Aufgabe der Java SE diverse Implementierungen für die Schnittstellen bereitzustellen; es gibt lediglich von Publisher eine Implementierung, java.util.concurrent.SubmissionPublisher<T> . Allgemein sind die Implementierungen aber nicht trivial und daher ist es ratsam, sich mit zum Beispiel RxJava, „Reactive Extensions for the JVM“ (https://github.com/ReactiveX/RxJava) und Akka Streams (http://doc.akka.io/docs/akka/2.5.3/java/stream/index.html) zu beschäftigen, die Datenströme sehr schön in eine Kette verarbeiten können. Der eigentliche Wert der Java SE-Schnittstellen ist, dass es damit zu einer offiziellen API wird und Algorithmen problemlos unter den reaktiven Bibliotheken ausgetauscht werden können.

Ähnliche Beiträge

• Klassen mit einer abstrakten Methode als funktionale Schnittstelle in Java 8?
• Abstrakte überschriebe Schnittellenoperationen nehmen Default-Methoden weg
• Eine Sich-Selbst-Implementierung
• Offtopic: Flow, statische Typprüfung für JavaScript
• Klassen mit einer abstrakten Methode als funktionale Schnittstelle?