前言📫 作者简介：小明java问道之路，专注于研究计算机底层，就职于金融公司后端高级工程师，擅长交易领域的高安全/可用/并发/性能的设计和架构📫 

🏆 Java领域优质创作者、阿里云专家博主、华为云专家🏆

🔥 如果此文还不错的话，还请👍关注、点赞、收藏三连支持👍一下博主哦

本文导读

Java代码为了更好的发展和性能，开发了 异步编程的模式，Future异步编程和CompletableFuture 接口都可以实现异步编程，我们通过源码深入理解其原理和设计的思想，Java9中提供了反应式编程（Flow API）我们分析其源码并提供一个响应式查询实战。

一、同步与异步

1、为什么要有异步

在Java发展的这20年，他只做了一件事不被淘汰，为了不被淘汰不断的更新jdk的版本，以便使用计算机硬件、操作系统以及新的编程概念。

Java一开始提供了 synchronized 锁、Runable，后面java5有引入了 java.util.concurrent 包，java7中的 forkjoin 框架 java.util.concurrent.RecursiveTask，到java8中Stream流、lambda表达式的支持，这一切都是为了支持高并发。

即便如此，多线程虽然极大的提升了性能，如果合理的使用线程池的话，好处，第一可以降低资源消耗，重复利用已创建的线程；第二：提高响应速度，任务可以不需要等到线程创建就能立即执行；第三：提高线程的可管理性。统一分配、调优和监控。但是线程池也不是没有缺点，使用k个线程的线程池就只能并发的执行k个任务，其他任务还是回休眠或者阻塞。

这时候如果有线程不和其他任务相关联，又可以不用阻塞，就好了。Java8考虑到了，充分发挥了计算机硬件的处理能力，异步API 应运而生。

2、什么是同步？什么是异步？

同步就是 a 程序强依赖 b 程序，我必须等到你的回复或者执行完毕，才能做出下一步响应，类似于编程中程序被解释器（JVM）顺序执行一样（加载 > 验证 > 准备 > 解析 > 初始化）；

异步则相反，a 程序不强依赖 b 程序，响应的时间也无所谓，无论你返回还是不返回，a 程序都能继续运行，也就是说我不存在等待对方的概念，a 程序就是 异步非阻塞的。

下面举一个例子就说明什么是同步、什么是异步

异步编程涉及两种风格，Future风格API 和反应式风格API ，Future<Integer> fun(int a){}，fun( a , x-> {})，这两个模式的实战会在后面小结讲解。

​二、Future异步编程

1、Future 接口

Java5中就引入了Future 接口，他的涉及初衷就是异步计算，例如我们结算一个商户下的所有订单，这个时候并不需要for循环去累加，Future 接口使用的时候只需要封装 Callable中，再提交给ExecutorService。

2、Future 接口的使用

Future 接口的使用看下Java8之前是如何使用异步的

public static void main(String[] args) throws Exception {
    List<OrderInfo> orderInfos = Arrays.asList(new OrderInfo("123", BigDecimal.ONE),
            new OrderInfo("456", BigDecimal.TEN), new OrderInfo("789", BigDecimal.TEN));

    // 创建 ExecutorService 通过它可以向线程池提交任务
    ExecutorService executorService = Executors.newCachedThreadPool();

    // 异步操作的同时，可以进行其他操作
    Future<BigDecimal> decimalFuture = executorService.submit(new Callable<BigDecimal>() {
        @Override
        public BigDecimal call() throws Exception {
            return reduceAmt(orderInfos);
        }
    });

    // Java8 写法
    Future<BigDecimal> decimalFuture = executorService.submit(() -> reduceAmt(orderInfos));

    System.out.println(decimalFuture.get());
}

private static BigDecimal reduceAmt(List<OrderInfo> orderInfos) {
    return orderInfos.stream()
            .map(OrderInfo::getOrderAmt)
            .reduce(BigDecimal.ZERO, BigDecimal::add);
}

异步编程可以在 ExecutorService 中，以并发的方式调用另一个线程执行操作，后续调用get() 方法获取操作结果，如果操作完成会立刻返回，如果操作没有完成则回阻塞线程，直到操作完成返回。

3、Future 接口的缺陷（局限性）

Future接口 还提供了方法来检测异步计算是否已经结束（isDone() 方法），等待异步操作结束。

