JAVA基于CompletableFuture的流水线并行处理深度实践，满满干货

大家好，又见面啦。

在项目开发中，后端服务对外供给API接口一般都会重视呼应时长。可是某些状况下，由于事务规划逻辑的原因，咱们的接口或许会是一个聚合信息处理类的处理逻辑，比方咱们从多个不同的当地获取数据，然后汇总处理为终究的成果再回来给调用方，这种状况下，往往会导致咱们的接口呼应特别的慢。

而假如咱们想要动手进行优化的时分呢，就会触及到串行处理改并行处理的问题。在JAVA中并行处理的才能支撑现已相对完善，经过对CompletableFuture的合理运用，能够让咱们面临这种聚合类处理的场景会愈加的得心应手。

好啦，话不多说，接下来就让咱们一同来品味下JAVA中组合式并行处理这道饕餮大餐吧。

前菜：先看个实践场景

在开端享受这顿大餐前，咱们先来个前菜开开胃。

例如现在有这么个需求：

需求描绘： 完结一个全网比价服务，比方能够从某宝、某东、某夕夕去获取某个产品的价格、优惠金额，并核算出实践付款金额，终究回来价格最优的渠道与价格信息。

这儿假定每个渠道获取原价格与优惠券的接口现已完结、且都是需求调用HTTP接口查询的耗时操作，Mock接口每个耗时1s左右。

依据开端的需求了解，咱们能够很自然的写出对应完结代码：

public PriceResult getCheapestPlatAndPrice(String product) {
    PriceResult mouBaoPrice = computeRealPrice(HttpRequestMock.getMouBaoPrice(product),            HttpRequestMock.getMouBaoDiscounts(product));
    PriceResult mouDongPrice = computeRealPrice(HttpRequestMock.getMouDongPrice(product),            HttpRequestMock.getMouDongDiscounts(product));
    PriceResult mouXiXiPrice = computeRealPrice(HttpRequestMock.getMouXiXiPrice(product),            HttpRequestMock.getMouXiXiDiscounts(product));
    // 核算并选出实践价格最低的渠道
    return Stream.of(mouBaoPrice, mouDongPrice, mouXiXiPrice).            min(Comparator.comparingInt(PriceResult::getRealPrice))            .get();
}

一切顺利成章，运转测试下：

05:24:54.779[main|1]获取某宝上 Iphone13的价格完结： 5199
05:24:55.781[main|1]获取某宝上 Iphone13的优惠完结： -200
05:24:55.781[main|1]某宝终究价格核算完结：4999
05:24:56.784[main|1]获取某东上 Iphone13的价格完结： 5299
05:24:57.786[main|1]获取某东上 Iphone13的优惠完结： -150
05:24:57.786[main|1]某东终究价格核算完结：5149
05:24:58.788[main|1]获取某夕夕上 Iphone13的价格完结： 5399
05:24:59.791[main|1]获取某夕夕上 Iphone13的优惠完结： -5300
05:24:59.791[main|1]某夕夕终究价格核算完结：99
获取最优价格信息：【渠道：某夕夕, 原价：5399, 扣头：0, 实付价：99】
-----履行耗时： 6122ms  ------

成果契合预期，功用一切正常，便是耗时长了点。试想一下，假如你在某个APP操作查询的时分，等候6s才回来成果，估计会直接把APP给卸载了吧？

整理下前面代码的完结思路：

一切的环节都是串行的，每个环节耗时加到一同，接口总耗时肯定很长。

但实践上，每个渠道之间的操作是互不搅扰的，那咱们自然而然的能够想到，能够经过多线程的办法，一同去别离履行各个渠道的逻辑处理，终究将各个渠道的成果汇总到一同比对得到最低价格。

所以整个履行进程会变成如下的作用：

为了提高功用，咱们采用线程池来负责多线程的处理操作，由于咱们需求得到各个子线程处理的成果，所以咱们需求运用 Future来完结：

public PriceResult getCheapestPlatAndPrice2(String product) {
    Future<PriceResult> mouBaoFuture = threadPool.submit(() -> computeRealPrice(HttpRequestMock.getMouBaoPrice(product), HttpRequestMock.getMouBaoDiscounts(product)));
    Future<PriceResult> mouDongFuture = threadPool.submit(() -> computeRealPrice(HttpRequestMock.getMouDongPrice(product), HttpRequestMock.getMouDongDiscounts(product)));
    Future<PriceResult> mouXiXiFuture = threadPool.submit(() -> computeRealPrice(HttpRequestMock.getMouXiXiPrice(product), HttpRequestMock.getMouXiXiDiscounts(product)));
    // 等候一切线程成果都处理完结，然后从成果中核算出最低价
    return Stream.of(mouBaoFuture, mouDongFuture, mouXiXiFuture)
        .map(priceResultFuture -> {
            try {
                return priceResultFuture.get(5L, TimeUnit.SECONDS);
            } catch (Exception e) {
                return null;
            }
        })
        .filter(Objects::nonNull).min(Comparator.comparingInt(PriceResult::getRealPrice)).get();
}

