其他分享
首页 > 其他分享> > CompletableFuture 异步编排

CompletableFuture 异步编排

作者:互联网

 

 

1、创建异步对象

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

1 #runAsync 无返回值
2 public static CompletableFuture<Void> runAsync(Runable runable)
3 public static CompletableFuture<Void> runAsync(Runnable runnable, Executor executor)
4 
5 #supplyAsync 有返回值
6 public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier)
7 public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier, Executor executor)

2、Completable启动异步任务

runAsync(),无返回值

 1 package com.lian.gulimall.search.thread;
 2 
 3 import java.util.concurrent.CompletableFuture;
 4 import java.util.concurrent.ExecutorService;
 5 import java.util.concurrent.Executors;
 6 
 7 public class ThreadTestAsyn {
 8 
 9     //创建线程池
10     public static ExecutorService executor = Executors.newFixedThreadPool(10);
11     
12     public static void main(String[] args) throws Exception {
13         //1、Async创建异步对象,没返回值
14         CompletableFuture<Void> future = CompletableFuture.runAsync(() -> {
15             //异步任务内容
16             System.out.println("current thread: " + Thread.currentThread().getId());
17             int i = 10 / 2;
18            System.out.println("run result: " + i);
19         }, executor);
20         System.out.println("main...end..");
21     }
22 }

supplyAsync(),有返回值

 1 package com.lian.gulimall.search.thread;
 2 
 3 import java.util.concurrent.CompletableFuture;
 4 import java.util.concurrent.ExecutorService;
 5 import java.util.concurrent.Executors;
 6 
 7 public class ThreadTestAsyn {
 8 
 9     //创建线程池
10     public static ExecutorService executor = Executors.newFixedThreadPool(10);
11     
12     public static void main(String[] args) throws Exception {
13     
14         //2、Supply,有返回值
15         CompletableFuture<Integer> future = CompletableFuture.supplyAsync(() -> {
16             System.out.println("current thread: " + Thread.currentThread().getId());
17             int i = 10 / 2;
18             System.out.println("run result: " + i);
19             return i;
20         }, executor);
21         Integer integer = future.get();
22         System.out.println(integer);
23         System.out.println("main...end..");
24     }
25 }

3、Completable异步完成回调与异常感知

whenComplete:返回结果和异常
exceptionally:如果出现异常,返回默认值

 1 package com.lian.gulimall.search.thread;
 2 
 3 import java.util.concurrent.CompletableFuture;
 4 import java.util.concurrent.ExecutorService;
 5 import java.util.concurrent.Executors;
 6 
 7 public class ThreadTestAsyn {
 8 
 9     //创建线程池
10     public static ExecutorService executor = Executors.newFixedThreadPool(10);
11     
12     public static void main(String[] args) throws Exception {
13         //1、Async创建异步对象,没返回值
14 //        CompletableFuture<Void> future = CompletableFuture.runAsync(() -> {
15 //            //异步任务内容
16 //            System.out.println("current thread: " + Thread.currentThread().getId());
17 //            int i = 10 / 2;
18 //            System.out.println("run result: " + i);
19 //        }, executor);
20 
21         //2、Supply,有返回值
22         CompletableFuture<Integer> future = CompletableFuture.supplyAsync(() -> {
23             System.out.println("current thread: " + Thread.currentThread().getId());
24             int i = 10 / 0;
25             System.out.println("run result: " + i);
26             return i;
27             //参数(结果,异常),虽然能得到异常信息,但是没法修改返回数据
28         }, executor).whenComplete((res,exception)->{
29             System.out.println("async completed...result is: "+res+",exception is: "+exception);
30         //如果结果出现异常,就给一个默认返回值,感知异常,同时返回默认值
31         }).exceptionally((t)->{
32             return 4;
33         });
34         //获取异步结果
35         Integer integer = future.get();
36         System.out.println(integer);
37         System.out.println("main...end..");
38     }
39 }

4、handle

