尚品汇-CompletableFuture异步编排-优化商品详情页（二十五）

news2024/11/15 21:29:09

（1）创建异步对象

（2）计算完成时回调方法

（3）线程串行化与并行化方法

（4）多任务组合

（5）优化商品详情页

（1）创建异步对象

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

没有指定Executor的方法会使用ForkJoinPool.commonPool() 作为它的线程池执行异步代码。

如果指定线程池，则。

- runAsync方法不支持返回值。

- supplyAsync可以支持返回值。

（2）计算完成时回调方法

当CompletableFuture的计算结果完成，或者抛出异常的时候，可以执行特定的Action。主要是下面的方法：

whenComplete可以处理正常或异常的计算结果，exceptionally处理异常情况。

BiConsumer<? super T,? super Throwable>可以定义处理业务

whenComplete 和 whenCompleteAsync 的区别：

whenComplete：是执行当前任务的线程执行继续执行 whenComplete 的任务。

whenCompleteAsync：是执行把 whenCompleteAsync 这个任务继续提交给线程池来进行执行。

方法不以Async结尾，意味着Action使用相同的线程执行，而Async可能会使用其他线程执行（如果是使用相同的线程池，也可能会被同一个线程选中执行）

创建测试类：

没有返回值

使用lambda表达式：

有返回值

回调函数：

添加异常代码：

异常处理回调函数：

回调函数：

（3）线程串行化与并行化方法

thenApply 方法：当一个线程依赖另一个线程时，获取上一个任务返回的结果，并返回当前任务的返回值。

thenAccept方法：消费处理结果。接收任务的处理结果，并消费处理，无返回结果。

这个相当于串行化执行

thenRun方法：只要上面的任务执行完成，就开始执行thenRun，只是处理完任务后，执行 thenRun的后续操作

带有Async默认是异步执行的。这里所谓的异步指的是不在当前线程内执行。

Function<? super T,? extends U>

T：上一个任务返回结果的类型

U：当前任务的返回值类型

让B睡2秒：

并行化：
ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(50, 500, 30, TimeUnit.SECONDS, new ArrayBlockingQueue<>(10000));

// 线程1执行返回的结果：hello
CompletableFuture<String> futureA = CompletableFuture.supplyAsync(() -> "hello");

// 线程2 获取到线程1执行的结果
CompletableFuture<Void> futureB = futureA.thenAcceptAsync((s) -> {
    delaySec(1);
    printCurrTime(s+" 第一个线程");
}, threadPoolExecutor);

CompletableFuture<Void> futureC = futureA.thenAcceptAsync((s) -> {
    delaySec(3);
    printCurrTime(s+" 第二个线程");
}, threadPoolExecutor);

private static void printCurrTime(String str) {
    System.out.println(str);
}

private static void delaySec(int i) {
    try {
        Thread.sleep(i*1000);
    } catch (InterruptedException e) {
        e.printStackTrace();
    }
}

（4）多任务组合

public static CompletableFuture<Void> allOf(CompletableFuture<?>... cfs);

public static CompletableFuture<Object> anyOf(CompletableFuture<?>... cfs);

allOf：等待所有任务完成

anyOf：只要有一个任务完成

（5）优化商品详情页

@Service
public class ItemServiceImpl implements ItemService {

    @Autowired
    private ProductFeignClient productFeignClient;
    

    @Autowired
    private ThreadPoolExecutor threadPoolExecutor;