上述代码中，将三个不同渠道对应的Callable函数逻辑放入到ThreadPool中去履行，回来Future目标，然后再逐一经过Future.get()接口堵塞获取各自渠道的成果，终究经比较处理后回来最低价信息。

履行代码，能够看到履行成果与进程如下：

05:42:25.291[pool-1-thread-2|13]获取某东上 Iphone13的价格完结： 5299
05:42:25.291[pool-1-thread-3|14]获取某夕夕上 Iphone13的价格完结： 5399
05:42:25.291[pool-1-thread-1|12]获取某宝上 Iphone13的价格完结： 5199
05:42:26.294[pool-1-thread-2|13]获取某东上 Iphone13的优惠完结： -150
05:42:26.294[pool-1-thread-3|14]获取某夕夕上 Iphone13的优惠完结： -5300
05:42:26.294[pool-1-thread-1|12]获取某宝上 Iphone13的优惠完结： -200
05:42:26.294[pool-1-thread-2|13]某东终究价格核算完结：5149
05:42:26.294[pool-1-thread-3|14]某夕夕终究价格核算完结：99
05:42:26.294[pool-1-thread-1|12]某宝终究价格核算完结：4999
获取最优价格信息：【渠道：某夕夕, 原价：5399, 扣头：0, 实付价：99】
-----履行耗时： 2119ms  ------

成果与第一种完结办法共同，可是接口总耗时从6s下降到了2s，作用仍是很明显的。可是，是否还能再压缩一些呢？

依据上面依照渠道拆分并行处理的思路持续推动，咱们能够看出每个渠道内的处理逻辑其实能够分为3个首要步骤：

1. 获取原始价格（耗时操作）
2. 获取扣头优惠（耗时操作）
3. 得到原始价格和扣头优惠之后，核算实付价格

这3个步骤中，第1、2两个耗时操作也是相对独立的，假如也能并行处理的话，呼应时长上应该又会缩短一些，即如下的处理流程：

咱们当然能够持续运用上面提到的线程池+Future的办法，但Future在应对并行成果组合以及后续处理等方面显得无能为力，弊端明显：

代码写起来会非常拖沓：先封装Callable函数放到线程池中去履行查询操作，然后分三组堵塞等候成果并核算出各自成果，终究再堵塞等候价格核算完结后汇总得到终究成果。

说到这儿呢，就需求咱们新的主人公CompletableFuture登场了，经过它咱们能够很轻松的来完结使命的并行处理，以及各个并行使命成果之间的组合再处理等操作。咱们运用CompletableFuture编写完结代码如下：

public PriceResult getCheapestPlatAndPrice3(String product) {
    CompletableFuture<PriceResult> mouBao = CompletableFuture.supplyAsync(() -> HttpRequestMock.getMouBaoPrice(product)).thenCombine(CompletableFuture.supplyAsync(() -> HttpRequestMock.getMouBaoDiscounts(product)), this::computeRealPrice);
    CompletableFuture<PriceResult> mouDong = CompletableFuture.supplyAsync(() -> HttpRequestMock.getMouDongPrice(product)).thenCombine(CompletableFuture.supplyAsync(() -> HttpRequestMock.getMouDongDiscounts(product)), this::computeRealPrice);
    CompletableFuture<PriceResult> mouXiXi = CompletableFuture.supplyAsync(() ->  HttpRequestMock.getMouXiXiPrice(product)).thenCombine(CompletableFuture.supplyAsync(() -> HttpRequestMock.getMouXiXiDiscounts(product)), this::computeRealPrice);
    // 排序并获取最低价格
    return Stream.of(mouBao, mouDong, mouXiXi)
            .map(CompletableFuture::join)
            .sorted(Comparator.comparingInt(PriceResult::getRealPrice))
            .findFirst()
            .get();
}

看下履行成果契合预期，而接口耗时则降到了1s（由于咱们依靠的每一个查询实践操作的接口耗时都是模拟的1s，所以这个成果现已算是此复合接口能到达的极限值了）。

