Unterschied zwischen CompletableFuture, Future und RxJava’s Observable

Ich würde gerne den Unterschied zwischen wissen
CompletableFuture,Future und Observable RxJava.

Was ich weiß ist, dass alle asynchron sind, aber

Future.get() blockiert den Thread

CompletableFuture gibt die Callback-Methoden an

RxJava Observable — ähnlich zu CompletableFuture mit anderen Vorteilen (nicht sicher)

Zum Beispiel: wenn der Kunde mehrere Serviceanrufe tätigen muss und wann wir verwenden Futures (Java) Future.get() wird nacheinander ausgeführt … würde gerne wissen, wie es in RxJava besser ist …

Und die Dokumentation http://reactivex.io/intro.html sagt

Es ist schwierig, Futures zu verwenden, um bedingte asynchrone Ausführungsabläufe optimal zusammenzustellen (oder unmöglich, da die Latenzen jeder Anforderung zur Laufzeit variieren). Das geht natürlich, wird aber schnell kompliziert (und damit fehleranfällig) oder blockiert vorzeitig auf Future.get(), wodurch der Vorteil der asynchronen Ausführung zunichte gemacht wird.

Wirklich interessiert zu wissen, wie RxJava löst dieses Problem. Ich fand es schwierig, aus der Dokumentation zu verstehen.

Zukünfte

Zukünfte wurden in Java 5 (2004) eingeführt. Sie sind im Grunde Platzhalter für das Ergebnis einer Operation, die noch nicht abgeschlossen ist. Sobald der Vorgang abgeschlossen ist, wird die Future enthält dieses Ergebnis. Eine Operation kann beispielsweise a sein Lauffähig oder Abrufbar Instanz, die an eine gesendet wird ExecutorService. Der Übergebende des Vorgangs kann die verwenden Future Objekt, um zu überprüfen, ob der Vorgang erledigt()oder warten Sie, bis die Blockierung beendet ist erhalten() Methode.

Beispiel:

/**
* A task that sleeps for a second, then returns 1
**/
public static class MyCallable implements Callable<Integer> {

    @Override
    public Integer call() throws Exception {
        Thread.sleep(1000);
        return 1;
    }

}

public static void main(String[] args) throws Exception{
    ExecutorService exec = Executors.newSingleThreadExecutor();
    Future<Integer> f = exec.submit(new MyCallable());

    System.out.println(f.isDone()); //False

    System.out.println(f.get()); //Waits until the task is done, then prints 1
}

Erfüllbare Futures

Erfüllbare Futures wurden in Java 8 (2014) eingeführt. Sie sind in der Tat eine Weiterentwicklung der regulären Futures, inspiriert von Googles Hörbare ZukunftTeil von Guave Bibliothek. Sie sind Futures, die es Ihnen auch ermöglichen, Aufgaben in einer Kette aneinander zu reihen. Sie können sie verwenden, um einem Worker-Thread zu sagen, dass er “eine Aufgabe X erledigen soll, und wenn Sie fertig sind, machen Sie diese andere Sache mit dem Ergebnis von X”. Mit CompletableFutures können Sie etwas mit dem Ergebnis der Operation tun, ohne tatsächlich einen Thread zu blockieren, um auf das Ergebnis zu warten. Hier ist ein einfaches Beispiel:

/**
* A supplier that sleeps for a second, and then returns one
**/
public static class MySupplier implements Supplier<Integer> {

    @Override
    public Integer get() {
        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            //Do nothing
        }
        return 1;
    }
}

/**
* A (pure) function that adds one to a given Integer
**/
public static class PlusOne implements Function<Integer, Integer> {

    @Override
    public Integer apply(Integer x) {
        return x + 1;
    }
}

public static void main(String[] args) throws Exception {
    ExecutorService exec = Executors.newSingleThreadExecutor();
    CompletableFuture<Integer> f = CompletableFuture.supplyAsync(new MySupplier(), exec);
    System.out.println(f.isDone()); // False
    CompletableFuture<Integer> f2 = f.thenApply(new PlusOne());
    System.out.println(f2.get()); // Waits until the "calculation" is done, then prints 2
}

RxJava

RxJava ist ganze Bibliothek für reaktive Programmierung bei Netflix erstellt. Auf den ersten Blick scheint es ähnlich zu sein Streams von Java 8. Es ist, außer dass es viel mächtiger ist.

