CompletableFuture异步和线程池讲解

2021/11/2 6:11:46

本文主要是介绍CompletableFuture异步和线程池讲解,对大家解决编程问题具有一定的参考价值,需要的程序猿们随着小编来一起学习吧!

一、线程回顾

1、初始化线程的 4 种方式 1)、继承 Thread 2)、实现 Runnable 接口 3)、实现 Callable 接口 + FutureTask (可以拿到返回结果,可以处理异常) 4)、线程池 方式 1 和方式 2:主进程无法获取线程的运算结果。不适合当前场景 方式 3:主进程可以获取线程的运算结果,但是不利于控制服务器中的线程资源。可以导致 服务器资源耗尽。 方式 4:通过如下两种方式初始化线程池 Executors.newFiexedThreadPool(3); //或者 new ThreadPoolExecutor(corePoolSize, maximumPoolSize, keepAliveTime, TimeUnit unit, workQueue, threadFactory, handler); 通过线程池性能稳定,也可以获取执行结果,并捕获异常。但是,在业务复杂情况下,一 个异步调用可能会依赖于另一个异步调用的执行结果。   2、线程池的七大参数 corePoolSize:池中一直保持的线程的数量,即使线程空闲。除非设置了 allowCoreThreadTimeOut maximumPoolSize:池中允许的最大的线程数 keepAliveTime:当线程数大于核心线程数的时候,线程在最大多长时间没有接到新任务就会终止释放, 最终线程池维持在 corePoolSize 大小 unit:时间单位 workQueue:阻塞队列,用来存储等待执行的任务,如果当前对线程的需求超过了 corePoolSize 大小,就会放在这里等待空闲线程执行。 threadFactory:创建线程的工厂,比如指定线程名等 handler:拒绝策略,如果线程满了,线程池就会使用拒绝策略。

 

 

运行流程: 1、线程池创建,准备好 core 数量的核心线程,准备接受任务 2、新的任务进来,用 core 准备好的空闲线程执行。 (1) 、core 满了,就将再进来的任务放入阻塞队列中。空闲的 core 就会自己去阻塞队 列获取任务执行 (2) 、阻塞队列满了,就直接开新线程执行,最大只能开到 max 指定的数量 (3) 、max 都执行好了。Max-core 数量空闲的线程会在 keepAliveTime 指定的时间后自 动销毁。最终保持到 core 大小 (4) 、如果线程数开到了 max 的数量,还有新任务进来,就会使用 reject 指定的拒绝策 略进行处理 3、所有的线程创建都是由指定的 factory 创建的。   3、常见的 4 种线程池 newCachedThreadPool 创建一个可缓存线程池,如果线程池长度超过处理需要,可灵活回收空闲线程,若无可回收,则新建线程。   newFixedThreadPool 创建一个定长线程池,可控制线程最大并发数,超出的线程会在队列中等待。   newScheduledThreadPool 创建一个定长线程池,支持定时及周期性任务执行。   newSingleThreadExecutor 创建一个单线程化的线程池,它只会用唯一的工作线程来执行任务,保证所有任务按照指定顺序(FIFO, LIFO, 优先级)执行。   4、开发中为什么使用线程池   降低资源的消耗   通过重复利用已经创建好的线程降低线程的创建和销毁带来的损耗   提高响应速度   因为线程池中的线程数没有超过线程池的最大上限时,有的线程处于等待分配任务的状态,当任务来时无需创建新的线程就能执行   提高线程的可管理性   线程池会根据当前系统特点对池内的线程进行优化处理,减少创建和销毁线程带来的系统开销。无限的创建和销毁线程不仅消耗系统资源,还降低系统的稳定性,使用线程池进行统一分配  

二、CompletableFuture 异步编排

业务场景: 查询商品详情页的逻辑比较复杂,有些数据还需要远程调用,必然需要花费更多的时间。

假如商品详情页的每个查询,需要如下标注的时间才能完成

那么,用户需要 5.5s 后才能看到商品详情页的内容。很显然是不能接受的。 如果有多个线程同时完成这 6 步操作,也许只需要 1.5s 即可完成响应。   Future 是 Java 5 添加的类,用来描述一个异步计算的结果。你可以使用`isDone`方法检查计算是否完成,或者使用`get`阻塞住调用线程,直到计算完成返回结果,你也可以使用`cancel`方法停止任务的执行。   CompletableFuture 和 FutureTask 同属于 Future 接口的实现类,都可以获取线程的执行结果。

 

 