06:01:13.354[ForkJoinPool.commonPool-worker-6|17]获取某夕夕上 Iphone13的优惠完结： -5300
06:01:13.354[ForkJoinPool.commonPool-worker-13|16]获取某夕夕上 Iphone13的价格完结： 5399
06:01:13.354[ForkJoinPool.commonPool-worker-4|15]获取某东上 Iphone13的优惠完结： -150
06:01:13.354[ForkJoinPool.commonPool-worker-9|12]获取某宝上 Iphone13的价格完结： 5199
06:01:13.354[ForkJoinPool.commonPool-worker-11|14]获取某东上 Iphone13的价格完结： 5299
06:01:13.354[ForkJoinPool.commonPool-worker-2|13]获取某宝上 Iphone13的优惠完结： -200
06:01:13.354[ForkJoinPool.commonPool-worker-13|16]某夕夕终究价格核算完结：99
06:01:13.354[ForkJoinPool.commonPool-worker-11|14]某东终究价格核算完结：5149
06:01:13.354[ForkJoinPool.commonPool-worker-2|13]某宝终究价格核算完结：4999
获取最优价格信息：【渠道：某夕夕, 原价：5399, 扣头：0, 实付价：99】
-----履行耗时： 1095ms  ------

好啦，经过餐前的前菜，大家应该能够看出来串行与并行处理逻辑的差异、以及并行处理逻辑的完结战略了吧？这儿咱们应该也能够看出CompletableFuture在应对并行处理场景下的强大优势。当然咯，上面也只是小小的窥探了下CompletableFuture功用的冰上一角，下面就让咱们一同来深化了解下，享受并消化CompletableFuture这道主菜吧！

主菜：CompletableFuture深化了解

好啦，下面该主菜上场了。

作为JAVA8之后加入的新成员，CompletableFuture的完结与运用上，也处处体现出了函数式异步编程的味道。一个CompletableFuture目标能够被一个环节接一个环节的处理、也能够对两个或许多个CompletableFuture进行组合处理或许等候成果完结。经过对CompletableFuture各种办法的合理运用与组合搭配，能够让咱们在许多的场景都能够应付自如。

下面就来一同了解下这些办法以及对应的运用办法吧。

Future与CompletableFuture

首先，先来理一下Future与CompletableFuture之间的联系。

Future

假如触摸过多线程相关的概念，那Future应该不会生疏，早在Java5中就现已存在了。

该怎样了解Future呢？举个生活中的比方：

你去咖啡店点了一杯咖啡，然后服务员会给你一个订单小票。 当服务员在后台制造咖啡的时分，你并没有在店里等候，而是出门到近邻甜品店又买了个面包。 当面包买好之后，你回到咖啡店，拿着订单小票去取咖啡。 取到咖啡后，你边喝咖啡边把面包吃了……嗝~

是不是很熟悉的生活场景？ 比照到咱们多线程异步编程的场景中，咖啡店的订单小票其实便是Future，经过Future能够让稍后恰当的时分能够获取到对应的异步履行线程中的履行成果。

上面的场景，咱们翻译为代码完结逻辑：

public void buyCoffeeAndOthers() throws ExecutionException, InterruptedException {
    goShopping();
    // 子线程中去处理做咖啡这件事，回来future目标
    Future<Coffee> coffeeTicket = threadPool.submit(this::makeCoffee);
    // 主线程同步去做其他的作业
    Bread bread = buySomeBread();
    // 主线程其他作业并行处理完结，堵塞等候获取子线程履行成果
    Coffee coffee = coffeeTicket.get();
    // 子线程成果获取完结，主线程持续履行
    eatAndDrink(bread, coffee);
}

编码源于生活、代码中的规划逻辑，许多时分都是与生活哲学匹配的。

CompletableFuture

Future在应对一些简略且彼此独立的异步履行场景很快捷，可是在一些杂乱的场景，比方一同需求多个有依靠联系的异步独立处理的时分，或许是一些相似流水线的异步处理场景时，就显得无能为力了。比方：

• 一同履行多个并行使命，等候最快的一个完结之后就能够持续往后处理
• 多个异步使命，每个异步使命都需求依靠前一个异步使命履行的成果再去履行下一个异步使命，终究只需求一个终究的成果
• 等候多个异步使命悉数履行完结后触发下一个动作履行
• …

所以呢， 在JAVA8开端引进了全新的CompletableFuture类，它是Future接口的一个完结类。也便是在Future接口的基础上，额外封装供给了一些履行办法，用来解决Future运用场景中的一些不足，对流水线处理才能供给了支撑。

