当前位置：首页 > news >正文

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

news 来源：原创 2024/9/22 13:43:27

（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 {@Autowiredprivate ProductFeignClient productFeignClient;@Autowiredprivate ThreadPoolExecutor threadPoolExecutor;@Overridepublic Map<String, Object> getBySkuId(Long skuId) {Map<String, Object> result = new HashMap<>();// 通过skuId 查询skuInfo  获取异步对象，需要返回CompletableFuture<SkuInfo> skuCompletableFuture = CompletableFuture.supplyAsync(() -> {SkuInfo skuInfo = productFeignClient.getSkuInfo(skuId);// 保存skuInforesult.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);// 保存valuesSkuJsonresult.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 {@Beanpublic ThreadPoolExecutor threadPoolExecutor(){/*** 核心线程数* 拥有最多线程数* 表示空闲线程的存活时间* 存活时间单位* 用于缓存任务的阻塞队列* 省略：*  threadFactory：指定创建线程的工厂*  handler：表示当workQueue已满，且池中的线程数达到maximumPoolSize时，线程池拒绝添加新任务时采取的策略。*/return new ThreadPoolExecutor(50,500,30, TimeUnit.SECONDS,new ArrayBlockingQueue<>(10000));}
}