1、创建异步对象 CompletableFuture 提供了四个静态方法来创建一个异步操作。   runXxxx 都是没有返回结果的,supplyXxx 都是可以获取返回结果的 可以传入自定义的线程池,否则就用默认的线程池;
CompletableFuture<Void> future = CompletableFuture.runAsync(() -> {
        System.out.println("当前线程:" + Thread.currentThread().getId());
        int i = 10 / 2;
        System.out.println("运行结果:" + i);
        }, executor);
CompletableFuture<Integer> future = CompletableFuture.supplyAsync(() -> {
        System.out.println("当前线程:" + Thread.currentThread().getId());
        int i = 10 / 0;
        System.out.println("运行结果:" + i);
             return i;
         }, executor);
Integer result = future.get();
  2、计算完成时回调方法 whenComplete 可以处理正常和异常的计算结果,exceptionally 处理异常情况。 whenComplete 和 whenCompleteAsync 的区别: whenComplete:是执行当前任务的线程执行继续执行 whenComplete 的任务。 whenCompleteAsync:是执行把 whenCompleteAsync 这个任务继续提交给线程池 来进行执行。 方法不以 Async 结尾,意味着 Action 使用相同的线程执行,而 Async 可能会使用其他线程 执行(如果是使用相同的线程池,也可能会被同一个线程选中执行)
CompletableFuture<Integer> future = CompletableFuture.supplyAsync(() -> {
             System.out.println("当前线程:" + Thread.currentThread().getId());
             int i = 10 / 0;
             System.out.println("运行结果:" + i);
             return i;
         }, executor).whenComplete((res,exception) -> {
             //虽然能得到异常信息,但是没法修改返回数据
             System.out.println("异步任务成功完成了...结果是:" + res + "异常是:" + exception);
         }).exceptionally(throwable -> {
             //可以感知异常,同时返回默认值
             return 10;
         });

 

3、handle 方法

和 complete 一样,可对结果做最后的处理(可处理异常),可改变返回值。

/**
         * 方法执行完后端处理
         */
         CompletableFuture<Integer> future = CompletableFuture.supplyAsync(() -> {
             System.out.println("当前线程:" + Thread.currentThread().getId());
             int i = 10 / 2;
             System.out.println("运行结果:" + i);
             return i;
         }, executor).handle((result,thr) -> {
             if (result != null) {
                 return result * 2;
             }
             if (thr != null) {
                 System.out.println("异步任务成功完成了...结果是:" + result + "异常是:" + thr);
                 return 0;
             }
             return 0;
         });

 

4、线程串行化方法

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

thenAccept 方法:消费处理结果。接收任务的处理结果,并消费处理,无返回结果。 thenRun 方法:只要上面的任务执行完成,就开始执行 thenRun,只是处理完任务后,执行thenRun 的后续操作 带有 Async 默认是异步执行的。同之前。 以上都要前置任务成功完成。 Function<? super T,? extends U> T:上一个任务返回结果的类型U:当前任务的返回值类型
CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
            System.out.println("当前线程:" + Thread.currentThread().getId());
            int i = 10 / 2;
            System.out.println("运行结果:" + i);
            return i;
        }, executor).thenApplyAsync(res -> {
            System.out.println("任务2启动了..." + res);
            return "Hello" + res;
        }, executor);
        System.out.println("main......end....." + future.get());

 

  5、两任务组合 - 都要完成

 

两个任务必须都完成,触发该任务。

thenCombine:组合两个 future,获取两个 future 的返回结果,并返回当前任务的返回值 thenAcceptBoth:组合两个 future,获取两个 future 任务的返回结果,然后处理任务,没有返回值。 runAfterBoth:组合两个 future,不需要获取 future 的结果,只需两个 future 处理完任务后, 处理该任务。   6、两任务组合 - 一个完成

 

当两个任务中,任意一个 future 任务完成的时候,执行任务。

applyToEither:两个任务有一个执行完成,获取它的返回值,处理任务并有新的返回值。 acceptEither:两个任务有一个执行完成,获取它的返回值,处理任务,没有新的返回值。 runAfterEither:两个任务有一个执行完成,不需要获取 future 的结果,处理任务,也没有返回值。   7、多任务组合

allOf:等待所有任务完成

anyOf:只要有一个任务完成  