下一节中，咱们就来进一步的了解下CompletableFuture的具体运用场景与运用办法。

CompletableFuture运用办法

创立CompletableFuture并履行

当咱们需求进行异步处理的时分，咱们能够经过CompletableFuture.supplyAsync办法，传入一个具体的要履行的处理逻辑函数，这样就轻松的完结了CompletableFuture的创立与触发履行。

办法称号	作用描绘
supplyAsync	静态办法，用于构建一个CompletableFuture<T>目标，并异步履行传入的函数，答应履行函数有回来值T。
runAsync	静态办法，用于构建一个CompletableFuture<Void>目标，并异步履行传入函数，与supplyAsync的差异在于此办法传入的是Callable类型，仅履行，没有回来值。

运用示例：

public void testCreateFuture(String product) {
    // supplyAsync， 履行逻辑有回来值PriceResult
    CompletableFuture<PriceResult> supplyAsyncResult =
            CompletableFuture.supplyAsync(() -> HttpRequestMock.getMouBaoPrice(product));
    // runAsync, 履行逻辑没有回来值
    CompletableFuture<Void> runAsyncResult =
            CompletableFuture.runAsync(() -> System.out.println(product));
}

特别弥补：

supplyAsync或许runAsync创立后便会立即履行，无需手动调用触发。

环环相扣处理

在流水线处理场景中，往往都是一个使命环节处理完结后，下一个使命环节接着上一环节处理成果持续处理。CompletableFuture用于这种流水线环节驱动类的办法有许多，彼此之间首要是在回来值或许给到下一环节的入参上有些许差异，运用时需求留意区分：

具体的办法的描绘归纳如下：

办法称号	作用描绘
thenApply	对CompletableFuture的履行后的具体成果进行追加处理，并将当时的CompletableFuture泛型目标更改为处理后新的目标类型，回来当时CompletableFuture目标。
thenCompose	与thenApply相似。差异点在于：此办法的入参函数回来一个CompletableFuture类型目标。
thenAccept	与thenApply办法相似，差异点在于thenAccept回来void类型，没有具体成果输出，适合无需回来值的场景。
thenRun	与thenAccept相似，差异点在于thenAccept能够将前面CompletableFuture履行的实践成果作为入参进行传入并运用，可是thenRun办法没有任何入参，只能履行一个Runnable函数，而且回来void类型。

由于上述thenApply、thenCompose办法的输出仍然都是一个CompletableFuture目标，所以各个办法是能够一环接一环的进行调用，构成流水线式的处理逻辑：

希望总是夸姣的，可是实践状况却总不尽善尽美。在咱们编列流水线的时分，假如某一个环节履行抛出反常了，会导致整个流水线后续的环节就没法再持续下去了，比方下面的比方：

public void testExceptionHandle() {
    CompletableFuture.supplyAsync(() -> {
        throw new RuntimeException("supplyAsync excetion occurred...");
    }).thenApply(obj -> {
        System.out.println("thenApply executed...");
        return obj;
    }).join();
}

履行之后会发现，supplyAsync抛出反常后，后边的thenApply并没有被履行。

那假如咱们想要让流水线的每个环节处理失利之后都能让流水线持续往下面环节处理，让后续环节能够拿到前面环节的成果或许是抛出的反常并进行对应的应对处理，就需求用到handle和whenCompletable办法了。

先看下两个办法的作用描绘：

办法称号	作用描绘
handle	与thenApply相似，差异点在于handle履行函数的入参有两个，一个是CompletableFuture履行的实践成果，一个是是Throwable目标，这样假如前面履行呈现反常的时分，能够经过handle获取到反常并进行处理。
whenComplete	与handle相似，差异点在于whenComplete履行后无回来值。

咱们对上面一段代码示例修改运用handle办法来处理：

public void testExceptionHandle() {
    CompletableFuture.supplyAsync(() -> {
        throw new RuntimeException("supplyAsync excetion occurred...");
    }).handle((obj, e) -> {
        if (e != null) {
            System.out.println("thenApply executed, exception occurred...");
        }
        return obj;
    }).join();
}

再履行能够发现，即便前面环节呈现反常，后边环节也能够持续处理，且能够拿到前一环节抛出的反常信息：

thenApply executed, exception occurred...

多个CompletableFuture组合操作

前面一直在介绍流水线式的处理场景，可是许多时分，流水线处理场景也不会是一个链路顺序往下走的状况，许多时分为了提高并行功率，一些没有依靠的环节咱们会让他们一同去履行，然后在某些环节需求依靠的时分，进行成果的依靠兼并处理，相似如下图的作用。

