CompletableFuture、Future 和 RxJava 的 Observable 的区别

Difference between CompletableFuture, Future and RxJava's Observable

提问by shiv455

I would like to know the difference between CompletableFuture,Futureand ObservableRxJava.

我想知道CompletableFuture,Future和之间的区别 ObservableRxJava。

What I know is all are asynchronous but

我所知道的是都是异步的但是

Future.get()blocks the thread

Future.get()阻塞线程

CompletableFuturegives the callback methods

CompletableFuture给出回调方法

RxJava Observable--- similar to CompletableFuturewith other benefits(not sure)

RxJava Observable---CompletableFuture与其他好处类似（不确定）

For example: if client needs to make multiple service calls and when we use Futures(Java) Future.get()will be executed sequentially...would like to know how its better in RxJava..

例如：如果客户端需要进行多个服务调用，而我们使用时Futures（Java）Future.get()将按顺序执行……想知道它在 RxJava 中的表现如何。

And the documentation http://reactivex.io/intro.htmlsays

和文档http://reactivex.io/intro.html说

It is difficult to use Futures to optimally compose conditional asynchronous execution flows (or impossible, since latencies of each request vary at runtime). This can be done, of course, but it quickly becomes complicated (and thus error-prone) or it prematurely blocks on Future.get(), which eliminates the benefit of asynchronous execution.

使用 Futures 来优化组合条件异步执行流是很困难的（或者不可能，因为每个请求的延迟在运行时会有所不同）。这当然可以完成，但它很快变得复杂（因此容易出错）或者它过早地阻塞 Future.get()，这消除了异步执行的好处。

Really interested to know how RxJavasolves this problem. I found it difficult to understand from the documentation.

真的很想知道如何RxJava解决这个问题。我发现从文档中很难理解。

采纳答案by Malt

Futures

期货

Futureswere introduced in Java 5 (2004). They're basically placeholders for a result of an operation that hasn't finished yet. Once the operation finishes, the Futurewill contain that result. For example, an operation can be a Runnableor Callableinstance that is submitted to an ExecutorService. The submitter of the operation can use the Futureobject to check whether the operation isDone(), or wait for it to finish using the blocking get()method.

Futures是在 Java 5 (2004) 中引入的。它们基本上是尚未完成的操作结果的占位符。操作完成后，Future将包含该结果。例如，操作可以是提交给ExecutorService的Runnable或Callable实例。操作的提交者可以使用该对象来检查操作是否为isDone()，或者使用阻塞的get()方法等待它完成。Future

Example:

例子：

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

可完成期货

CompletableFutureswere introduced in Java 8 (2014). They are in fact an evolution of regular Futures, inspired by Google's Listenable Futures, part of the Guavalibrary. They are Futures that also allow you to string tasks together in a chain. You can use them to tell some worker thread to "go do some task X, and when you're done, go do this other thing using the result of X". Using CompletableFutures, you can do something with the result of the operation without actually blocking a thread to wait for the result. Here's a simple example:

CompletableFutures是在 Java 8 (2014) 中引入的。它们实际上是常规 Futures 的演变，受到谷歌的Listenable Futures 的启发，这是Guava库的一部分。它们是 Futures，还允许您将任务串成一个链。你可以使用它们告诉某个工作线程“去做一些任务 X，当你完成后，使用 X 的结果去做另一件事”。使用 CompletableFutures，您可以对操作的结果执行某些操作，而无需实际阻塞线程以等待结果。这是一个简单的例子：

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

RxJavais whole library for reactive programmingcreated at Netflix. At a glance, it will appear to be similar to Java 8's streams. It is, except it's much more powerful.

RxJava是Netflix 创建的用于响应式编程的完整库。乍一看，它看起来类似于Java 8 的流。它是，除了它更强大。

Similarly to Futures, RxJava can be used to string together a bunch of synchronous or asynchronous actions to create a processing pipeline. Unlike Futures, which are single-use, RxJava works on streamsof zero or more items. Including never-ending streams with an infinite number of items. It's also much more flexible and powerful thanks to an unbelievably rich set of operators.

与 Futures 类似，RxJava 可用于将一堆同步或异步操作串在一起以创建处理管道。与一次性使用的 Futures 不同，RxJava 可以处理零个或多个项目的流。包括具有无限数量项目的永无止境的流。由于一组令人难以置信的丰富的运算符，它也更加灵活和强大。

Unlike Java 8's streams, RxJava also has a backpressuremechanism, which allows it to handle cases in which different parts of your processing pipeline operate in different threads, at different rates.

与 Java 8 的流不同，RxJava 还具有背压机制，允许它处理处理管道的不同部分在不同线程中以不同速率运行的情况。