三.案例

1.MyThreadConfig.java
//@EnableConfigurationProperties(ThreadPoolConfigProperties.class)
@Configuration
public class MyThreadConfig {


    @Bean
    public ThreadPoolExecutor threadPoolExecutor(ThreadPoolConfigProperties pool) {
        return new ThreadPoolExecutor(
                pool.getCoreSize(),
                pool.getMaxSize(),
                pool.getKeepAliveTime(),
                TimeUnit.SECONDS,
                new LinkedBlockingDeque<>(100000),
                Executors.defaultThreadFactory(),
                new ThreadPoolExecutor.AbortPolicy()
        );
    }

}

 

2.ThreadPoolConfigProperties.java

@ConfigurationProperties(prefix = "gulimall.thread")
@Component
@Data
public class ThreadPoolConfigProperties {

    private Integer coreSize;

    private Integer maxSize;

    private Integer keepAliveTime;


}

 

3.使用异步线程

@Service("skuInfoService")
public class SkuInfoServiceImpl extends ServiceImpl<SkuInfoDao, SkuInfoEntity> implements SkuInfoService {
    @Resource
    private ThreadPoolExecutor executor;

    
    @Override
    public SkuItemVo item(Long skuId) throws ExecutionException, InterruptedException {

        SkuItemVo skuItemVo = new SkuItemVo();

        CompletableFuture<SkuInfoEntity> infoFuture = CompletableFuture.supplyAsync(() -> {
            //1、sku基本信息的获取  pms_sku_info
            SkuInfoEntity info = this.getById(skuId);
            skuItemVo.setInfo(info);
            return info;
        }, executor);


        CompletableFuture<Void> saleAttrFuture = infoFuture.thenAcceptAsync((res) -> {
            //3、获取spu的销售属性组合
            List<SkuItemSaleAttrVo> saleAttrVos = skuSaleAttrValueService.getSaleAttrBySpuId(res.getSpuId());
            skuItemVo.setSaleAttr(saleAttrVos);
        }, executor);


        CompletableFuture<Void> descFuture = infoFuture.thenAcceptAsync((res) -> {
            //4、获取spu的介绍    pms_spu_info_desc
            SpuInfoDescEntity spuInfoDescEntity = spuInfoDescService.getById(res.getSpuId());
            skuItemVo.setDesc(spuInfoDescEntity);
        }, executor);


        CompletableFuture<Void> baseAttrFuture = infoFuture.thenAcceptAsync((res) -> {
            //5、获取spu的规格参数信息
            List<SpuItemAttrGroupVo> attrGroupVos = attrGroupService.getAttrGroupWithAttrsBySpuId(res.getSpuId(), res.getCatalogId());
            skuItemVo.setGroupAttrs(attrGroupVos);
        }, executor);


        // Long spuId = info.getSpuId();
        // Long catalogId = info.getCatalogId();

        //2、sku的图片信息    pms_sku_images
        CompletableFuture<Void> imageFuture = CompletableFuture.runAsync(() -> {
            List<SkuImagesEntity> imagesEntities = skuImagesService.getImagesBySkuId(skuId);
            skuItemVo.setImages(imagesEntities);
        }, executor);

        CompletableFuture<Void> seckillFuture = CompletableFuture.runAsync(() -> {
            //3、远程调用查询当前sku是否参与秒杀优惠活动
            R skuSeckilInfo = seckillFeignService.getSkuSeckilInfo(skuId);
            if (skuSeckilInfo.getCode() == 0) {
                //查询成功
                SeckillSkuVo seckilInfoData = skuSeckilInfo.getData("data", new TypeReference<SeckillSkuVo>() {
                });
                skuItemVo.setSeckillSkuVo(seckilInfoData);

                if (seckilInfoData != null) {
                    long currentTime = System.currentTimeMillis();
                    if (currentTime > seckilInfoData.getEndTime()) {
                        skuItemVo.setSeckillSkuVo(null);
                    }
                }
            }
        }, executor);


        //等到所有任务都完成
        CompletableFuture.allOf(saleAttrFuture,descFuture,baseAttrFuture,imageFuture,seckillFuture).get();

        return skuItemVo;
    }

}

 



这篇关于CompletableFuture异步和线程池讲解的文章就介绍到这儿,希望我们推荐的文章对大家有所帮助,也希望大家多多支持为之网!


扫一扫关注最新编程教程