CompletableFuture比较于Future的一大优势，便是能够方便的完结多个并行环节的兼并处理。相关触及办法介绍归纳如下：

办法称号	作用描绘
thenCombine	将两个CompletableFuture目标组合起来进行下一步处理，能够拿到两个履行成果，并传给自己的履行函数进行下一步处理，终究回来一个新的CompletableFuture目标。
thenAcceptBoth	与thenCombine相似，差异点在于thenAcceptBoth传入的履行函数没有回来值，即thenAcceptBoth回来值为CompletableFuture<Void>。
runAfterBoth	等候两个CompletableFuture都履行完结后再履行某个Runnable目标，再履行下一个的逻辑，相似thenRun。
applyToEither	两个CompletableFuture中恣意一个完结的时分，持续履行后边给定的新的函数处理。再履行后边给定函数的逻辑，相似thenApply。
acceptEither	两个CompletableFuture中恣意一个完结的时分，持续履行后边给定的新的函数处理。再履行后边给定函数的逻辑，相似thenAccept。
runAfterEither	等候两个CompletableFuture中恣意一个履行完结后再履行某个Runnable目标，能够了解为thenRun的升级版，留意与runAfterBoth比照了解。
allOf	静态办法，堵塞等候一切给定的CompletableFuture履行结束后，回来一个CompletableFuture<Void>成果。
anyOf	静态办法，堵塞等候恣意一个给定的CompletableFuture目标履行结束后，回来一个CompletableFuture<Void>成果。

成果等候与获取

在履行线程中将使命放到作业线程中进行处理的时分，履行线程与作业线程之间是异步履行的形式，假如履行线程需求获取到共作业线程的履行成果，则能够经过get或许join办法，堵塞等候并从CompletableFuture中获取对应的值。

对get和join的办法功用含义阐明归纳如下：

办法称号	作用描绘
get()	等候CompletableFuture履行完结并获取其具体履行成果，或许会抛出反常，需求代码调用的当地手动try...catch进行处理。
get(long, TimeUnit)	与get()相同，只是答应设定堵塞等候超时时刻，假如等候超越设定时刻，则会抛出反常停止堵塞等候。
join()	等候CompletableFuture履行完结并获取其具体履行成果，或许会抛出运转时反常，无需代码调用的当地手动try…catch进行处理。

从介绍上能够看出，两者的差异就在于是否需求调用方显式的进行try…catch处理逻辑，运用代码示例如下：

public void testGetAndJoin(String product) {
    // join无需显式try...catch...
    PriceResult joinResult = CompletableFuture.supplyAsync(() -> HttpRequestMock.getMouXiXiPrice(product))
            .join();
    try {
        // get显式try...catch...
        PriceResult getResult = CompletableFuture.supplyAsync(() -> HttpRequestMock.getMouXiXiPrice(product))
                .get(5L, TimeUnit.SECONDS);
    } catch (Exception e) {
        e.printStackTrace();
    }
}

CompletableFuture办法及其Async版别

咱们在运用CompletableFuture的时分会发现，有许多的办法，都会一同有两个以Async命名结尾的办法版别。以前面咱们用的比较多的thenCombine办法为例：

1. thenCombine(CompletionStage, BiFunction)
2. thenCombineAsync(CompletionStage, BiFunction)
3. thenCombineAsync(CompletionStage, BiFunction, Executor)

从参数上看，差异并不大，仅第三个办法入参中多了线程池Executor目标。看下三个办法的源码完结，会发现其全体完结逻辑都是共同的，只是是运用线程池这个当地的逻辑有一点点的差异：

有爱好的能够去翻一下此部分的源码完结，这儿概括下三者的差异：

1. thenCombine办法，沿用上一个履行使命所运用的线程池进行处理
2. thenCombineAsync两个入参的办法，运用默许的ForkJoinPool线程池中的作业线程进行处理
3. themCombineAsync三个入参的办法，支撑自定义线程池并指定运用自定义线程池中的线程作为作业线程去处理待履行使命。

为了更好的了解下上述的三个差异点，咱们经过下面的代码来演示下：

• **用法1： **其中一个supplyAsync办法以及thenCombineAsync指定运用自定义线程池，另一个supplyAsync办法不指定线程池（运用默许线程池）

public PriceResult getCheapestPlatAndPrice4(String product) {
    // 结构自定义线程池
    ExecutorService executor = Executors.newFixedThreadPool(5);
    return
        CompletableFuture.supplyAsync(
            () -> HttpRequestMock.getMouXiXiPrice(product), 
            executor
        ).thenCombineAsync(
            CompletableFuture.supplyAsync(() -> HttpRequestMock.getMouXiXiDiscounts(product)),
            this::computeRealPrice,
            executor
        ).join();
}

