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

我所知道的都是異步但是

Future.get()阻塞線程

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

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

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

文檔http:///intro.html說

很難使用Futures來(lái)優(yōu)化組合條件異步執(zhí)行流程(或者不可能,因?yàn)槊總€(gè)請(qǐng)求的延遲在運(yùn)行時(shí)會(huì)有所不同).當(dāng)然,這可以完成,但它很快變得復(fù)雜(因此容易出錯(cuò))或者過早地阻塞Future.get(),這消除了異步執(zhí)行的好處.

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

解決方法:

期貨

在Java 5(2004)中引入了Futures.它們基本上是一個(gè)尚未完成的操作結(jié)果的占位符.一旦操作完成,Future將包含該結(jié)果.例如,操作可以是提交到ExecutorService的Runnable或Callable實(shí)例.操作的提交者可以使用Future對(duì)象檢查操作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)中引入.它們實(shí)際上是常規(guī)期貨的演變,受到谷歌Listenable Futures(Guava庫(kù)的一部分)的啟發(fā).它們是Futures,它還允許您將任務(wù)串聯(lián)起來(lái).您可以使用它們來(lái)告訴某些工作線程“去做一些任務(wù)X,當(dāng)你完成后,使用X的結(jié)果去做其他事情”.使用CompletableFutures,您可以對(duì)操作結(jié)果執(zhí)行某些操作,而不會(huì)實(shí)際阻止線程等待結(jié)果.這是一個(gè)簡(jiǎn)單的例子：

/**
* 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的完整庫(kù).一目了然,它看起來(lái)與Java 8’s streams相似.它是,除了它更強(qiáng)大.

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

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

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

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

獎(jiǎng)勵(lì)：Java 9 Reactive Streams

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