多线程并发编排CompletableFuture分解+代码实践

原创

小哥 3年前 (2022-11-09) 阅读数 5 #大杂烩

1.从问题开始

假设在编写购物网站时，用户点击查看商品，我们的后端必须向用户显示商品的所有多样化维度信息（专业名称应为sku和spu)，如果我们以单线程方式进行，逐步串行查询商品信息，ServiceImpl图层查询代码如下：

@Override //SkuInfoServiceImpl  @TableName("pms_sku_info")
public SkuItemVo item(Long skuId) {
    SkuItemVo skuItemVo = new SkuItemVo();
    //1、sku基本信息的获取  查pms_sku_info表
    SkuInfoEntity skuInfoEntity = this.getById(skuId);
    skuItemVo.setInfo(skuInfoEntity);
    Long spuId = skuInfoEntity.getSpuId();
    Long catalogId = skuInfoEntity.getCatalogId();

    //2、sku图片信息    查pms_sku_images表
    List skuImagesEntities = skuimagesService.list(new QueryWrapper().eq("sku_id", skuId));
    skuItemVo.setimages(skuimagesEntities);

    //3、获取spu销售属性的组合。-> 需要依赖第一个一个一个1请求返回值的步骤。spuId
    List saleAttrVos=skuSaleAttrValueService.listSaleAttrs(spuId);
    skuItemVo.setSaleAttr(saleAttrVos);

    //4、获取spu的介绍-> 需要依赖第一个一个一个1请求返回值的步骤。spuId
    SpuInfoDescEntity byId = spuInfoDescService.getById(spuId);
    skuItemVo.setDesc(byId);

    //5、获取spu规格参数信息-> 需要依赖第一个一个一个1请求返回值的步骤。spuId和catalogId
    List spuItemAttrGroupVos=productAttrValueService.getProductGroupAttrsBySpuId(spuId, catalogId);
    skuItemVo.setGroupAttrs(spuItemAttrGroupVos);
    //TODO 6，秒杀商品优惠信息

    return skuItemVo;
}

这种循序渐进的调查没有问题，但速度肯定会很慢。如何解决这个问题？答案是使用多线程异步编排。CompletableFuture类。

二、CompletableFuture类介绍

在Java 8中, 一个新的包含50围绕方法类: CompletableFuture，提供了非常强大的异步编程功能。它通过函数编程以回调方式处理计算，并提供转换和组合等方法。

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

runAsync方法不支持返回值。
supplyAsync可以支持返回值。

static CompletableFuture runAsync(Runnable runnable)
public static CompletableFuture runAsync(Runnable runnable, Executor executor)
public static  CompletableFuture supplyAsync(Supplier supplier)
public static  CompletableFuture supplyAsync(Supplier supplier, Executor executor)

注：未指定Executor方法将使用ForkJoinPool.commonPool() 执行异步代码作为其线程池。如果指定了线程池，则使用指定的线程池运行。我们尽力使用自己的线程池。如何创建和配置线程池，请参阅本专栏的最后一篇文章“线程池实战分析”+完整代码。

CompletableFuture 提供了三种类型的线程序列化方法。

thenApply 方法：当一个线程依赖于另一个线程时，它会得到前一个任务返回的结果，并返回当前任务的返回值。
thenAccept方法：消耗处理结果。接收任务的处理结果，并在不返回结果的情况下使用处理。
thenRun方法：只要完成上述任务，就将开始执行。thenRun，在处理完任务后，执行 thenRun后续操作。

public  CompletableFuture thenApply(Function fn)
public  CompletableFuture thenApplyAsync(Function fn)
public  CompletableFuture thenApplyAsync(Function fn, Executor executor)

public CompletionStage thenAccept(Consumer action);
public CompletionStage thenAcceptAsync(Consumer action);
public CompletionStage thenAcceptAsync(Consumer action,Executor executor);

public CompletionStage thenRun(Runnable action);
public CompletionStage thenRunAsync(Runnable action);
public CompletionStage thenRunAsync(Runnable action,Executor executor);

注：Async默认值为异步。这里的异步是指不在当前线程内执行。

3.多线程异步业务流程

货物的详细信息是多个sql，而有些步骤需要使用第一步查询的结果，因此通过异步排列，优化上述购物网站商品详情代码。

@Override // SkuInfoServiceImpl
public SkuItemVo item(Long skuId) throws ExecutionException, InterruptedException {
    SkuItemVo skuItemVo = new SkuItemVo();

    CompletableFuture infoFutrue = CompletableFuture.supplyAsync(() -> {
        //1 sku基本信息
        SkuInfoEntity info = getById(skuId);
        skuItemVo.setInfo(info);
        return info;
    }, executor);
    // 无需获取返回值
    CompletableFuture imageFuture = CompletableFuture.runAsync(() -> {
        //2 sku图片信息
        List images = imagesService.getImagesBySkuId(skuId);
        skuItemVo.setImages(images);
    }, executor);
    // 在1之后
    CompletableFuture saleAttrFuture =infoFutrue.thenAcceptAsync(res -> {
        //3 获取spu销售属性组合 list
        List saleAttrVos = skuSaleAttrValueService.getSaleAttrsBuSpuId(res.getSpuId());
        skuItemVo.setSaleAttr(saleAttrVos);
    },executor);
    // 在1之后
    CompletableFuture descFuture = infoFutrue.thenAcceptAsync(res -> {
        //4 获取spu介绍
        SpuInfoDescEntity spuInfo = spuInfoDescService.getById(res.getSpuId());
        skuItemVo.setDesc(spuInfo);
    },executor);
    // 在1之后
    CompletableFuture baseAttrFuture = infoFutrue.thenAcceptAsync(res -> {
        //5 获取spu规格参数信息
        List attrGroups = attrGroupService.getAttrGroupWithAttrsBySpuId(res.getSpuId(), res.getCatalogId());
        skuItemVo.setGroupAttrs(attrGroups);
    }, executor);

    // 6.查询当前sku是否参加秒杀优惠
    CompletableFuture secKillFuture = CompletableFuture.runAsync(() -> {
        R skuSeckillInfo = seckillFeignService.getSkuSeckillInfo(skuId);
        if (skuSeckillInfo.getCode() == 0) {
            SeckillInfoVo seckillInfoVo = skuSeckillInfo.getData(new TypeReference() {});
            skuItemVo.setSeckillInfoVo(seckillInfoVo);
        }
    }, executor);

    // 等待所有任务完成后再返回
    CompletableFuture.allOf(imageFuture,saleAttrFuture,descFuture,baseAttrFuture,secKillFuture).get();
    return skuItemVo;
}

从那时起，原来的单线程逻辑已经完全优化为多线程执行，完美地提高了效率！
完结撒花o( ￣▽￣ )ブ