【业务功能篇93】微服务-springcloud-多线程-异步处理-异步编排-CompletableFutrue-实战运用

ゝ一纸荒年。 2024-03-26 22:24 55阅读 0赞

### 异步处理编排 ###

我们可以在商品详细信息查询的位置实现CompletableFuture的异步编排处理。

> 根据业务分析：3.4.5数据接口的入参信息需要来源于1数据接口的返回信息，也就是skuid 所以可以设计 1 3 4 5 串行线程 ，而 3 4 5依赖1 ，需要等1 执行完成  
> 2接口数据没有关系，所以就与上面的线程是并行关系  
> 所以就是 1 2 接口数据是并行执行 而3 4 5与1 是串行执行 进行异步编排设计，提高接口访问效率  
> 这样处理后，原本5个接口 假如每个接口都需要2s,那么一共就是要10s了，现在就是缩减了，因为1与2接口是并行执行，而3 4 5也是并行的。所以时间上肯定是能较大程度缩减的  
> ![image.png][]

先定义线程池

// 第一种获取的方式
     //ExecutorService service = Executors.newFixedThreadPool(10);
     // Executors.newCachedThreadPool()
     // Executors.newScheduledThreadPool() 定时任务的线程池
     // Executors.newSingleThreadExecutor(); // 线程池中永远只有一个线程去处理，并发的情况下会被变为同步的处理
    
     // 第二种方式： 直接new ThreadPoolExecutor()对象，并且手动的指定对应的参数
     // corePoolSize:线程池的核心线程数量 线程池创建出来后就会 new Thread() 5个
     // maximumPoolSize:最大的线程数量，线程池支持的最大的线程数
     // keepAliveTime:存活时间，当线程数大于核心线程，空闲的线程的存活时间 8-5=3
     // unit:存活时间的单位
     // BlockingQueue<Runnable> workQueue:阻塞队列 当线程数超过了核心线程数据，那么新的请求到来的时候会加入到阻塞的队列中
     // new LinkedBlockingQueue<>() 默认队列的长度是 Integer.MAX 那这个就太大了，所以我们需要指定队列的长度
     // threadFactory:创建线程的工厂对象
     // RejectedExecutionHandler handler:当线程数大于最大线程数的时候会执行的淘汰策略
    @Configuration
    public class MyThreadPoolConfig {
        
    
        @Bean
        public ThreadPoolExecutor threadPoolExecutor()
        {
        
            return new ThreadPoolExecutor(20
                    ,200
                    ,10
                    , TimeUnit.SECONDS
                    ,new LinkedBlockingQueue(10000)
                    , Executors.defaultThreadFactory()
                    ,new ThreadPoolExecutor.AbortPolicy()
            );
        }
    }

具体的编排处理

//注入线程配置类的线程池  用于服务方法接口的线程创建所需
    @Autowired
        ThreadPoolExecutor threadPoolExecutor;
    
    @Override
        public SpuItemVO item(Long skuId) throws ExecutionException, InterruptedException {
        
            SpuItemVO vo = new SpuItemVO();
            CompletableFuture<SkuInfoEntity> skuInfoFuture = CompletableFuture.supplyAsync(() -> {
        
                // 1.sku的基本信息 pms_sku_info
                SkuInfoEntity skuInfoEntity = getById(skuId);
                vo.setInfo(skuInfoEntity);
    
                return skuInfoEntity;
            }, threadPoolExecutor);
    
            //thenAcceptAsync 等待前面的异步任务完成后 获取其返回值做相应的业务处理，该任务没返回值，因为主要是获取前面的sku信息做入参 不需要有返回值
            CompletableFuture<Void> saleFuture = skuInfoFuture.thenAcceptAsync((res) -> {
        
                // 3.获取spu中的销售属性的组合
                List<SkuItemSaleAttrVo> saleAttrs = skuSaleAttrValueService
                        .getSkuSaleAttrValueBySpuId(res.getSpuId());
                vo.setSaleAttrs(saleAttrs);
            }, threadPoolExecutor);
    
            CompletableFuture<Void> spuFuture = skuInfoFuture.thenAcceptAsync((res) -> {
        
                // 4.获取SPU的介绍
                SpuInfoDescEntity spuInfoDescEntity = spuInfoDescService.getById(res.getSpuId());
                vo.setDesc(spuInfoDescEntity);
            }, threadPoolExecutor);
    
    
            CompletableFuture<Void> groupFuture = skuInfoFuture.thenAcceptAsync((res) -> {
        
                // 5.获取SPU的规格参数
                List<SpuItemGroupAttrVo> groupAttrVo = attrGroupService
                        .getAttrgroupWithSpuId(res.getSpuId(), res.getCatalogId());
                vo.setBaseAttrs(groupAttrVo);
            }, threadPoolExecutor);
    
            //runAsync 异步任务无需返回值
            CompletableFuture<Void> imageFuture = CompletableFuture.runAsync(() -> {
        
                // 2.sku的图片信息pms_sku_images
                List<SkuImagesEntity> images = skuImagesService.getImagesBySkuId(skuId);
                vo.setImages(images);
            }, threadPoolExecutor);
    
            CompletableFuture<Void> seckillFuture = CompletableFuture.runAsync(() -> {
        
                // 查询商品的秒杀活动
                R r = seckillFeignService.getSeckillSessionBySkuId(skuId);
                if(r.getCode() == 0){
        
                    SeckillVO seckillVO = JSON.parseObject(r.get("data").toString(),SeckillVO.class);
                    vo.setSeckillVO(seckillVO);
                }
            }, threadPoolExecutor);
    
    
            //最后  allOF等待阻塞上面的异步任务完成后再返回值，由于3 4 5是依赖于 1skuInfoFuture异步任务完成，所以无需将1写入也可以
            CompletableFuture.allOf(saleFuture,spuFuture,imageFuture,groupFuture,seckillFuture).get();
            return vo;
        }

[image.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/13/d655f6be78a84f5ab2845a5495480851.png