但是当长时间计算任务完成时，该计算的结果通知到另一个长时间运行的计算任务，这两个计算任务都完成后，将计算的结果与另一个查询操作结果合并。

就会发生很多性能问题：1、将两个异步计算合并为一个（这两个异步计算之间相互独立，同时第二个又依赖于第一个的结果。）2、此时就要等待 Future 集合中的所有任务都完成。3、当Future的完成事件发生时会收到通知，使用Future计算的结果进行下一步的操作，不只是简单地阻塞等待操作的结果，每一步都需要等待。

三、CompletableFuture 接口详解

1、CompletableFuture 的创建

CompletableFuture.runAsync()也可以用来创建CompletableFuture实例。与supplyAsync()不同的是，runAsync()传入的任务要求是Runnable类型的，所以没有返回值，runAsync适合创建不需要返回值的计算任务

通过该supplyAsync 函数创建的CompletableFuture实例会异步执行当前传入的计算任务，在调用端，则可以通过get或join获取最终计算结果。

同事直接new（构造函数创建）也是可以的，下面通过一个实战小例子看下，高并发高性能的添加购物车结构如何设计

import java.util.concurrent.CompletableFuture;
import org.springframework.core.task.AsyncTaskExecutor;

public class CompletableFutureImpl {
    @Autowired
    @Qualifier("asyncExecutor")
    private AsyncTaskExecutor asyncTaskExecutor;

    public DefaultResponseVO addGoodsCart(HttpServletRequest request, @Valid AddCartReqVO reqVo) {
        // 添加购物车
        GetCartItemEntity respVO = mallCartProcess.addGoodsCart(buildAddGoodsCartReqVO(reqBo));

        /**
         * 异步刷新到购物车
         */
        CompletableFuture voidCompletableFuture = CompletableFuture.runAsync(() -> syncAddCacheCheck(respVO), asyncTaskExecutor);
        // 通过supplyAsync 函数创建的
        CompletableFuture<Object> uCompletableFuture = CompletableFuture.supplyAsync(() -> syncAddCacheCheck(respVO));
        // 构造函数创建 
        CompletableFuture completableFuture = new CompletableFuture().runAsync(() -> syncAddCacheCheck(respVO));

        return new DefaultResponseVO(code, msg, respVO);
    }

    /**
     * 添加购物车缓存(异步刷新到购物车)
     */
    public void syncAddCacheCheck(GetCartItemEntity cartItemEntity) {
        try {
            // 添加购物车缓存(异步刷新到购物车)
            mallCartProcess.getCartInfo(cartItemEntity);
        } catch (Exception e) {
            logger.error("syncAddCache error", e);
        }
    }
}

​​2、CompletableFuture.supplyAsync 源码分析

本小节讲解CompletableFuture的底层实现

上面为java 8中 supplyAsync 函数的实现源码。可以看到，当 supplyAsync 入参只有 supplier 时，会默认使用asyncPool作为线程池（一般情况下为ForkJoinPool的commonPool），并调用内部方法asyncSupplyStage执行具体的逻辑。在 asyncSupplyStage 方法中，程序会创建一个空的CompletableFuture 返回给调用方。同时该 CompletableFuture 和传入的 supplier 会被包装在一个AsyncSupply 实例对象中，然后一起提交到线程池中进行处理。

值得注意的是，当supplyAsync返回时，调用方只会拿到一个空的CompletableFuture实例。看到这里，我们可以猜测，当计算最终完成时，计算结果会被set到对应的CompletableFuture的result字段中。调用方通过join或者get就能取到该CompletableFuture的result字段的值。所以，虽然实际创建CompletableFuture的线程和进行任务计算的线程不同，但是最终会通过result来进行结果的传递。这种方式与传统的Future中结果传递方式类似（计算线程set值，使用线程get值）。

​

上面为java 8中 AsyncSupply 的实现源码，AsyncSupply的源码很简单。首先，它实现了Runnable接口，所以被提交到线程池中后，工作线程会执行其run()方法。通过对AsyncSupply中run方法的分析，也基本证实我们之前的猜测。即计算任务由工作线程调用run方法执行，并设置到CompletableFuture的结果中。其他线程中的使用方，则可以调用该CompletableFuture的join或者get方法获取其结果。