    @Override
    public Map<String, Object> getBySkuId(Long skuId) {

        Map<String, Object> result = new HashMap<>();

        // 通过skuId 查询skuInfo  获取异步对象，需要返回
        CompletableFuture<SkuInfo> skuCompletableFuture = CompletableFuture.supplyAsync(() -> {
            SkuInfo skuInfo = productFeignClient.getSkuInfo(skuId);
            // 保存skuInfo
            result.put("skuInfo", skuInfo);
            return skuInfo;
        }, threadPoolExecutor);


        // 销售属性-销售属性值回显并锁定  依赖skuCompletableFuture 
        CompletableFuture<Void> spuSaleAttrCompletableFuture = skuCompletableFuture.thenAcceptAsync(skuInfo -> {
            List<SpuSaleAttr> spuSaleAttrList = productFeignClient.getSpuSaleAttrListCheckBySku(skuInfo.getId(), skuInfo.getSpuId());

            // 保存数据
            result.put("spuSaleAttrList", spuSaleAttrList);
        }, threadPoolExecutor);

        //根据spuId 查询map 集合属性  依赖skuCompletableFuture 
        // 获取商品切换数据
        CompletableFuture<Void> skuValueIdsMapCompletableFuture = skuCompletableFuture.thenAcceptAsync(skuInfo -> {
            Map skuValueIdsMap = productFeignClient.getSkuValueIdsMap(skuInfo.getSpuId());

            String valuesSkuJson = JSON.toJSONString(skuValueIdsMap);
            // 保存valuesSkuJson
            result.put("valuesSkuJson", valuesSkuJson);
        }, threadPoolExecutor);



        //获取三级分类信息  依赖skuCompletableFuture 
        CompletableFuture<Void> categoryViewCompletableFuture = skuCompletableFuture.thenAcceptAsync(skuInfo -> {
            BaseCategoryView categoryView = productFeignClient.getCategoryView(skuInfo.getCategory3Id());

            //分类信息
            result.put("categoryView", categoryView);
        }, threadPoolExecutor);



    //  获取海报数据   依赖skuCompletableFuture 
CompletableFuture<Void> spuPosterListCompletableFuture = skuInfoCompletableFuture.thenAcceptAsync(skuInfo -> {
    //  spu海报数据
    List<SpuPoster> spuPosterList = productFeignClient.findSpuPosterBySpuId(skuInfo.getSpuId());
    result.put("spuPosterList", spuPosterList);
},threadPoolExecutor);


  //获取商品最新价格   获取新的异步对象，不需要返回
        CompletableFuture<Void> skuPriceCompletableFuture = CompletableFuture.runAsync(() -> {
            BigDecimal skuPrice = productFeignClient.getSkuPrice(skuId);
            result.put("price", skuPrice);
        }, threadPoolExecutor);



//  获取sku平台信息 ，即规格数据   获取新的异步对象
CompletableFuture<Void> skuAttrListCompletableFuture = CompletableFuture.runAsync(() -> {
    List<BaseAttrInfo> attrList = productFeignClient.getAttrList(skuId);
    //  使用拉姆达表示
    List<Map<String, String>> skuAttrList = attrList.stream().map((baseAttrInfo) -> {
        Map<String, String> attrMap = new HashMap<>();
        attrMap.put("attrName", baseAttrInfo.getAttrName());
        attrMap.put("attrValue", baseAttrInfo.getAttrValueList().get(0).getValueName());
        return attrMap;
    }).collect(Collectors.toList());
    result.put("skuAttrList", skuAttrList);
},threadPoolExecutor);




    //多任务组合  --所有异步任务执行完成才是完成
       
        CompletableFuture.allOf(skuCompletableFuture, spuSaleAttrCompletableFuture, skuValueIdsMapCompletableFuture,skuPriceCompletableFuture, categoryViewCompletableFuture,spuPosterListCompletableFuture,skuAttrListCompletableFuture ).join();
        return result;
    }
}

创建一个线程池的配置类：

ThreadPoolConfig

package com.atguigu.gmall.item.config;
@Configuration
public class ThreadPoolConfig {
    @Bean
    public ThreadPoolExecutor threadPoolExecutor(){
        /**
         * 核心线程数
         * 拥有最多线程数
         * 表示空闲线程的存活时间
         * 存活时间单位
         * 用于缓存任务的阻塞队列
         * 省略：
         *  threadFactory：指定创建线程的工厂
         *  handler：表示当workQueue已满，且池中的线程数达到maximumPoolSize时，线程池拒绝添加新任务时采取的策略。
         */
        return new ThreadPoolExecutor(50,500,30, TimeUnit.SECONDS,new ArrayBlockingQueue<>(10000));
    }
}