CompletableFuture，Future和RxJava的Observable之间的区别

Question

我想知道的区别 CompletableFuture，Future和Observable RxJava。

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

Future.get() 阻塞线程

CompletableFuture 提供回调方法

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

例如：如果客户端需要进行多个服务调用，并且当我们使用Futures（Java）时Future.get()将依次执行…希望了解其在RxJava中的效果如何。

并且文档http://reactivex.io/intro.html说

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

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

Answer 1

期货

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

例：

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

可完成的未来

Java
8（2014）中引入了CompletableFutures。实际上，它们是常规期货的演变，受到了Guava库一部分的Google的Listenable
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是在Netflix创建的用于反应式编程的完整库。乍一看，它看起来类似于Java
8的stream。是的，只是功能更强大。

与Futures相似，RxJava可用于将一系列同步或异步操作串在一起以创建处理管道。与一次性使用的Futures不同，RxJava在零个或多个项目 流
上工作。包括无数个项目的无休止的流。多亏了令人难以置信的丰富操作员集合，它也更加灵活和强大。

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

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

如果您想了解它，则可以在官方网站上找到整页的各种教程，以及官方文档和Javadoc。您还可以看一下其中的一些视频，例如这段视频，简要介绍了Rx，还讨论了Rx和Futures之间的区别。

奖金：Java 9反应流

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