对上述代码完结战略的解读，以及与履行成果的联系展示如下图所示，能够看出，没有指定自定义线程池的supplyAsync办法，其运用了默许的ForkJoinPool作业线程来运转，而别的两个指定了自定义线程池的办法，则运用了自定义线程池来履行。

• 用法2： 不指定自定义线程池，运用默许线程池战略，运用thenCombine办法

public PriceResult getCheapestPlatAndPrice5(String product) {
    return
        CompletableFuture.supplyAsync(
            () -> HttpRequestMock.getMouXiXiPrice(product)
        ).thenCombine(
            CompletableFuture.supplyAsync(() -> HttpRequestMock.getMouXiXiDiscounts(product)),
            this::computeRealPrice
        ).join();
}

履行成果如下，能够看到履行线程称号与用法1示例比较发生了改变。由于没有指定线程池，所以两个supplyAsync办法都是用的默许的ForkJoinPool线程池，而thenCombine运用的是上一个使命所运用的线程池，所以也是用的ForkJoinPool。

14:34:27.815[ForkJoinPool.commonPool-worker-1|12]获取某夕夕上 Iphone13的价格
14:34:27.815[ForkJoinPool.commonPool-worker-2|13]获取某夕夕上 Iphone13的优惠
14:34:28.831[ForkJoinPool.commonPool-worker-2|13]获取某夕夕上 Iphone13的优惠完结： -5300
14:34:28.831[ForkJoinPool.commonPool-worker-1|12]获取某夕夕上 Iphone13的价格完结： 5399
14:34:28.831[ForkJoinPool.commonPool-worker-2|13]某夕夕终究价格核算完结：99
获取最优价格信息：【渠道：某夕夕, 原价：5399, 扣头：0, 实付价：99】
-----履行耗时： 1083ms  ------

现在，咱们知道了办法称号带有Async和不带Async的完结战略上的差异点就在于运用哪个线程池来履行罢了。那么，对咱们实践的指导意义是啥呢？实践运用的时分，咱们怎样判别自己应该运用带Async结尾的办法、仍是不带Async结尾的办法呢？

上面是Async结尾办法默许运用的ForkJoinPool创立的逻辑，这儿能够看出，默许的线程池中的作业线程数是CPU核数 - 1，而且指定了默许的丢弃战略等，这便是一个首要关键点。

所以说，契合以下几个条件的时分，能够考虑运用带有Async后缀的办法，指定自定义线程池：

• 默许线程池的线程数满意不了实践诉求
• 默许线程池的类型不契合自己事务诉求
• 默许线程池的队列满处理战略不满意自己诉求

与Stream结合运用的留意点

在我前面的文档中，有详尽全面的介绍过Stream流相关的运用办法（不清楚的同学速点《吃透JAVA的Stream流操作，多年实践总结》了解下啦）。在触及批量进行并行处理的时分，经过Stream与CompletableFuture结合运用，能够简化咱们的许多编码逻辑。可是在运用细节方面需求留意下，防止达不到运用CompletableFuture的预期作用。

需求场景： 在同一个渠道内，传入多个产品，查询不同产品对应的价格与优惠信息，并选出实付价格最低的产品信息。

结合前面的介绍剖析，咱们应该知道最佳的办法，便是一同并行的办法去各自恳求数据，终究兼并处理即可。所以咱们规划依照如下的战略来完结：

先看第一种编码完结：

public PriceResult comparePriceInOnePlat(List<String> products) {
    return products.stream()
            .map(product ->
                    CompletableFuture.supplyAsync(() -> HttpRequestMock.getMouBaoPrice(product))
                            .thenCombine(
                                    CompletableFuture.supplyAsync(() -> HttpRequestMock.getMouBaoDiscounts(product)),
                                    this::computeRealPrice))
            .map(CompletableFuture::join)
            .sorted(Comparator.comparingInt(PriceResult::getRealPrice))
            .findFirst()
            .get();
}

关于List的处理场景，这儿采用了Stream办法来进行遍历与成果的搜集、排序与回来。看似正常，可是履行的时分会发现，并没有到达咱们预期的作用：