可对结果做异常处理, 可改变返回值

 1 package com.lian.gulimall.search.thread;
 2 
 3 import java.util.concurrent.CompletableFuture;
 4 import java.util.concurrent.ExecutorService;
 5 import java.util.concurrent.Executors;
 6 
 7 public class ThreadTestAsyn {
 8 
 9     //创建线程池
10     public static ExecutorService executor = Executors.newFixedThreadPool(10);
11     public static void main(String[] args) throws Exception {
12         //1、Async创建异步对象,没返回值
13         //方法执行完成后得处理,无论成功与失败
14         CompletableFuture<Integer> future = CompletableFuture.supplyAsync(() -> {
15             System.out.println("current thread: " + Thread.currentThread().getId());
16             int i = 10 / 4;
17             System.out.println("run result: " + i);
18             return i;
19             //参数(结果,异常),虽然能得到异常信息,但是没法修改返回数据
20         }, executor).handle((res,thr)->{
21             //如果返回结果不为空,就将结果×2
22             if (res != null){
23                 return res*2;
24             }
25             //如果异常不为空,就返回0
26             if (thr != null) {
27                 return 0;
28             }
29             //如果两个都不走就返回0
30             return 0;
31         });
32         //获取异步结果
33         Integer integer = future.get();
34         System.out.println(integer);
35     }
36 }

5、CompletableFuture线程串行化

 

 

 