Ähnlich wie Futures kann RxJava verwendet werden, um eine Reihe synchroner oder asynchroner Aktionen aneinanderzureihen, um eine Verarbeitungspipeline zu erstellen. Im Gegensatz zu Futures, die nur einmal verwendet werden, arbeitet RxJava weiter Ströme von null oder mehr Artikeln. Einschließlich endloser Streams mit einer unendlichen Anzahl von Artikeln. Es ist auch viel flexibler und leistungsfähiger dank einer unglaublich reichen Reihe von Operatoren.

Im Gegensatz zu den Streams von Java 8 hat RxJava auch eine Gegendruck Mechanismus, der es ermöglicht, Fälle zu behandeln, in denen verschiedene Teile Ihrer Verarbeitungspipeline in verschiedenen Threads arbeiten, zu unterschiedlichen Preisen.

Der Nachteil von RxJava ist, dass es trotz der soliden Dokumentation aufgrund des damit verbundenen Paradigmenwechsels eine schwierig zu erlernende Bibliothek ist. Das Debuggen von Rx-Code kann auch ein Albtraum sein, insbesondere wenn mehrere Threads beteiligt sind, und noch schlimmer – wenn Gegendruck erforderlich ist.

Wenn Sie darauf eingehen wollen, gibt es ein Ganzes Seite von verschiedenen Tutorials auf der offiziellen Website, plus die offizielle Dokumentation und Javadoc. Sie können sich auch einige der Videos ansehen, z Dieses hier die eine kurze Einführung in Rx gibt und auch über die Unterschiede zwischen Rx und Futures spricht.

Bonus: Reaktive Java 9-Streams

Reaktive Streams von Java 9 auch bekannt Flow-API sind eine Reihe von Schnittstellen, die von verschiedenen implementiert werden reaktive Ströme Bibliotheken wie z Rx Java 2, Akka-Streamsund Scheitelpunkt. Sie ermöglichen es diesen reaktiven Bibliotheken, sich miteinander zu verbinden, während der so wichtige Gegendruck erhalten bleibt.

Ich arbeite seit 0.9 mit Rx Java, jetzt bei 1.3.2 und migriere bald auf 2.x. Ich benutze dies in einem privaten Projekt, an dem ich bereits seit 8 Jahren arbeite.

Ohne diese Bibliothek würde ich überhaupt nicht mehr programmieren. Am Anfang war ich skeptisch, aber es ist eine völlig andere Geisteshaltung, die Sie schaffen müssen. Ziemlich schwierig am Anfang. Ich habe manchmal stundenlang auf die Murmeln geschaut.. lol

Es ist nur eine Frage der Übung und des wirklichen Kennenlernens des Flusses (auch bekannt als Vertrag von Observables und Observer), sobald Sie dort angekommen sind, werden Sie es hassen, es anders zu tun.

Für mich gibt es keinen wirklichen Nachteil an dieser Bibliothek.

Anwendungsfall: Ich habe eine Monitoransicht, die 9 Messgeräte enthält (CPU, Speicher, Netzwerk usw.). Beim Starten der Ansicht abonniert sich die Ansicht selbst bei einer Systemmonitorklasse, die ein Observable (Intervall) zurückgibt, das alle Daten für die 9 Zähler enthält. Es wird jede Sekunde ein neues Ergebnis zur Ansicht geschoben (also kein Polling !!!). Dieses Observable verwendet eine Flatmap, um gleichzeitig (asynchron!) Daten aus 9 verschiedenen Quellen abzurufen und das Ergebnis in ein neues Modell zu komprimieren, das Ihre Ansicht auf onNext() erhalten wird.

Wie zum Teufel machst du das mit Futures, Completables etc … Viel Glück! 🙂

Rx Java löst für mich viele Probleme beim Programmieren und macht es in gewisser Weise viel einfacher …

Vorteile:

