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

我想知道它们之间的区别
CompletableFuture,Future和Observable RxJava.

我所知道的都是异步但是

Future.get()阻塞线程

CompletableFuture提供了回调方法

RxJava Observable —与CompletableFuture类似,具有其他好处(不确定)

例如：如果客户端需要进行多次服务调用,当我们使用Futures(Java)时,Future.get()将按顺序执行…想知道它在RxJava中的效果如何…

文档http:///intro.html说

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

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

解决方法:

期货

在Java 5(2004)中引入了Futures.它们基本上是一个尚未完成的操作结果的占位符.一旦操作完成,Future将包含该结果.例如,操作可以是提交到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)中引入.它们实际上是常规期货的演变,受到谷歌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是在Netflix创建的reactive programming的完整库.一目了然,它看起来与Java 8’s streams相似.它是,除了它更强大.

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

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

RxJava的缺点是,尽管有可靠的文档,但由于涉及范式的转变,它是一个具有挑战性的库. Rx代码也可能是调试的噩梦,特别是如果涉及多个线程,甚至更糟 – 如果需要背压.

如果你想进入它,官方网站上有各种各样的教程page,加上官方的documentation和Javadoc.你还可以看看一些视频,如this one,它简要介绍了Rx以及谈论Rx和期货之间的差异.

奖励：Java 9 Reactive Streams

Java 9’s Reactive Streams又名Flow API是由各种reactive streams库(例如RxJava 2,Akka Streams和Vertx)实现的一组接口.它们允许这些反应库互连,同时保留所有重要的背压.

来源：https://www./content-1-476401.html