因此，我们只需要搞清楚其run()中的实现即可。在 run() 中，程序首先检查了传入的CompletableFuture 和 Supplier 是否为空，如果均不为空，再检查 CompletableFuture 的 d.result是否为空，如果不为空，则说明 CompletableFuture 的值已经被其他线程主动设置过了（这也是CompletableFuture与Future最大的不同之处），因此这里就不会再被重新设置一次。如果 d.result 为空，则调用Supplier（源码中的 f 变量）的get()方法，执行具体的计算，然后通过 completeValue 方法将结果设置到CompletableFuture中。最后，调用CompletableFuture的postComplete()方法，执行连接到当前CompletableFuture上的后置任务。

3、CompletableFuture.runAsync 源码分析

通过上面的源码可以看出，runAsync也会生成一个空的CompletableFuture，并包装在AsyncRun中提交到线程池中执行。这与supplyAsync是完全一致的。由于Runnable没有返回值，这里返回的CompletableFuture的结果值是Void类型的。

AsyncRun的run中，计算的执行是通过调用传入的Runnable（源码中的 f 变量）的run方法进行的。由于没有返回值，所以这里在设置CompletableFuture的值时，使用其completeNull()方法，设置一个特殊的空值标记。 设计方面和supplyAsync一致

4、CompletableFuture API 实战

thenApply、thenAccept、thenRun

thenApply 提交的任务类型需遵从Function签名，也就是有入参和返回值，其中入参为前置任务的结果。thenAccept 提交的任务类型需遵从Consumer签名，也就是有入参但是没有返回值，其中入参为前置任务的结果。thenRun 提交的任务类型需遵从Runnable签名，即没有入参也没有返回值。

thenCombine、thenCompose

thenCombine最大的不同是连接任务可以是一个独立的CompletableFuture。嵌套获取层级也越来越深。因此，需要一种方式，能将这种嵌套模式展开，使其没有那么多层级。thenCompose的主要目的就是解决这个问题（这里也可以将thenCompose的作用类比于stream接口中的flatMap，因为他们都可以将类型嵌套展开）。

whenComplete、handle

whenComplete主要用于注入任务完成时的回调通知逻辑（获得的结果是前置任务的结果，whenComplete中的逻辑不会影响计算结果）。handle与handle接收的处理函数有返回值，而且返回值会影响最终获取的计算结果（产生了新的结果）

四、反应式编程

1、什么是反应式（resctive）编程

反应式编程是最近几年才提出的概念，主要有四个特征：响应式，反应式编程的响应速度应该很快，而且是稳定可预测的。韧性，系统出现失败时，任然可以继续响应服务，任何一个组件都能以异步的方式想其他组件分发任务。弹性，影响代码响应的因素的代码（系统）的负载能力，反应式编程可以增加分配的资源，受流量影响后有自动适配的能力，服务更大的负载。消息驱动，各个组件之间松耦合，组件隔离，跨组件通信使用异步消息传递。

反应式（resctive）编程在应用层的主要特征是任务以异步的方式执行，非阻塞的处理事件流，充分利用多核CPU的特点，大多反应式框架（RxJava等）都可以独立开辟线程池，用于执行阻塞式操作，主线程池中运行都是无阻塞的。

2、反应式流（Flow API）

2.1、发布订阅模式

Future 和 CompletableFuture 的思维模式是计算的执行是独立且并发的。使用 get()方法可以在执行结束后获取 Future 对象的执行结果。因此，Future 是一个一次性对象，它只能从头到尾执行代码一次。

与此相反，反应式编程的思维模式是类 Future 的对象随着时间的推移可以产生很多的结果。举个例子是 Web 服务器的监听组件对象。该组件监听来自网络的 HTTP请求，并根据请求的内容返回相应的数据。

​

2.2、Flow API 源码解析

Java9 使用 java.util.concurrent.Flow 提供的接口对反应式编程进行建模，实现了名为“发布-订阅”的模型（也叫协议，简写为 pub-sub ）

反应式编程有三个主要的概念，分别是：订阅者可以订阅的发布者；名为订阅的连接；消息(也叫事件)，它们通过连接传输。

反应式流（Flow API）规范可以总结为4个接口：Publisher（发布者）、Subscriber（订阅者）、Subscription（订阅）和Processor（处理者）