• Staatenlos !!! (Wichtig zu erwähnen, vielleicht das Wichtigste)
• Thread-Management out of the box
• Erstellen Sie Sequenzen, die ihren eigenen Lebenszyklus haben
• Alles ist beobachtbar, so dass die Verkettung einfach ist
• Weniger Code zu schreiben
• Einzelnes Glas auf Klassenpfad (sehr leicht)
• Sehr zeitgleich
• Keine Callback-Hölle mehr
• Abonnentenbasiert (enger Vertrag zwischen Verbraucher und Erzeuger)
• Rückstaustrategien (Leistungsschalter etc.)
• Hervorragende Fehlerbehandlung und -wiederherstellung
• Sehr schöne Dokumentation (Murmeln <3)
• Komplette Kontrolle
• Viel mehr …

Nachteile: – Schwer zu testen

Javas Future ist ein Platzhalter für etwas, das in Zukunft mit einer blockierenden API vervollständigt wird. Du musst es benutzen isDone() Methode, um sie regelmäßig abzufragen, um zu überprüfen, ob diese Aufgabe abgeschlossen ist. Natürlich können Sie Ihren eigenen asynchronen Code implementieren, um die Abfragelogik zu verwalten. Es verursacht jedoch mehr Boilerplate-Code und Debug-Overhead.

Javas CompletableFuture wird von Scala’s Future innoviert. Es trägt eine interne Callback-Methode. Sobald es fertig ist, wird die Callback-Methode ausgelöst und teilt dem Thread mit, dass die Downstream-Operation ausgeführt werden soll. Deshalb hat es thenApply -Methode, um weitere Operationen an dem Objekt auszuführen, das in die eingeschlossen ist CompletableFuture.

RxJavas Observable ist eine verbesserte Version von CompletableFuture. Es ermöglicht Ihnen, den Gegendruck zu handhaben. In dem thenApply Methode (und sogar mit seinen Brüdern thenApplyAsync), die wir oben erwähnt haben, kann diese Situation eintreten: Die Downstream-Methode möchte einen externen Dienst aufrufen, der manchmal nicht verfügbar ist. In diesem Fall ist die CompleteableFuture wird vollständig fehlschlagen und Sie müssen den Fehler selbst beheben. Jedoch, Observable ermöglicht es Ihnen, den Gegendruck zu handhaben und die Ausführung fortzusetzen, sobald der externe Dienst verfügbar ist.

Darüber hinaus gibt es eine ähnliche Schnittstelle von Observable: Flowable. Sie sind für unterschiedliche Zwecke konzipiert. Normalerweise Flowable widmet sich der Handhabung der kalten und nicht zeitgesteuerten Operationen, während Observable widmet sich der Abwicklung von Hinrichtungen, die sofortige Antworten erfordern. Sehen Sie hier die offiziellen Dokumente: https://github.com/ReactiveX/RxJava#backpressure

Alle drei Schnittstellen dienen dazu, Werte vom Erzeuger zum Verbraucher zu übertragen. Es gibt zwei Arten von Verbrauchern:

• synchron: Der Konsument führt einen blockierenden Aufruf durch, der zurückkehrt, wenn der Wert bereit ist
• asynchron: Wenn der Wert bereit ist, wird eine Callback-Methode des Verbrauchers aufgerufen

Kommunikationsschnittstellen unterscheiden sich auch auf andere Weise:

• in der Lage, einzelne Werte von mehreren Werten zu übertragen
• Bei mehreren Werten kann Gegendruck unterstützt werden oder nicht

Als Ergebnis:

• Future überträgt Einzelwert über synchrone Schnittstelle

• CompletableFuture überträgt einzelne Werte sowohl über synchrone als auch über asynchrone Schnittstellen

• Rx überträgt mehrere Werte über eine asynchrone Schnittstelle mit Gegendruck

Außerdem unterstützen alle diese Kommunikationseinrichtungen die Übertragung von Ausnahmen. Dies ist nicht immer der Fall. Zum Beispiel, BlockingQueue nicht.

Der Hauptvorteil von CompletableFuture gegenüber dem normalen Future besteht darin, dass CompletableFuture die extrem leistungsstarke Stream-API nutzt und Ihnen Callback-Handler zum Verketten Ihrer Aufgaben zur Verfügung stellt, was bei Verwendung des normalen Future absolut fehlt. Zusammen mit der Bereitstellung einer asynchronen Architektur ist CompletableFuture der richtige Weg, um rechenintensive Map-Reduce-Aufgaben zu bewältigen, ohne sich große Gedanken über die Anwendungsleistung machen zu müssen.