Java 8原生API也能够开发响应式代码？

前段时间工做上比较忙，这篇文章一直没来得及写，本文是阅读《Java8实战》的时候，了解到Java 8里已经提供了一个异步非阻塞的接口（CompletableFuture），能够实现简单的响应式编程的模式，所以用这篇文章作个梳理。我是带着下面这几个问题去学习CompletableFuture这个接口的，

1. CompletableFuture是为了解决什么问题而设计的？
2. 它的使用场景是什么？开源软件中有实战使用案例吗？
3. CompletableFuture的经常使用API都有哪些？如何使用？
4. CompletableFuture和RxJava有什么不一样？

这篇文章梳理下来，基本上能够回答前面四个问题，OK，咱们进入正文。

基本概念

RPC（远程方法调用）的四种方式有：oneway、sync、future和callback，在dubbo或bolt这类通讯框架中，默认使用的是sync模式（同步+阻塞），future和callback都属于异步模式，不过future模式在get的时候会阻塞，callback模式则不须要等待结果，有结果后服务端会回调请求方。

异步调用这类模式，比较适合的场景是IO密集型场景，要执行不少远程调用的任务，而且这些调用耗时可能比较久。以openwrite中的一个case为例：我发布一篇文章，须要给几个不一样的写做平台建立文章，这时候我不但愿这个过程是顺序的，就比较适合用异步调用模式。

Future模式除了在get()调用的时候会阻塞外，还有其余的局限性，例如：没有使用Java Lambda表达式的优点，对一连串的异步调用能够支持，可是写出来的代码会比较复杂。

CompletableFuture的经常使用API

阅读CompletableFuture的API的时候，我有一个体会——CompletableFuture之于Future，除了增长了回调这个最重要的特性，其余的特性有点像Stream对于集合迭代的加强。

使用CompletableFuture，咱们能够像Stream同样使用一部调用，能够处理一些级联的异步调用（相似于Stream里的flatMap）、能够过滤一些无用的异步调用（anyOf、allOf）。

下面这张图是我按照本身的理解，梳理除了CompletableFuture常见的API，阅读的时候须要注意下面几个点：

1. 把握几个大的分类：建立CompletableFuture、获取CompletableFuture的执行结果、主动结束CompletableFuture、异步调用任务的组合处理；
2. 看着方法多，可是有规律可循，例如apply字样的接口，传入的方法参数都是有返回值的；
3. 带either字样的，都是多个异步任务有一个知足条件便可的；
4. 带executor方法的，都表示该方法能够用自定义的线程池来优化性能。

Dubbo项目中的使用案例

Dubbo对于异步化的支持起始在2.6.x中就有提供，是在发布bean的时候加个属性配置——async=true，而后利用上下文将异步标识一层层传递下去。在以前的公司中有一次排查dubbo（当时咱们用的是dubbox）异步调用的问题，最后查到的缘由就是多个异步调用，上下文里的信息串了。

Dubbo 2.7 中使用了 JDK1.8 提供的 CompletableFuture 原生接口对自身的异步化作了改进。CompletableFuture 能够支持 future 和 callback 两种调用方式。在Dubbo最新的master代码中，我知道了Dubbo的异步结果的定义，它的类图以下，能够看出AsyncRpcResult是一个CompletableFuture接口的实现。

实战Demo

经过下面的例子，能够看出CompletableFuture的最大好处——callback特性。首先定义一个接口，其中包括同步接口和该接口的异步版本。

public interface AsyncInterfaceExample {

    String computeSomeThine();

    CompletableFuture<String> computeSomeThingAsync();
}
复制代码

而后定义该接口的实现类，能够看出，若是要讲现有的同步接口异步化，是比较容易的；

public class AsyncInterfaceExampleImpl implements AsyncInterfaceExample {

    @Override
    public String computeSomeThine() {
        try {
            Thread.sleep(2000);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
        return "hello, world";
    }

    @Override
    public CompletableFuture<String> computeSomeThingAsync() {
        return CompletableFuture.supplyAsync(this::computeSomeThine);
    }
}
复制代码

而后看下咱们的测试case，以下：

public class AsyncInterfaceExampleTest {

    private static String getOtherThing() {
        try {
            Thread.sleep(2000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        return "other";
    }

    public static void main(String[] args) {
        AsyncInterfaceExample asyncInterfaceExample = new AsyncInterfaceExampleImpl();

        //case1 同步调用
        long start = System.currentTimeMillis();
        String someThing = asyncInterfaceExample.computeSomeThine();
        String other = getOtherThing();
        System.out.println("cost:" + (System.currentTimeMillis() - start) + " result:" + someThing + other);

        //case2 异步调用，使用回调
        start = System.currentTimeMillis();
        CompletableFuture<String> someThingFuture = asyncInterfaceExample.computeSomeThingAsync();
        other = getOtherThing();

        long finalStart = start;
        String finalOther = other;
        someThingFuture.whenComplete((returnValue, exception) -> {
            if (exception == null) {
                System.out.println(
                    "cost:" + (System.currentTimeMillis() - finalStart) + " result:" + returnValue + finalOther);
            } else {
                exception.printStackTrace();
            }
        });
    }
}
复制代码

上面这个案例的执行结果以下图所示：

*** 本号专一于后端技术、JVM问题排查和优化、Java面试题、我的成长和自我管理等主题，为读者提供一线开发者的工做和成长经验，期待你能在这里有所收获。