我想知道它們之間的區(qū)別
CompletableFuture,Future和Observable RxJava.

我所知道的都是異步但是

Future.get()阻塞線程

CompletableFuture提供了回調(diào)方法

RxJava Observable —與CompletableFuture類似,具有其他好處(不確定)

例如：如果客戶端需要進(jìn)行多次服務(wù)調(diào)用,當(dāng)我們使用Futures(Java)時,Future.get()將按順序執(zhí)行…想知道它在RxJava中的效果如何…

文檔http:///intro.html說

很難使用Futures來優(yōu)化組合條件異步執(zhí)行流程(或者不可能,因為每個請求的延遲在運行時會有所不同).當(dāng)然,這可以完成,但它很快變得復(fù)雜(因此容易出錯)或者過早地阻塞Future.get(),這消除了異步執(zhí)行的好處.

真的很想知道RxJava如何解決這個問題.我發(fā)現(xiàn)從文檔中很難理解.

解決方法:

期貨

在Java 5(2004)中引入了Futures.它們基本上是一個尚未完成的操作結(jié)果的占位符.一旦操作完成,Future將包含該結(jié)果.例如,操作可以是提交到ExecutorService的Runnable或Callable實例.操作的提交者可以使用Future對象檢查操作isDone(),或者等待它使用阻塞get()方法完成.

例：

/**
* 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
}

CompletableFutures

CompletableFutures在Java 8(2014)中引入.它們實際上是常規(guī)期貨的演變,受到谷歌Listenable Futures(Guava庫的一部分)的啟發(fā).它們是Futures,它還允許您將任務(wù)串聯(lián)起來.您可以使用它們來告訴某些工作線程“去做一些任務(wù)X,當(dāng)你完成后,使用X的結(jié)果去做其他事情”.使用CompletableFutures,您可以對操作結(jié)果執(zhí)行某些操作,而不會實際阻止線程等待結(jié)果.這是一個簡單的例子：

/**
* 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是在Netflix創(chuàng)建的reactive programming的完整庫.一目了然,它看起來與Java 8’s streams相似.它是,除了它更強(qiáng)大.

與Futures類似,RxJava可用于將一堆同步或異步操作串聯(lián)在一起以創(chuàng)建處理管道.與一次性使用的Futures不同,RxJava適用于零個或多個項目的流.包括具有無限數(shù)量項目的永無止境的流.由于擁有令人難以置信的豐富set of operators,它也更加靈活和強(qiáng)大.

與Java 8的流不同,RxJava還具有backpressure機(jī)制,允許它處理處理管道的不同部分以不同速率在不同線程中運行的情況.

RxJava的缺點是,盡管有可靠的文檔,但由于涉及范式的轉(zhuǎn)變,它是一個具有挑戰(zhàn)性的庫. Rx代碼也可能是調(diào)試的噩夢,特別是如果涉及多個線程,甚至更糟 – 如果需要背壓.

如果你想進(jìn)入它,官方網(wǎng)站上有各種各樣的教程page,加上官方的documentation和Javadoc.你還可以看看一些視頻,如this one,它簡要介紹了Rx以及談?wù)揜x和期貨之間的差異.

獎勵：Java 9 Reactive Streams

Java 9’s Reactive Streams又名Flow API是由各種reactive streams庫(例如RxJava 2,Akka Streams和Vertx)實現(xiàn)的一組接口.它們允許這些反應(yīng)庫互連,同時保留所有重要的背壓.