The downside of RxJava is that despite the solid documentation, it is a challenging library to learn due to the paradigm shift involved. Rx code can also be a nightmare to debug, especially if multiple threads are involved, and even worse - if backpressure is needed.

RxJava 的缺点是，尽管有可靠的文档，但由于所涉及的范式转换，它是一个具有挑战性的库。Rx 代码也可能是调试的噩梦，尤其是在涉及多个线程的情况下，更糟糕的是 - 如果需要背压。

If you want to get into it, there's a whole pageof various tutorials on the official website, plus the official documentationand Javadoc. You can also take a look at some of the videos such as this onewhich gives a brief intro into Rx and also talks about the differences between Rx and Futures.

如果你想深入了解，官网上有一整页各种教程，还有官方文档和Javadoc。您还可以查看一些视频，例如这个视频，它简要介绍了 Rx，并讨论了 Rx 和 Futures 之间的区别。

Bonus: Java 9 Reactive Streams

奖励：Java 9 反应式流

Java 9's Reactive Streamsaka Flow APIare a set of Interfaces implemented by various reactive streamslibraries such as RxJava 2, Akka Streams, and Vertx. They allow these reactive libraries to interconnect, while preserving the all important back-pressure.

Java 9 的 Reactive Streamsaka Flow API是一组由各种反应流库（例如RxJava 2、Akka Streams和Vertx ）实现的接口。它们允许这些反应式库相互连接，同时保留所有重要的背压。

回答by Kristoffer

I have been working with Rx Java since 0.9, now at 1.3.2 and soon migrating to 2.x I use this in a private project where I already work on for 8 years.

我从 0.9 开始使用 Rx Java，现在是 1.3.2 并且很快迁移到 2.x 我在一个我已经工作了 8 年的私人项目中使用它。

I wouldn't program without this library at all anymore. In the beginning I was skeptic but it is a complete other state of mind you need to create. Quiete difficult in the beginning. I sometimes was looking at the marbles for hours.. lol

如果没有这个库，我就不会再编程了。一开始我持怀疑态度，但这是你需要创造的一种完全不同的心态。刚开始时安静困难。我有时会盯着弹珠看好几个小时……哈哈

It is just a matter of practice and really getting to know the flow (aka contract of observables and observer), once you get there, you'll hate to do it otherwise.

这只是一个练习和真正了解流程的问题（又名可观察者和观察者的契约），一旦你到达那里，你就会讨厌这样做。

For me there is not really a downside on that library.

对我来说，那个库并没有真正的缺点。

Use case: I have a monitor view that contains 9 gauges (cpu, mem, network, etc...). When starting up the view, the view subscribes itselfs to a system monitor class that returns an observable (interval) that contains all the data for the 9 meters. It will push each second a new result to the view (so not polling !!!). That observable uses a flatmap to simultaneously (async!) fetch data from 9 different sources and zips the result into a new model your view will get on the onNext().

用例：我有一个包含 9 个仪表（CPU、内存、网络等...）的监视器视图。启动视图时，视图将自己订阅到一个系统监视器类，该类返回一个包含 9 米的所有数据的可观察值（间隔）。它将每秒向视图推送一个新结果（因此不轮询！！！）。该 observable 使用平面图同时（异步！）从 9 个不同来源获取数据，并将结果压缩到一个新模型中，您的视图将在 onNext() 上获得。

How the hell you gonna do that with futures, completables etc ... Good luck ! :)

你怎么会用期货、可完成物等来做这件事……祝你好运！:)

Rx Java solves many issues in programming for me and makes in a way a lot easier...

Rx Java 为我解决了许多编程问题，并在某种程度上让我变得更容易......

Advantages:

好处：

• Statelss !!! (important thing to mention, most important maybe)
• Thread management out of the box
• Build sequences that have their own lifecycle
• Everything are observables so chaining is easy
• Less code to write
• Single jar on classpath (very lightweight)
• Highly concurrent
• No callback hell anymore
• Subscriber based (tight contract between consumer and producer)
• Backpressure strategies (circuit breaker a like)
• Splendid error handling and recovering
• Very nice documentation (marbles <3)
• Complete control
• Many more ...

• 无国界！！！（重要的事情要提，最重要的也许是）
• 开箱即用的线程管理
• 构建具有自己生命周期的序列
• 一切都是可观察的，所以链接很容易
• 编写更少的代码
• 类路径上的单个 jar（非常轻量级）
• 高并发
• 不再有回调地狱
• 基于订阅者（消费者和生产者之间的紧密契约）
• 背压策略（断路器之类）
• 出色的错误处理和恢复
• 非常好的文档（大理石 <3）
• 完全控制
• 还有很多 ...

Disadvantages: - Hard to test

缺点： - 难以测试