1、thenRunAsync,不能获取到上一步的执行结果,无返回值

 1 public class ThreadTestAsyn {
 2 
 3     //创建线程池
 4     public static ExecutorService executor = Executors.newFixedThreadPool(10);
 5     public static void main(String[] args) throws Exception {
 6     //thenRun 不能获取到上一步的执行结果,无返回值
 7         CompletableFuture<Void> future = CompletableFuture.supplyAsync(() -> {
 8             System.out.println("current thread: " + Thread.currentThread().getId());
 9             int i = 10 / 4;
10             System.out.println("run result: " + i);
11             return i;
12             //参数(结果,异常),虽然能得到异常信息,但是没法修改返回数据
13         }, executor).thenRunAsync(() -> {
14             System.out.println("task2 startting");
15         }, executor);
16    }

2、thenAccept 能接收上一步的执行结果,但是无返回值

 1 public class ThreadTestAsyn {
 2 
 3     //创建线程池
 4     public static ExecutorService executor = Executors.newFixedThreadPool(10);
 5     public static void main(String[] args) throws Exception {
 6            //thenAccept 能接收上一步的执行结果,但是无返回值
 7         CompletableFuture.supplyAsync(() -> {
 8             System.out.println("current thread: " + Thread.currentThread().getId());
 9             int i = 10 / 4;
10             System.out.println("run result: " + i);
11             return i;
12             //参数(结果,异常),虽然能得到异常信息,但是没法修改返回数据
13         }, executor).thenAcceptAsync((res)->{
14             System.out.println("task2 startting"+res);
15         },executor);
16    }

3、thenApply 既能接收上一步的执行结果,还有返回值

 1 public class ThreadTestAsyn {
 2 
 3     //创建线程池
 4     public static ExecutorService executor = Executors.newFixedThreadPool(10);
 5     public static void main(String[] args) throws Exception {
 6         //thenApply 既能接收上一步的执行结果,还有返回值
 7         CompletableFuture<String> future2 = CompletableFuture.supplyAsync(() -> {
 8             System.out.println("current thread: " + Thread.currentThread().getId());
 9             int i = 10 / 4;
10             System.out.println("run result: " + i);
11             return i;
12             //参数(结果,异常),虽然能得到异常信息,但是没法修改返回数据
13         }, executor).thenApplyAsync((res) -> {
14             System.out.println("task2 startting" + res);
15             return "hello" + res;
16         }, executor);
17         System.out.println("main...end...."+future2.get());
18     }

两任务组合-都要完成

 

 

 1 public class ThreadTestAsyn {
 2 
 3     //创建线程池
 4     public static ExecutorService executor = Executors.newFixedThreadPool(10);
 5     public static void main(String[] args) throws Exception {
 6         CompletableFuture<Integer> future01 = CompletableFuture.supplyAsync(() -> {
 7             System.out.println("task1 thread start: " + Thread.currentThread().getId());
 8             int i = 10 / 4;
 9             System.out.println("task1 end");
10             return i;
11         }, executor);
12 
13         CompletableFuture<String> future02 = CompletableFuture.supplyAsync(() -> {
14             System.out.println("task2 thread start: " + Thread.currentThread().getId());
15             System.out.println("task2 end");
16             return "hello";
17         }, executor);
18 
19         //组合两个future,不需要获取future的结果,只需两个future处理完任务后,处理该任务
20         future01.runAfterBothAsync(future02,()->{
21             System.out.println("task3 start...");
22         },executor);
23 
24         //组合两个future,获取两个future任务的返回结果,然后处理任务,没有返回值
25         future01.thenAcceptBothAsync(future02,(t,u)->{
26             System.out.println("task3 start...,result is:"+t+"--》"+u);
27         },executor);
28 
29         //组合两个future,获取两个future任务的返回结果,并返回当前任务的返回值
30         CompletableFuture<String> future = future01.thenCombineAsync(future02, (f1, f2) -> {
31             return f1 + ":" + f2 + "--> haha";
32         }, executor);
33         System.out.println("main..end"+future.get());
34 
35     }

两任务组合-其中一个完成

 

 

 1 public class ThreadTestAsyn {
 2 
 3     //创建线程池
 4     public static ExecutorService executor = Executors.newFixedThreadPool(10);
 5     public static void main(String[] args) throws Exception {
 6 
 7     CompletableFuture<Object> future01 = CompletableFuture.supplyAsync(() -> {
 8             System.out.println("task1 thread start: " + Thread.currentThread().getId());
 9             int i = 10 / 4;
10             System.out.println("task1 end");
11             return i;
12         }, executor);
13 
14         CompletableFuture<Object> future02 = CompletableFuture.supplyAsync(() -> {
15             System.out.println("task2 thread start: " + Thread.currentThread().getId());
16             System.out.println("task2 end");
17             return "hello";
18         }, executor);
19 
20         //两个任务,只要有一个完成,就执行任务3,不感知结果,也没有返回值
21         future01.runAfterEitherAsync(future02,()->{
22             System.out.println("task3 start...");
23         },executor);
24 
25         //只接收上一次任务的返回结果,没有返回值,感知结果,没有返回值
26         future01.acceptEitherAsync(future02,(res)->{
27             System.out.println("task3 start..."+res);
28         },executor);
29 
30         //感知结果,也有返回值
31         future01.applyToEitherAsync(future02,(res)->{
32             System.out.println("task3 start..."+res);
33             return res.toString() + "--> haha";
34         },executor);
35     }

多任务组合

 

 

 1 public class ThreadTestAsyn {
 2 
 3     //创建线程池
 4     public static ExecutorService executor = Executors.newFixedThreadPool(10);
 5     public static void main(String[] args) throws Exception {
 6     CompletableFuture<String> futureImg = CompletableFuture.supplyAsync(() -> {
 7             System.out.println("product info");
 8             return "hello.jpg";
 9         }, executor);
10 
11         CompletableFuture<String> futureAttr = CompletableFuture.supplyAsync(() -> {
12             System.out.println("product info");
13             return "black 256g";
14         }, executor);
15 
16         CompletableFuture<String> futureDes = CompletableFuture.supplyAsync(() -> {
17             System.out.println("product info");
18             return "huawei";
19         }, executor);
20 
21         CompletableFuture<Void> future = CompletableFuture.allOf(futureImg, futureAttr, futureDes);
22         future.get();  //get方法阻塞等待所有结果完成
23 
24         System.out.println(futureImg.get()+futureAttr.get()+futureDes.get());
25     }

五、案例演示

在商品详情页时,为了提升效率,节省时间,采用异步编排方式

第1步、配置属性和类绑定赋值

package com.lian.gulimall.product.config;

import lombok.Data;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.stereotype.Component;

@ConfigurationProperties(prefix = "gulimall.thread")
@Component
@Data
public class ThreadPoolConfigProperties {
    private Integer coreSize;      //核心线程数
    private Integer maxSize;       //最大线程数
    private Integer keepAliveTime; //存活时间
}

第2步、配置application.properties,给实体类赋值

gulimall.thread.core-size=20
gulimall.thread.max-size=200
gulimall.thread.keep-alive-time=10

第3步、自定义线程池

package com.lian.gulimall.product.config;

import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * 自定义线程池
 * 注入到spring容器就不需要配置 自动导入配置属性注解了
 */
//@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<>(10000),
                Executors.defaultThreadFactory(),
                new ThreadPoolExecutor.AbortPolicy());
    }
}

第4步、商品详情页的控制层

package com.lian.gulimall.product.web;

import com.lian.gulimall.product.service.SkuInfoService;
import com.lian.gulimall.product.vo.SkuItemVo;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Controller;
import org.springframework.ui.Model;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;

/**
 * 详情页
 */
@Controller
public class ItemController {

    @Autowired
    SkuInfoService skuInfoService;

    /**
     * 展示当前sku的详情
     * @return
     */
    @GetMapping("/{skuId}.html")
    public String skuItem(@PathVariable("skuId") Long skuId, Model model) throws Exception{
        System.out.println("准备查询"+skuId+"的详情");
        //根据skuId查询到所有的属性都封装到 SkuItemVo 中
        SkuItemVo vos = skuInfoService.item(skuId);
        model.addAttribute("item",vos);
        return "item";
    }
}

第5步、业务层

@Service("skuInfoService")
public class SkuInfoServiceImpl extends ServiceImpl<SkuInfoDao, SkuInfoEntity> implements SkuInfoService {

    @Autowired
    SkuImagesService skuImagesService; //图片的sku

    @Autowired
    ProductAttrValueService productAttrValueService; //商品属性值

    @Autowired
    SpuInfoDescService spuInfoDescService; //商品描述

    @Autowired
    AttrGroupService attrGroupService;  //组信息

    @Autowired
    SkuSaleAttrValueService skuSaleAttrValueService;

    @Autowired
    ThreadPoolExecutor executor;   //注入线程池

    @Override
    public SkuItemVo item(Long skuId) throws Exception{
    
        //商品详情页返回数据都封装到 SkuItemVo
        SkuItemVo skuItemVo = new SkuItemVo();

        /**
         * 使用异步编排,节省时间提升效率,一起执行不阻塞等待
         * 第一种:supplyAsync 有返回值,其他任务可以用
         * 开启一个异步任务,创建异步对象
         * infoFuture 任务完成后,saleAttrFuture、descFuture、baseAttrFuture 才开始执行,因为他们都需要依赖任务1的数据结果
         */
        CompletableFuture<SkuInfoEntity> infoFuture = CompletableFuture.supplyAsync(() -> {
            //1、sku基本信息获取,标题、副标题、价格等 pms_sku_info
            SkuInfoEntity info = baseMapper.selectById(skuId);
            skuItemVo.setInfo(info);
            //因为其他任务要用基本信息,所以我们返回基本信息
            return info;
            //executor代表要放到自己的线程池里面
        }, executor);

        //第二种:接下来接收任务的返回结果,accept只是接收上一个任务的结果,自己不返回结果
        CompletableFuture<Void> saleAttrFuture = infoFuture.thenAcceptAsync((res) -> {
            //执行第二个任务
            //3、获取spu的销售属性组合
            List<SkuItemSaleAttrVo> saleAttrVos = skuSaleAttrValueService.getSaleAttrsBySpuId(res.getSpuId());
            skuItemVo.setSaleAttr(saleAttrVos);
        });

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

        //继续执行任务,任务3、4、5都依赖任务1的结果 获取spuId
        CompletableFuture<Void> baseAttrFuture = infoFuture.thenAcceptAsync((res) -> {
            //5、获取spu的规格参数信息
            List<SpuItemAttrGroupVo> attrGroupVos = attrGroupService.getAttrGroupWithAttrsBySpuId(res.getSpuId(), res.getCatalogId());
            skuItemVo.setGroupAttrs(attrGroupVos);
        }, executor);

        /**
         * 任务2 不需要依赖任务1提供的结果数据,所以不需要等待任务1完成,直接和任务1同步执行,所以自己也开启一个异步任务
         * runAsync 代表不需要返回结果,因为也没有其他任务需要依赖任务2的数据
         */
        CompletableFuture<Void> imageFuture = CompletableFuture.runAsync(() -> {
            //2、sku图片信息 pms_sku_images
            List<SkuImagesEntity> images = skuImagesService.getImagesBySkuId(skuId);
            skuItemVo.setImages(images);
        }, executor);

        /**
         * 等待所有任务都完成,因为每一个任务都是在给 vo 中封装数据
         * get()方法就是阻塞等待所有任务都执行完
         * infoFuture 也可以不写,因为别人是依赖她的,如果别人都执行完了,那么她肯定也执行完了
         */
        CompletableFuture.allOf(infoFuture, saleAttrFuture, descFuture, baseAttrFuture, imageFuture).get();

        return skuItemVo;
    }

 

标签:异步,System,编排,CompletableFuture,executor,println,public,out
来源: https://www.cnblogs.com/caicz/p/16525466.html