07:37:15.408[ForkJoinPool.commonPool-worker-9|12]获取某宝上 Iphone13黑色的价格完结： 5199
07:37:15.408[ForkJoinPool.commonPool-worker-2|13]获取某宝上 Iphone13黑色的优惠完结： -200
07:37:15.408[ForkJoinPool.commonPool-worker-2|13]某宝终究价格核算完结：4999
07:37:16.410[ForkJoinPool.commonPool-worker-9|12]获取某宝上 Iphone13白色的价格完结： 5199
07:37:16.410[ForkJoinPool.commonPool-worker-11|14]获取某宝上 Iphone13白色的优惠完结： -200
07:37:16.410[ForkJoinPool.commonPool-worker-11|14]某宝终究价格核算完结：4999
07:37:17.412[ForkJoinPool.commonPool-worker-11|14]获取某宝上 Iphone13赤色的价格完结： 5199
07:37:17.412[ForkJoinPool.commonPool-worker-9|12]获取某宝上 Iphone13赤色的优惠完结： -200
07:37:17.412[ForkJoinPool.commonPool-worker-9|12]某宝终究价格核算完结：4999
获取最优价格信息：【渠道：某宝, 原价：5199, 扣头：0, 实付价：4999】
-----履行耗时： 3132ms  ------

从上述履行成果能够看出，其具体处理的时分，其实是依照下面的逻辑去处理了：

为什么会呈现这种实践与预期的差异呢？原因就在于咱们运用的Stream上面！虽然Stream中运用两个map办法，但Stream处理的时分并不会别离遍历两遍，其实写法等同于下面这种写到1个map中处理，改为下面这种写法，其实大家也就更容易理解为啥会没有到达咱们预期的全体并行作用了：

public PriceResult comparePriceInOnePlat1(List<String> products) {
    return products.stream()
        .map(product -> CompletableFuture.supplyAsync(() -> HttpRequestMock.getMouBaoPrice(product)).thenCombine(CompletableFuture.supplyAsync(() -> HttpRequestMock.getMouBaoDiscounts(product)), this::computeRealPrice).join())
        .sorted(Comparator.comparingInt(PriceResult::getRealPrice))
        .findFirst()
        .get();
    }

既然如此，这种场景是不是就不能运用Stream了呢？也不是，其实咱们拆开成两个Stream分步操作下其实就能够了。

再看下面的第二种完结代码：

public PriceResult comparePriceInOnePlat2(List<String> products) {
    // 先触发各自渠道的并行处理
    List<CompletableFuture<PriceResult>> completableFutures = products.stream()
            .map(product -> CompletableFuture.supplyAsync(() -> HttpRequestMock.getMouBaoPrice(product)).thenCombine(CompletableFuture.supplyAsync(() -> HttpRequestMock.getMouBaoDiscounts(product)), this::computeRealPrice))
            .collect(Collectors.toList());
    // 在独立的流中，等候一切并行处理结束，做终究成果处理
    return completableFutures.stream()
            .map(CompletableFuture::join)
            .sorted(Comparator.comparingInt(PriceResult::getRealPrice))
            .findFirst()
            .get();
}

履行成果：

07:39:16.072[ForkJoinPool.commonPool-worker-6|17]获取某宝上 Iphone13赤色的价格完结： 5199
07:39:16.072[ForkJoinPool.commonPool-worker-9|12]获取某宝上 Iphone13黑色的价格完结： 5199
07:39:16.072[ForkJoinPool.commonPool-worker-2|13]获取某宝上 Iphone13黑色的优惠完结： -200
07:39:16.072[ForkJoinPool.commonPool-worker-11|14]获取某宝上 Iphone13白色的价格完结： 5199
07:39:16.072[ForkJoinPool.commonPool-worker-4|15]获取某宝上 Iphone13白色的优惠完结： -200
07:39:16.072[ForkJoinPool.commonPool-worker-13|16]获取某宝上 Iphone13赤色的优惠完结： -200
07:39:16.072[ForkJoinPool.commonPool-worker-2|13]某宝终究价格核算完结：4999
07:39:16.072[ForkJoinPool.commonPool-worker-4|15]某宝终究价格核算完结：4999
07:39:16.072[ForkJoinPool.commonPool-worker-13|16]某宝终究价格核算完结：4999
获取最优价格信息：【渠道：某宝, 原价：5199, 扣头：0, 实付价：4999】
-----履行耗时： 1142ms  ------

从履行成果能够看出，三个产品并行处理，全体处理耗时比较前面编码办法有很大提高，到达了预期的作用。

归纳下：

由于Stream的操作具有延迟履行的特色，且只要遇到停止操作（比方collect办法）的时分才会真正的履行。所以遇到这种需求并行处理且需求兼并多个并行处理流程的状况下，需求将并行流程与兼并逻辑放到两个Stream中，这样别离触发完结各自的处理逻辑，就能够了。