Publisher负责生成数据，并将数据发送给 Subscription（每个Subscriber对应一个Subscription）。Publisher接口声明了一个方法 subscribe()，Subscriber可以通过该方法向 Publisher发起订阅。 

一旦Subscriber订阅成功，就可以接收来自Publisher的事件。这些事件是通过Subscriber接口上的方法发送的

Subscriber的第一个事件是通过对 onSubscribe()方法的调用接收的。Publisher调用 onSubscribe() 方法时，会将Subscription对象传递给 Subscriber。通过Subscription，Subscriber可以管理其订阅情况

Subscriber 可以通过调用 request() 方法来请求 Publisher 发送数据，或者通过调用 cancel()方法表明它不再对数据感兴趣并且取消订阅。当调用 request() 时，Subscriber 可以传入一个long类型的数值以表明它愿意接受多少数据。这也是回压能够发挥作用的地方，以避免Publisher 发送多于 Subscriber能够处理的数据量。在 Publisher 发送完所请求数量的数据项之后，Subscriber 可以再次调用 request()方法来请求更多的数据。

Subscriber 请求数据之后，数据就会开始流经反应式流。Publisher 发布的每个数据项都会通过调用 Subscriber 的 onNext()方法递交给 Subscriber。如果有任何错误，就会调用 onError()方法。如果 Publisher 没有更多的数据，也不会继续产生更多的数据，那么将会调用 Subscriber 的onComplete() 方法来告知 Subscriber 它已经结束

反应式流规范的接口本身并不支持以函数式的方式组成这样的流。Reactor 项目是反应式流规范的一个实现，提供了一组用于组装反应式流的函数式API。有我们自己实现。

​2.3、Flow API 实战

FlowImpl ：创建Publisher并向其订阅TempSubscriber

public class FlowImpl {
    public static void main(String[] args) {
        // 创建 Publisher 并向其订阅 Subscriber
        getOrderAmt("20220727123").subscribe(new SubscriberImpl());
    }

    // Publisher是个函数式接口
    private static Flow.Publisher<OrderInfo> getOrderAmt(String orderId) {
        return subscriber -> subscriber.onSubscribe(new SubscriptionImpl(subscriber, orderId));
    }
}

Subscription接口：实现向 Subscriber 发送 OrderInfo Steam

public class SubscriptionImpl implements Flow.Subscription {
    private final Flow.Subscriber<? super OrderInfo> subscriber;
    private final String orderId;

    public SubscriptionImpl(Flow.Subscriber<? super OrderInfo> subscriber, String orderId) {
        this.subscriber = subscriber;
        this.orderId = orderId;
    }

    @Override
    public void request(long n) {
        // 另起一个线程向 subscriber 发送下一个元素
        Executors.newSingleThreadExecutor().submit(() -> {
            for (long i = 0L; i < n; i++) // subscriber 每处理一个请求执行一次循环
                try {
                    // 将当前 订单号 发送给 Subscriber
                    subscriber.onNext(OrderInfo.reduceAmt(orderId));
                } catch (Exception e) {
                    // 查询报错将这个报错信息传给 Subscriber
                    subscriber.onError(e);
                    e.printStackTrace();
                    break;
                }
        });
    }

    @Override
    public void cancel() {
        // 如果 Subscription 被取消了，向 subscriber 发送一个完成信号
        subscriber.onComplete();
    }
}

​

Subscriber接口：实现打印输出收到的 订单 数据

public class SubscriberImpl implements Flow.Subscriber<OrderInfo> {
    private Flow.Subscription subscription;
    @Override
    public void onSubscribe(Flow.Subscription subscription) {
        this.subscription = subscription;
        subscription.request(1);
    }
    @Override
    public void onNext(OrderInfo orderInfo) {
        System.out.println(orderInfo);
        subscription.request(1);
    }
    @Override
    public void onError(Throwable throwable) {
        System.out.println(throwable.getMessage());
    }
    @Override
    public void onComplete() {
        System.out.println("Done!");
    }
}

总结

Java代码为了更好的发展和性能，开发了 异步编程的模式，Future异步编程和CompletableFuture 接口都可以实现异步编程，我们通过源码深入理解其原理和设计的思想，Java9中提供了反应式编程（Flow API）我们分析其源码并提供一个响应式查询实战。