分享一篇多线程编排,纯技术干货,不讲废话。

 

0x1

先写个Demo

我们以商品详情页为例,如果想在一个接口内返回商详中所有的数据,大概需要以下几个步骤:

  1. 根据商品id,请求商品详细数据(接口名:spuDetail,耗时:400ms)

  2. 根据商品id和位置信息,请求该商品支持的配送区域(接口名:saleArea,耗时:350ms

  3. 根据1返回的数据,获取商品对应的店铺code,请求店铺信息(接口名:findShopInfo,耗时:400ms)

  4. 根据1返回的数据,请求商品规格接口(接口名:spuSpecInfo,耗时:350ms根据1返回的信息,请求该店铺的推荐商品(接口名:appSearch,耗时:400ms)

  5. 根据3返回的信息,请求商品的优惠券信息(接口名:showCouponTag,耗时:350ms

一般的情况下,很多小伙伴都会按下面的写法一把梭哈,一个个请求,等到拿到数据之后进行商详页数据的拼装并返回。

为了更直观的看到整个过程,我们使用简单直观的方法来模拟接口请求:

public class SkuDetail {

    public static String spuDetail(String skuCode) {
        System.out.println("商品基础信息===》" + skuCode);
        mockTime(400L);
        return "商品基础信息";
    }

    public static String saleArea(String skuCode) {
        System.out.println("销售区域===》" + skuCode);
        mockTime(350L);
        return "销售区域";
    }

    public static String findShopInfo(String shopCode) {
        System.out.println("店铺信息===》" + shopCode);
        mockTime(400L);
        return "店铺信息";
    }

    public static String spuSpecInfo(String spuCode) {
        System.out.println("规格接口===》" + spuCode);
        mockTime(350L);
        return "规格接口";
    }

    public static String appSearch(String shopCode) {
        System.out.println("推荐===》" + shopCode);
        mockTime(400L);
        return "推荐";
    }

    public static String showCouponTag(String shopCode) {
        System.out.println("优惠券===》" + shopCode);
        mockTime(350L);
        return "优惠券";
    }


    /**
     * 模拟耗时
     * @param time
     */
    private static void mockTime(long time) {
        try {
            Thread.sleep(time);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }


    public static void main(String[] args) throws InterruptedException, ExecutionException {
        final long begin = System.currentTimeMillis();
        String skuCode = "123";
        //根据商品id,请求商品详细数据
        String spuDetail = spuDetail(skuCode);
        // 根据商品id和位置信息,请求该商品支持的配送区域
        String saleArea = saleArea(skuCode);
        // 模拟根据1返回的数据,获取商品对应的店铺code,请求店铺信息
        String shopCode = skuCode;
        String shopInfo = findShopInfo(shopCode);
        // 模拟根据1返回的数据,通过spuCode请求商品规格接口
        String spuCode = skuCode;
        String spuSpecInfo = spuSpecInfo(spuCode);
        // 根据3返回的信息,请求该店铺的推荐商品
        String recommend = appSearch(shopCode);
        // 根据3返回的信息,请求商品的优惠券信息
        String couponTag = showCouponTag(skuCode);

        // 组装VO返回
        SkuVO skuVO = new SkuVO(spuDetail, saleArea, shopInfo, spuSpecInfo, recommend, couponTag);
        // 模拟返回
        System.out.println(skuVO);
        final long end = System.currentTimeMillis();
        System.out.println("整个过程耗时=================》" + (end - begin));
    }
}

@Data
@NoArgsConstructor
@AllArgsConstructor
class SkuVO{
    private String spuDetail;
    private String saleArea;
    private String shopInfo;
    private String spuSpecInfo;
    private String recommend;
    private String couponTag;
}

程序运行结果:

商品基础信息===》123
销售区域===》123
店铺信息===》123
规格接口===》123
推荐===》123
优惠券===》123
SkuVO(spuDetail=商品基础信息, saleArea=销售区域, shopInfo=店铺信息, spuSpecInfo=规格接口, recommend=推荐, couponTag=优惠券)
整个过程耗时=================》2313

这样编写代码,整个商详页的数据获取需要耗时=================2313ms

这代码是早上写的,工位是下午收拾的,人是连夜扛着火车走的。

有些同学可能觉得看起来好像还行,能用就行,要啥自行车。但是仔细想想,这个接口还有挺大的优化空间:

仔细看看,接口虽然多,但是依赖性不大,如果要加快接口响应速度,可以尝试使用:多线程(依赖性不大,可以多线程请求)、Future(获取店铺信息时,需要商品基础信息返回后才能操作)、CountDownLatch(所有接口请求完之后,才能响应前端。

但是,上面说到的这些,你写我推荐,真写我不写

为啥?因为这光想想就觉得复杂,本来多线程就是个挺复杂的玩意,这写起来太麻烦了。

人一懒,就总能想出偷懒的办法。

这不,JDK8就提供了一个用来处理类似问题的利器:CompletableFuture

 

0x2

多线程编排利器

CompletableFuture是JDK8新增的,它提供了非常强大的Future的扩展功能,可以帮助我们简化异步编程的复杂性,并且提供了函数式编程的能力,可以通过回调的方式处理计算结果,也提供了转换和组合 CompletableFuture 的方法。

它实现了Future接口,还有CompletionStage接口,这让他拥有Future的功能,也拥有一个阶段完成以后可能会触发另外一个阶段的能力。

CompletableFuture 提供了四个静态方法来创建一个异步操作。

public class CompletableFutureTimplements FutureT>, CompletionStageT>

这里其实分两组supplyAsync(带返回值)、runAsync(不带返回值),至于这两组下面带不带线程池的区别,只是使用者是否自定义线程池供异步任务执行使用。

不指定线程池的时候,其内部也是会有默认的ForkJoin池供异步执行任务使用。

我们来看看这东西怎么使用:

public static void main(String[] args) throws InterruptedException, ExecutionException {
        final long begin = System.currentTimeMillis();
        String skuCode = "123";
        SkuVO result = new SkuVO();
        //根据商品id,请求商品详细数据
        CompletableFuture spuDetail = CompletableFuture.supplyAsync(() -> {
            String detail = spuDetail(skuCode);
            result.setSpuDetail(detail);
            return detail;
        });

        // 根据商品id和位置信息,请求该商品支持的配送区域
        CompletableFuture saleArea = CompletableFuture.runAsync(() -> result.setSaleArea(saleArea(skuCode)));

        // spuDetail执行完之后,获取spuCode继续执行其他
        CompletableFuture shopInfo = spuDetail.thenApplyAsync(shopCode -> {
            String info = findShopInfo(shopCode);
            result.setShopInfo(info);
            return info;
        });
        // 模拟根据1返回的数据,通过spuCode请求商品规格接口
        CompletableFuture spuSpecInfo = spuDetail.thenAcceptAsync(spuCode -> result.setSpuSpecInfo(spuSpecInfo(spuCode)));

        // 根据1返回的信息,请求该店铺的推荐商品
        CompletableFuture recommend = spuDetail.thenAcceptAsync(shopCode ->
                result.setRecommend(appSearch(shopCode))
        );
        // 根据1返回的信息,请求商品的优惠券信息
        CompletableFuture couponTag = spuDetail.thenAcceptAsync(shopCode -> result.setCouponTag(showCouponTag(shopCode)));

        // 等待所有任务完成
        CompletableFuture.allOf(spuDetail, saleArea, shopInfo, spuSpecInfo, recommend, couponTag).get();
        // 模拟返回
        System.out.println(result);
        final long end = System.currentTimeMillis();
        System.out.println("整个过程耗时=================》" + (end - begin));
    }

程序运行结果:

商品基础信息===》123
销售区域===》123
店铺信息===》商品基础信息
规格接口===》商品基础信息
推荐===》店铺信息
优惠券===》店铺信息
SkuVO(spuDetail=商品基础信息, saleArea=销售区域, shopInfo=店铺信息, spuSpecInfo=规格接口, recommend=推荐, couponTag=优惠券)
整个过程耗时=================》1297

相比于没有做任务编排的请求方式(2313ms),进行多线程编排之后的速度快了近1倍。

 

0x3

深入了解CompletableFuture

在简单的了解了CompletableFuture的应用场景之后,我们可以仔细看看CompletableFuture。我们平时的使用中,基本上就使用到这个类里面的静态方法:

方法名 作用

supplyAsync(Supplier supplier)

异步执行对应的supplier,带返回值的

supplyAsync(Supplier supplier,Excutor excutor)

使用自定义的线程池异步执行对应的supplier,带返回值的

runAsync(Runnable runnable)

异步执行对应的runnable任务,无返回值

runAsync(Runnable runnable,Excutor excutor)

使用自定义的线程池异步执行对应的runnable任务,无返回值

completedFuture(U value)

返回值为value的CompletedFuture,这个应用场景较少

allOf(CompletableFuture>… cfs)

返回一个新的 CompletableFuture,它在所有给定的 CompletableFuture 完成时完成

anyOf(CompletableFuture>… cfs)

返回一个新的 CompletableFuture,它在任何给定的 CompletableFuture 完成时完成,结果相同

上面的方法中,除了不太常用的completedFuture,就只有anyOf我们在上面的程序中没有使用过。

这个有什么用呢?比如dubbo中,有一种策略是对所有服务端发起调用,只要有一个有返回数据,就将数据返回。这种场景下面就可以使用到我们的anyOf。

public class AnyOfDemo {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        CompletableFuture first = CompletableFuture.supplyAsync(getStringSupplier(5000L));
        CompletableFuture second = CompletableFuture.supplyAsync(getStringSupplier(6000L));
        CompletableFuture three = CompletableFuture.supplyAsync(getStringSupplier(3000L));
        CompletableFuture result = CompletableFuture.anyOf(first, second, three);        System.out.println(result.get());    }    private static Supplier getStringSupplier(Long time) {        return () -> {            System.out.println(time + "准备执行");            try {                Thread.sleep(time);            } catch (InterruptedException e) {                e.printStackTrace();            }            System.out.println(time + "执行完毕");            return String.valueOf(time);        };    }}

程序输出结果:

6000准备执行
5000准备执行
3000准备执行
3000执行完毕
3000

我们可以看到,调用的时候是三个都执行,只要第一返回之后,其余的接调用就会中断。

 

0x4

实例方法任务编排

上面罗列出来的几个静态方法,可以用于产生CompleteFuture实例,之后便可以调用器实例方法,进行多线程任务的编排。

我们通过下面的例子来看看它提供的强大的API。

为避免过多无用的描述,做如下规定:

实例方法中,有很多名字为xxx、xxxAsync、xxxAsync,他们功能都是类似的,带Async表示其实另起线程异步执行,两个带Async的方法,入参一个有executor,表示使用指定的线程池执行,没有话用默认的线程池执行。后续描述中,只会出现xxx,另外两个方法不做描述。

isDone()

判断是否执行完成

public static void main(String[] args) throws InterruptedException {
        CompletableFuture completableFuture = CompletableFuture.runAsync(() -> {
            try {
                Thread.sleep(4000L);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println("执行完成");
        });
        System.out.println(completableFuture.isDone());
        Thread.sleep(5000L);
        System.out.println(completableFuture.isDone());
    }

输出结果:
false
执行完成
true

get()

用于获取执行结果

public static void main(String[] args) throws InterruptedException, ExecutionException {
        CompletableFuture integerCompletableFuture = CompletableFuture.supplyAsync(() -> 100);
        System.out.println(integerCompletableFuture.get());
}
输出结果:
100

get(long timeout, TimeUnit unit)

带超时时间的get,规定时间内获取执行结果,如果超时会抛异常

public static void main(String[] args) throws InterruptedException, ExecutionException, TimeoutException {
        CompletableFuture integerCompletableFuture = CompletableFuture.supplyAsync(() -> {
            try {
                // 模拟耗时操作
                Thread.sleep(2000L);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return 100;
        });
        System.out.println(integerCompletableFuture.get(1000L, TimeUnit.MILLISECONDS));
    }

运行结果:
Exception in thread "main" java.util.concurrent.TimeoutException

getNow(valueIfAbsent)

如果完成,则返回结果值(或抛出任何遇到的异常),否则将入参作为返回值返回。

public static void main(String[] args) throws InterruptedException, ExecutionException, TimeoutException {
        CompletableFuture integerCompletableFuture = CompletableFuture.supplyAsync(() -> {
            try {
                Thread.sleep(2000L);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return 100;
        });
        // 如果还没执行完成,就返回1000
        System.out.println(integerCompletableFuture.getNow(1000));
    }
运行结果:
1000

complete(T value)

如果尚未完成,将get和相关方法返回的值设置为给定值。

public static void main(String[] args) throws InterruptedException, ExecutionException, TimeoutException {
        CompletableFuture integerCompletableFuture = CompletableFuture.supplyAsync(() -> {
            try {
                Thread.sleep(2000L);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return 100;
        });
        // 如果还没执行完成,将返回值设置1000
        integerCompletableFuture.complete(1000);
        System.out.println(integerCompletableFuture.get());
    }
运行结果:
1000

completeExceptionally(Throwable ex)

如果get的时候仍未执行完成,抛出给定的异常

public static void main(String[] args) throws InterruptedException, ExecutionException, TimeoutException {
        CompletableFuture integerCompletableFuture = CompletableFuture.supplyAsync(() -> {
            try {
                Thread.sleep(2000L);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return 100;
        });
        // 如果还没执行完成,get的时候抛出异常
        integerCompletableFuture.completeExceptionally(new TimeoutException("超时了"));
        System.out.println(integerCompletableFuture.get());
    }

运行结果:
Exception in thread "main" java.util.concurrent.ExecutionException: java.util.concurrent.TimeoutException: 超时了

thenApply(Function super T,extends U> fn)

执行完当前任务后,将当前任务返回值传递给另一个任务(fn),并【同步】执行该任务,执行结果封装成CompletableFuture进行返回。

public static void main(String[] args) throws InterruptedException, ExecutionException, TimeoutException {
        CompletableFuture integerCompletableFuture = CompletableFuture.supplyAsync(() -> 100);
        //将第一个的返回值进行加工,并返回新的CompletableFuture
        CompletableFuture apply = integerCompletableFuture.thenApply(value -> value * 10);
        System.out.println(apply.get());
    }
运行结果:
1000

thenAccept(Consumer super T> action)

执行完当前任务后,将当前任务返回值传递给另一个任务(action)并【同步】执行给定任务,该方法无返回值

public static void main(String[] args) throws InterruptedException, ExecutionException, TimeoutException {
    CompletableFuture integerCompletableFuture = CompletableFuture.supplyAsync(() -> 100);
    //将第一个的返回值进行加工,但是不会返回
    integerCompletableFuture.thenAccept(value -> System.out.println("在方法内消耗上一个返回值" + value));
}
运行结果:
在方法内消耗上一个返回值100

thenRun(Runnable action)

执行完当前任务后,【同步】执行对应的action,该方法不关注当前方法的返回值

public static void main(String[] args) throws InterruptedException, ExecutionException, TimeoutException {
        CompletableFuture integerCompletableFuture = CompletableFuture.supplyAsync(() -> {
            System.out.println("执行完毕");
            return 100;
        });
        //上一个执行完成后触发,无返回
        integerCompletableFuture.thenRun(() -> System.out.println("上一个执行完成后触发"));
    }

执行结果:
执行完毕
上一个执行完成后触发

thenCombine(CompletionStage extends U> otherBiFunction super T,super U,extends V> fn)

在两个任务都执行完成后,将结果交给fn进行处理,处理完的结果包装成CompletableFuture

public static void main(String[] args) throws InterruptedException, ExecutionException, TimeoutException {
        CompletableFuture one = CompletableFuture.supplyAsync(() -> {
            System.out.println("one 执行完毕");
            return 100;
        });
        CompletableFuture two = CompletableFuture.supplyAsync(() -> {
            System.out.println("two 执行完毕");
            return 10;
        });

        // 将one、two两个的结果传递给func处理,并返回新的CompletableFuture
        CompletableFuture three = one.thenCombine(two, (v1, v2) -> v1 * v2);
        System.out.println(three.get());
    }
运行结果:1000

thenAcceptBoth(CompletionStage extends U> other,
BiConsumer super Tsuper U> action)

在两个任务都执行完成后,将结果交给fn进行处理,本身也无返回值。

public static void main(String[] args) throws InterruptedException, ExecutionException, TimeoutException {
        CompletableFuture one = CompletableFuture.supplyAsync(() -> {
            System.out.println("one 执行完毕");
            return 100;
        });
        CompletableFuture two = CompletableFuture.supplyAsync(() -> {
            System.out.println("two 执行完毕");
            return 10;
        });

        // 将one、two两个的结果传递给func处理,不返回
        one.thenAcceptBoth(two, (v1, v2) -> System.out.println(v1 * v2));
    }
运行结果:1000

runAfterBoth(CompletionStage> other,
Runnable action)

在两个任务都执行完成后,执行对应的action,不关心两个任务的返回值,且本身也无返回值。

public static void main(String[] args) throws InterruptedException, ExecutionException, TimeoutException, IOException {
        CompletableFuture one = CompletableFuture.supplyAsync(() -> {
            try {
                Thread.sleep(2000L);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println("one 执行完毕");
            return 100;
        });
        CompletableFuture two = CompletableFuture.supplyAsync(() -> {
            System.out.println("two 执行完毕");
            return 10;
        });

        long start = System.currentTimeMillis();
        // 将one、two两个的结果传递给func处理,不返回
        one.runAfterBoth(two, () -> {
            long end = System.currentTimeMillis();
            System.out.println("程序耗时:" + (end - start) + "ms");
            System.out.println("都运行完了");
        });
        System.in.read();
    }
运行结果:
two 执行完毕
one 执行完毕
程序耗时:2014ms
都运行完了

applyToEither(CompletionStage extends T> otherFunction super TU> fn)

两个任务,谁执行的快就用谁的返回值进行下一步的操作,执行结果包装成CompletableFuture进行返回。

public static void main(String[] args) throws InterruptedException, ExecutionException, TimeoutException, IOException {
        CompletableFuture one = CompletableFuture.supplyAsync(() -> {
            try {
                Thread.sleep(2000L);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println("one 执行完毕");
            return 100;
        });
        CompletableFuture two = CompletableFuture.supplyAsync(() -> {
            System.out.println("two 执行完毕");
            return 10;
        });

        // 将one、two两个先执行完的结果传递给func加工,返回新CompletableFuture
        CompletableFuture result = one.applyToEither(two, v -> v * 10);
        System.out.println(result.get());
    }
执行结果:
two 执行完毕
100

acceptEither(CompletionStage extends T> otherConsumer super T> action)

两个任务,谁执行的快就用谁的返回值进行下一步的操作,无执行结果。

public static void main(String[] args) throws InterruptedException, ExecutionException, TimeoutException, IOException {
        CompletableFuture one = CompletableFuture.supplyAsync(() -> {
            try {
                Thread.sleep(2000L);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println("one 执行完毕");
            return 100;
        });
        CompletableFuture two = CompletableFuture.supplyAsync(() -> {
            System.out.println("two 执行完毕");
            return 10;
        });

        // 将one、two两个先执行完的结果传递给func加工,不返回
        one.acceptEither(two, System.out::println);
    }

执行结果:
two 执行完毕
10

runAfterEither(CompletionStage> other,
Runnable action)

两个任务,只要有一个完成,就触发下一步操作。不关心上一步返回值,本身也无返回值。

public static void main(String[] args) throws InterruptedException, ExecutionException, TimeoutException, IOException {
        CompletableFuture one = CompletableFuture.supplyAsync(() -> {
            try {
                Thread.sleep(2000L);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println("one 执行完毕");
            return 100;
        });
        CompletableFuture two = CompletableFuture.supplyAsync(() -> {
            System.out.println("two 执行完毕");
            return 10;
        });

        // 将one、two两个先执行完的结果传递给func加工,不返回
        one.runAfterEither(two,()->{
            System.out.println("有一个执行完了");
        });
    }
运行结果:
two 执行完毕
有一个执行完了

thenCompose(Function super Textends CompletionStageU>> fn)

处理完第一个任务,将第一个任务的返回值传递给第二个任务,由第二个任务执行后返回新的CompletableFuture。乍一看跟thenApply差不多。两者本质的区别在于:thenApply对第一个任务的返回值进行转换,thenCompose将第一个任务的返回值传递给另一个CompletableFuture。

public static void main(String[] args) throws InterruptedException, ExecutionException, TimeoutException, IOException {
        CompletableFuture one = CompletableFuture.supplyAsync(() -> {
            return 100;
        });

        // 将one执行完的结果传递给另外的completeFuture处理,返回新的CompletableFuture
        CompletableFuture integerCompletableFuture = one.thenCompose(value -> CompletableFuture.supplyAsync(() -> value * 10));
        System.out.println(integerCompletableFuture.get());
    }
运行结果:1000

whenComplete(BiConsumer super Tsuper Throwable> action)

跟thenApply类似,都是将上一次的结果进行加工,同时whenComplete支持对上一次处理发生的异常进行处理。不同的是:whenComplete 的返回值是上个任务的返回值其无法对结果的栈地址进行改变,无需显式return。而thenApply可以改变前任务的结果。

public static void main(String[] args) throws InterruptedException, ExecutionException, TimeoutException, IOException {
        CompletableFuture one = CompletableFuture.supplyAsync(() -> {
            return 100;
        });

        // whenComplete 无法改变上一个返回值的【地址】
        CompletableFuture integerCompletableFuture = one.whenComplete((value, exception) -> {
            value = value * 10;
        });
        System.out.println(integerCompletableFuture.get());
    }

输出结果:100

如果是对于实体里某个字段的变更,这种是可以的:

public static void main(String[] args) throws InterruptedException, ExecutionException, TimeoutException, IOException {

        CompletableFuture one = CompletableFuture.supplyAsync(() -> {
            Foo foo = new Foo();
            foo.setId(1);
            foo.setName("tomcat");
            return foo;
        });

        // whenComplete 无法改变上一个返回值的【地址】
        CompletableFuture integerCompletableFuture = one.whenComplete((value, exception) -> {
            value.setId(10);
            value.setName("test");
        });
        System.out.println(integerCompletableFuture.get());
    }

    @Data
   static class Foo{
        private Integer id;
        private String name;
   }

输出结果:
Test.Foo(id=10, name=test)

handle(BiFunction super TThrowableextends U> fn)

跟whenComplete类似,不同的是:whenComplete对上次的执行结果进行加工,不会返回新的地址。handle接受上一次的结果,自己处理后有自己的返回值。

public static void main(String[] args) throws InterruptedException, ExecutionException, TimeoutException, IOException {

        CompletableFuture one = CompletableFuture.supplyAsync(() -> {
            Foo foo = new Foo();
            foo.setId(1);
            foo.setName("tomcat");
            return foo;
        });

        // handle自行处理后,有自己的返回值
        CompletableFuture integerCompletableFuture = one.handle((value, exception) -> {
            Foo foo = new Foo();
            foo.setName(value.getName());
            foo.setId(value.getId());
            return foo;
        });
        System.out.println(integerCompletableFuture.get());
    }

    @Data
    static class Foo {
        private Integer id;
        private String name;
    }

exceptionally(Function

 

 

extends T> fn)

 

用于处理异常,当thenApply执行的时候发生异常,可以由exceptionaly捕获异常,并处理

public static void main(String[] args) throws InterruptedException, ExecutionException, TimeoutException, IOException {

        CompletableFuture completableFuture = CompletableFuture.supplyAsync(() -> {
            return 0;
        }).thenApply(value -> {
            // 这里会有异常
            return 100 / value;
        }).exceptionally(e -> 10);

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

输出结果:
10

好了 关于CompletableFuture的介绍就到这了,以后就可以很愉快的进行多线程的编排了。


0 条评论

发表回复

Avatar placeholder

您的电子邮箱地址不会被公开。 必填项已用 * 标注

此站点使用Akismet来减少垃圾评论。了解我们如何处理您的评论数据

蜀ICP备16001794号
© 2014 - 2024 linpxing.cn All right reserved.