甜点：并发和并行的差异

对一个吃货而言，主餐结束，总得来点餐后甜点才够满意。

在前面的内容中呢，咱们始终是在环绕并行处理这个话题在打开。实践作业的时分，咱们关于并发这个词肯定也不生疏，高并发这个词，就像高端人士酒杯中那八二年的拉菲一般，成了每一个开发人员简历上用来显示实力的一个标签。

那么，并发和并行到底啥差异？这儿咱们也简略的概括下。

并发

关于并发的具体内容，能够拜见我写的另一篇内容，也即本篇文章的姊妹篇《不堆概念，换个视点聊多线程并发编程》，下面这儿简略的介绍下并发的概念。

所谓并发，其重视的点是服务器的吞吐量状况，也便是服务器能够在单位时刻内一同处理多少个恳求。并发是经过多线程的办法来完结的，充分运用当时CPU多核才能，一同运用多个进程去处理事务，使得同一个机器在相一同间内能够处理更多的恳求，提高吞吐量。

一切的操作在一个线程中串行推动，假如有多个线程同步处理，则一同有多个恳求能够被处理。可是由于是串行处理，所以假如某个环节需求对外交互时，比方等候网络IO的操作，会使妥当时线程处于堵塞状态，直到资源可用时被唤醒持续往后履行。

关于高并发场景，服务器的线程资源是非常宝贵的。假如频繁的处于堵塞则会导致糟蹋，且线程频繁的堵塞、唤醒切换动作，也会加剧全体体系的功用损耗。所以并发这种多线程场景，更适合CPU密集型的操作。

并行

所谓并行，便是将同一个处理流程没有彼此依靠的部分放到多个线程中进行一同并行处理，以此来到达相关于串行形式更短的单流程处理耗时的作用，进而提高体系的全体呼应时长与吞吐量。

依据异步编程完结的并行操作也是凭借线程池的办法，经过多线程一同履行来完结功率提高的。与并发的差异在于：并行经过将使命切分为一个个可独立处理的小使命块，然后依据体系调度战略，将需求履行的使命块分配给空闲可用作业线程去处理，假如呈现需求等候的场景（比方IO恳求）则作业线程会将此使命先放下，持续处理后续的使命，等之前的使命IO恳求好了之后，体系重新分配可用的作业线程来处理。

依据上面的示意图介绍能够看出，异步并行编程，关于作业线程的运用率上升，不会呈现作业线程堵塞的状况，可是由于使命拆分、作业线程间的切换调度等体系层面的开销也会随之加大。

怎样选择

前面介绍了下并发与并行两种形式的特色、以及各自的优缺点。所以选择采用并发仍是并行办法来提高体系的处理功用，还需求结合实践项目场景来确认。

归纳而言

1. 假如事务处理逻辑是CPU密集型的操作，优先运用依据线程池完结并发处理方案（能够防止线程间切换导致的体系功用糟蹋）。
2. 假如事务处理逻辑中存在较多需求堵塞等候的耗时场景、且彼此之间没有依靠，比方本地IO操作、网络IO恳求等等，这种状况优先选择运用并行处理战略（能够防止宝贵的线程资源被堵塞等候）。

总结回顾

好啦，关于JAVA中CompletableFuture的运用，以及并行编程相关的内容呢就介绍到这儿啦。看到这儿，相信您应该有所收成吧？那么你的项目里有这种适兼并行处理的场景吗？你在处理并行场景的时分是怎样做的呢？评论区一同评论下吧~~

弥补：

本文中有提及CompletableFuture履行时所运用的默许线程池是ForkJoinPool，早在JAVA7版别就现已被引进，可是许多人对ForkJoinPool不是很了解，实践项目中运用的也比较少。其实对ForkJoinPool的合理运用，能够让咱们在面临某些多线程场景时会愈加的从容高效。在后边的文章中，我会针对ForkJoinPool有关的内容进行专门的介绍与讨论，假如有爱好，能够点个重视，及时获取后续的内容。

此外

• 关于本文中触及的演示代码的完好示例，我现已整理并提交到github中，假如您有需求，能够自取：github.com/veezean/Jav…

我是悟道，聊技能、又不只是聊技能~

假如觉得有用，请点赞 + 重视让我感受到您的支撑。也能够重视下我的公众号【架构悟道】，获取更及时的更新。

期待与你一同讨论，一同成长为更好的自己。

---

我正在参加技能社区创作者签约方案招募活动，点击链接报名投稿。