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

【45-线程的实现方式-线程池的创建方式-线程池的执行顺序-CompletableFutrue异步处理】

news 来源：原创 2024/5/14 19:47:58

一.知识回顾

【0.三高商城系统的专题专栏都帮你整理好了，请点击这里！】
【1-系统架构演进过程】
【2-微服务系统架构需求】
【3-高性能、高并发、高可用的三高商城系统项目介绍】
【4-Linux云服务器上安装Docker】
【5-Docker安装部署MySQL和Redis服务】
【6-Git安装与配置过程、Gitee码云上创建项目、IDEA关联克隆的项目】
【7-创建商城系统的子模块并将修改后的信息使用Git提交到Gitee上】
【8-数据库表结构的创建&后台管理系统的搭建】
【9-前端项目的搭建部署、Node安装、VSCode安装】
【10-Node的安装以及全局环境变量的相关配置&解决启动报错的问题(1.Error: Cannot find module ‘fs/promises)(2.npm安装node-sass报错)】
【11-导入人人generator项目并自动生成相关的文件&商品子模块的调试&公共模块common子模块的抽离与实现&Lombok插件的安装】
【12-商品子模块整合MyBatisPlus技术&其它模块通过generator的自动生成与补充完善】
【13-项目中微服务组件的学习-SpringCloudAlibaba微服务生态体系的学习&SpringCloudAlibaba的依赖管理&项目中SpringBoot和SpringCloud版本的统一】
【14-微服务的注册中心与配置中心Nacos&Windows操作系统上安装Nacos和Linux操作系统上用Docker中安装Nacos&每个子项目模块使用Nacos进行服务注册与发现】
【15-项目中服务的远程调用之OpenFeign&订单模块与商品模块集成使用OpenFeign的案例】
【16-配置中心之Nacos的基本使用&Nacos服务之命令空间、Nacos服务之配置组、Nacos服务之配置拆分】
【17-微服务网关之Spring Cloud Gateway&Spring Cloud Gateway网关服务搭建】
【18-业务开发-基础业务-商品模块-分类管理-前后端管理系统的启动-为分类管理表增加数据-Json插件的下载-返回具有层级目录、父子关系结构的数据】
【19-业务开发-基础业务-商品模块-分类管理-管理系统新建菜单-后端项目renren注册到Nacos注册中心和配置中心去-项目gateway网关模块的搭建-浏览器的同源策略与解决跨域问题实操案例】
【20-业务开发-基础业务-商品模块-分类管理-前端展示后端具有层级关系的目录数据-商品系统三级分类的逻辑删除前后端代码实现】
【21-业务开发-基础业务-商品模块-分类管理-商品系统三级分类的新增类别前后端代码实现-商品系统三级分类的更新类别前后端代码实现-之前错误的Bug修正】
【22-业务开发-基础业务-商品模块-分类管理-商品系统三级分类拖拽页面的功能-前后端代码的逻辑实现-访问测试-拖拽开关的开启和关系-批量更新拖拽数据-批量删除选定数据】
【23-业务开发-基础业务-品牌管理-品牌管理项目搭建-品牌管理实现的增删改查操作测试-后端数据显示状态使用前端组件开关按钮展示-以及数据处理以及测试】
【24-业务开发-基础业务-品牌管理-图片管理-阿里云OSS服务开通和使用-阿里云OSS服务API使用-SpringCloudAlibaba OSS服务的使用】
【25-业务开发-基础业务-品牌管理-图片管理-图片上传方式的三种实现方式-第三方公共服务模块集成到项目中-服务端生成签名实战】
【26-业务开发-基础业务-品牌管理-图片管理-上传图片功能实现-基于阿里云OSS服务-解决跨域问题-设置跨域规则-修改ACL权限为公共读】
【27-业务开发-基础业务-品牌管理-图片管理-添加修改品牌信息并显示图片-前端数据校验-后端数据JSR303校验实现-统一异常处理-自定义响应编码规则-分组校验-自定义校验注解-项目Bug解决】
【28-业务开发-基础业务-属性管理-SKU和SPU基本概念-SKU和SPU关联关系-属性实体之间的关联关系-批量菜单创建】
【29-业务开发-基础业务-属性管理-属性组业务逻辑开发-页面布局-三级分类组件功能-属性组表单-父子组件传值-属性组数据展示-属性组数据添加-属性组数据修改-前后端项目整合交互测试】
【30-业务开发-基础业务-品牌管理-分类维护-解决分类维护业务开发中的一个Bug-品牌管理-分页插件-分页功能的逻辑实现-品牌管理-检索条件模糊查询品牌管理-增加更新操作中排序字段检验还是存在问题】
【31-业务开发-基础业务-品牌管理-级联类别信息业务功能实现-品牌管理和商品分类管理俩者业务关联出现数据冗余，导致数据不同步的问题-开启事务-项目测试】
【32-业务开发-基础业务-规格参数-保存数据-查询数据-更新操作之数据回显展示-更新操作-前后端项目交互整合与测试-总结收获】
【33-业务开发-基础业务-规格参数-销售属性-多表之间的关联增删改查操作-前后端项目交互整合与测试-Cannot read property ‘publish‘ of undefined】
【34-业务开发-基础业务-属性组和基本属性-属性组和基本属性建立关联-属性组和基本属性解除关联-未关联属性查询-确认新增】
【35-业务开发-基础业务-商品服务-新增商品-会员模块服务-mall-member-会员模块数据维护-规格参数维护-前端项目Bug解决-PubSub依赖缺失】
【36-业务开发-基础业务-商品服务SPU-前后端处理商品数据Json-发布商品前后端业务逻辑-feign服务远程调用-DTO数据传输对象-商品服务的检索-商品管理的检索项目中修改更正完善逻辑操作】
【37-业务开发-基础业务-库存管理- 仓库模块Nacos注册中心的配置-Gateway网关配置-仓库维护的增删改查-商品库存管理-采购流程-采购需求维护-采购需求合并-领取采购单完成采购操作】
【插入------>ElasticSearch专栏相关的知识内容都整理好了，在这里哟！】
【38-商品上架功能结合ElasticSearch全文检索的流程-商品ES关系映射模型&Docker安装ik分词器-实现上架功能复杂的逻辑实现-Postman+Kibana访问测试】
【39-商品整合thymeleaf模板引擎-商城用户端的实现逻辑-部署devtools工具依赖-商品后台-三级分类逻辑分析实现-Docker 安装部署Nginx-Nginx对网关实现反向代理负载均衡】
【40-系统性能压力测试基本概念-相关性能指标HPS&TPS&QPS&RT-安装Jmeter教程-JMeter测试流程-线程组-取样器-监视器-测试商城首页-JMeter Address 占用的问题】
【41-系统性能压力测试优化-JVM知识回顾-jconsole和jvisualvm-jvisualvm安装Visual GC插件-Nginx压力测试- 网关gateway压测-Nginx实现动静分离】
【42-缓存的基本概念-是否使用缓存的场景-本地缓存-分布式缓存-项目中整合Redis-修改三级分类逻辑代码+加入缓存-三级分类加入缓存后压力测试-缓存穿透-缓存雪崩-缓存击穿】
【43-本地锁-分布式锁概念原理-分布式锁解决方案-Redis实现分布式锁-Redisson分布式锁-项目整合Redisson-缓存数据一致性问题-解决缓存一致性的方案-SpringCache缓存】
【44-商城检索服务的搭建-页面跳转调整-elasticsearch检索服务前后端响应的VO对象-检索服务前后端逻辑实现-构建SearchRequest、SearchResponse对象】

二.线程的实现方式

2.1 继承Thread

class ThreadDemo01 extends Thread{@Overridepublic void run() {System.out.println("当前线程:" + Thread.currentThread().getName());}
}

2.2 实现Runnable接口

class ThreadDemo02 implements Runnable{@Overridepublic void run() {System.out.println("当前线程:" + Thread.currentThread().getName());}
}

2.3 Callable接口

class MyCallable implements Callable<Integer>{@Overridepublic Integer call() throws Exception {System.out.println("当前线程:" + Thread.currentThread().getName());return 123;}
}

public static void main(String[] args) throws ExecutionException, InterruptedException {System.out.println("main开始执行...");ThreadDemo01 t1 = new ThreadDemo01();t1.start();ThreadDemo02 t2 = new ThreadDemo02();new Thread(t2).start();new Thread(()->{System.out.println("当前线程:" + Thread.currentThread().getName());}).start();// 通过Callable接口来实现  FutureTask 本质上是一个Runnable接口FutureTask futureTask = new FutureTask(new MyCallable());Thread t3 = new Thread(futureTask);t3.start();// 阻塞等待子线程的执行完成，然后获取线程的返回结果Object o = futureTask.get();System.out.println("o = " + o);System.out.println("main方法结束了...");}

2.4 线程池创建线程的实现方式

上面的三种获取线程的方法是直接获取，没有对线程做相关的管理，这时可以通过线程池来更加高效的管理线程对象。

// 定义一个线程池对象private static ExecutorService service = Executors.newFixedThreadPool(5);

然后我们就可以通过这个线程池对象来获取对应的线程

        service.execute(new Runnable() {@Overridepublic void run() {System.out.println("线程池--》当前线程:" + Thread.currentThread().getName());}});

2.5 获取线程的区别

通过上面的介绍我们发现获取线程的方式

继承Thread对象
实现Runnable接口
实现Callable接口
线程池

继承Thread对象和实现Runnable接口没有办法获取返回结果的，实现Callable接口可以获取线程的返回结果。当然这三种方式都不能控制我们的资源，线程池可以控制资源。

三.线程池的详解

3.1 线程池的创建方式

通过Executors的静态方法
通过 new ThreadPoolExecutor方式创建

七大参数的作用

参数	作用
corePoolSize	核心线程数，线程池创建好后就准备就绪的线程数量，一直存在
maximumPoolSize	最大线程数量，控制资源
keepAliveTime	存活时间，如果当前线程数量大于核心线程数量，释放空闲的线程，最大线程-核心数量
unit	时间单位
BlockingQueue	阻塞队列，如果任务很多，就会把多的任务放在队列中
threadFactory	线程的工厂
handler	如果队列满了，按照指定的拒绝策略执行任务

    /*** 线程池详解* @param args*/public static void main(String[] args) {// 第一种获取的方式ExecutorService service = Executors.newFixedThreadPool(10);// 第二种方式： 直接new ThreadPoolExecutor()对象，并且手动的指定对应的参数// corePoolSize:线程池的核心线程数量 线程池创建出来后就会 new Thread() 5个// maximumPoolSize:最大的线程数量，线程池支持的最大的线程数// keepAliveTime:存活时间，当线程数大于核心线程，空闲的线程的存活时间 8-5=3// unit:存活时间的单位// BlockingQueue<Runnable> workQueue:阻塞队列 当线程数超过了核心线程数据，那么新的请求到来的时候会加入到阻塞的队列中// new LinkedBlockingQueue<>() 默认队列的长度是 Integer.MAX 那这个就太大了，所以我们需要指定队列的长度// threadFactory:创建线程的工厂对象// RejectedExecutionHandler handler:当线程数大于最大线程数的时候会执行的淘汰策略ThreadPoolExecutor poolExecutor = new ThreadPoolExecutor(5, 100, 10, TimeUnit.SECONDS, new LinkedBlockingQueue<>(10000), Executors.defaultThreadFactory(), new ThreadPoolExecutor.AbortPolicy());poolExecutor.execute(()->{System.out.println("----->" + Thread.currentThread().getName());});}

3.2 线程池的执行顺序

线程池创建，准备好core数量的核心线程，准备接收任务

1.先判断核心线程是否已满，未满分配线程
2.任务队列是否已满，未满放入队列
3.是否达到最大的线程数量，未达到创建新的线程
4.通过对应的reject指定的拒绝策略进行处理

线程池的面试题：

有一个线程池，core：5，max：50，queue：100，如果并发是200，那么线程池是怎么处理的？
首先 200个中的前面5个会直接被核心线程处理，然后6个到105个会加入到阻塞队列中，然后106到155的请求在最大线程数中，那么会创建对应的线程来处理这些请求，之后剩下的45个请求会被直接放弃

3.3 线程池的好处

降低资源消耗
提高响应速度
提高线程的管理

四.CompletableFutrue异步处理

一个商品详情页

展示SKU的基本信息 0.5s
展示SKU的图片信息 0.6s
展示SKU的销售信息 1s
spu的销售属性 1s
展示规格参数 1.5s
spu详情信息 1s

4.1 ComplatableFuture介绍

Future是Java 5添加的类，用来描述一个异步计算的结果。你可以使用 isDone方法检查计算是否完成，或者使用 get阻塞住调用线程，直到计算完成返回结果，你也可以使用 cancel方法停止任务的执行。

虽然 Future以及相关使用方法提供了异步执行任务的能力，但是对于结果的获取却是很不方便，只能通过阻塞或者轮询的方式得到任务的结果。阻塞的方式显然和我们的异步编程的初衷相违背，轮询的方式又会耗费无谓的CPU资源，而且也不能及时地得到计算结果，为什么不能用观察者设计模式当计算结果完成及时通知监听者呢？

很多语言，比如Node.js，采用回调的方式实现异步编程。Java的一些框架，比如Netty，自己扩展了Java的 Future接口，提供了 addListener等多个扩展方法；Google guava也提供了通用的扩展Future；Scala也提供了简单易用且功能强大的Future/Promise异步编程模式。

作为正统的Java类库，是不是应该做点什么，加强一下自身库的功能呢？

在Java 8中, 新增加了一个包含50个方法左右的类: CompletableFuture，提供了非常强大的Future的扩展功能，可以帮助我们简化异步编程的复杂性，提供了函数式编程的能力，可以通过回调的方式处理计算结果，并且提供了转换和组合CompletableFuture的方法。

CompletableFuture类实现了Future接口，所以你还是可以像以前一样通过 get方法阻塞或者轮询的方式获得结果，但是这种方式不推荐使用。

CompletableFuture和FutureTask同属于Future接口的实现类，都可以获取线程的执行结果。

4.2 创建异步对象

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

static CompletableFuture<Void> runAsync(Runnable runnable)
public static CompletableFuture<Void> runAsync(Runnable runnable, Executor executor)
public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier)
public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier, Executor executor)

方法分为两类：

runAsync 没有返回结果
supplyAsync 有返回结果

    private static ThreadPoolExecutor executor = new ThreadPoolExecutor(5,50,10, TimeUnit.SECONDS,new LinkedBlockingQueue<>(100), Executors.defaultThreadFactory(),new ThreadPoolExecutor.AbortPolicy());public static void main(String[] args) throws ExecutionException, InterruptedException {System.out.println("main -- 线程开始了...");// 获取CompletableFuture对象CompletableFuture<Void> voidCompletableFuture = CompletableFuture.runAsync(() -> {System.out.println("线程开始了...");int i = 100/50;System.out.println("线程结束了...");},executor);System.out.println("main -- 线程结束了...");System.out.println("------------");CompletableFuture<Integer> future = CompletableFuture.supplyAsync(() -> {System.out.println("线程开始了...");int i = 100 / 50;System.out.println("线程结束了...");return i;}, executor);System.out.println("获取的线程的返回结果是：" + future.get() );}

4.3 whenXXX和handle方法

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

public CompletableFuture<T> whenComplete(BiConsumer<? super T,? super Throwable> action);
public CompletableFuture<T> whenCompleteAsync(BiConsumer<? super T,? super Throwable> action);
public CompletableFuture<T> whenCompleteAsync(BiConsumer<? super T,? super Throwable> action, Executor executor);public CompletableFuture<T> exceptionally(Function<Throwable,? extends T> fn);public <U> CompletableFuture<U> handle(BiFunction<? super T, Throwable, ? extends U> fn) ;
public <U> CompletableFuture<U> handleAsync(BiFunction<? super T, Throwable, ? extends U> fn) ;
public <U> CompletableFuture<U> handleAsync(BiFunction<? super T, Throwable, ? extends U> fn, Executor executor) ;

4.4 线程串行方法

thenApply 方法：当一个线程依赖另一个线程时，获取上一个任务返回的结果，并返回当前任务的返回值。
thenAccept方法：消费处理结果。接收任务的处理结果，并消费处理，无返回结果。
thenRun方法：只要上面的任务执行完成，就开始执行thenRun，只是处理完任务后，执行 thenRun的后续操作

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

public <U> CompletableFuture<U> thenApply(Function<? super T,? extends U> fn)
public <U> CompletableFuture<U> thenApplyAsync(Function<? super T,? extends U> fn)
public <U> CompletableFuture<U> thenApplyAsync(Function<? super T,? extends U> fn, Executor executor)public CompletionStage<Void> thenAccept(Consumer<? super T> action);
public CompletionStage<Void> thenAcceptAsync(Consumer<? super T> action);
public CompletionStage<Void> thenAcceptAsync(Consumer<? super T> action,Executor executor);public CompletionStage<Void> thenRun(Runnable action);
public CompletionStage<Void> thenRunAsync(Runnable action);
public CompletionStage<Void> thenRunAsync(Runnable action,Executor executor);

4.5 两个都完成

上面介绍的相关方法都是串行的执行，接下来看看需要等待两个任务执行完成后才会触发的几个方法

thenCombine ：可以获取前面两线程的返回结果，本身也有返回结果
thenAcceptBoth:可以获取前面两线程的返回结果，本身没有返回结果
runAfterBoth：不可以获取前面两线程的返回结果，本身也没有返回结果

/*** @param args* @throws ExecutionException* @throws InterruptedException*/public static void main(String[] args) throws ExecutionException, InterruptedException {CompletableFuture<Integer> future1 = CompletableFuture.supplyAsync(() -> {System.out.println("任务1 线程开始了..." + Thread.currentThread().getName());int i = 100 / 5;System.out.println("任务1 线程结束了..." + Thread.currentThread().getName());return i;}, executor);CompletableFuture<Integer> future2 = CompletableFuture.supplyAsync(() -> {System.out.println("任务2 线程开始了..." + Thread.currentThread().getName());int i = 100 /10;System.out.println("任务2 线程结束了..." + Thread.currentThread().getName());return i;}, executor);// runAfterBothAsync 不能获取前面两个线程的返回结果，本身也没有返回结果CompletableFuture<Void> voidCompletableFuture = future1.runAfterBothAsync(future2, () -> {System.out.println("任务3执行了");},executor);// thenAcceptBothAsync 可以获取前面两个线程的返回结果，本身没有返回结果CompletableFuture<Void> voidCompletableFuture1 = future1.thenAcceptBothAsync(future2, (f1, f2) -> {System.out.println("f1 = " + f1);System.out.println("f2 = " + f2);}, executor);// thenCombineAsync: 既可以获取前面两个线程的返回结果，同时也会返回结果给阻塞的线程CompletableFuture<String> stringCompletableFuture = future1.thenCombineAsync(future2, (f1, f2) -> {return f1 + ":" + f2;}, executor);// 可以处理异步任务之后的操作System.out.println("获取的线程的返回结果是：" + stringCompletableFuture.get() );}

4.6 两个任务完成一个

在上面5个基础上我们来看看两个任务只要有一个完成就会触发任务3的情况

runAfterEither:不能获取完成的线程的返回结果，自身也没有返回结果
acceptEither:可以获取线程的返回结果，自身没有返回结果
applyToEither:既可以获取线程的返回结果，自身也有返回结果

/*** @param args* @throws ExecutionException* @throws InterruptedException*/public static void main(String[] args) throws ExecutionException, InterruptedException {CompletableFuture<Object> future1 = CompletableFuture.supplyAsync(() -> {System.out.println("任务1 线程开始了..." + Thread.currentThread().getName());int i = 100 / 5;System.out.println("任务1 线程结束了..." + Thread.currentThread().getName());return i;}, executor);CompletableFuture<Object> future2 = CompletableFuture.supplyAsync(() -> {System.out.println("任务2 线程开始了..." + Thread.currentThread().getName());int i = 100 /10;try {Thread.sleep(5000);} catch (InterruptedException e) {e.printStackTrace();}System.out.println("任务2 线程结束了..." + Thread.currentThread().getName());return i+"";}, executor);// runAfterEitherAsync 不能获取前面完成的线程的返回结果，自身也没有返回结果future1.runAfterEitherAsync(future2,()->{System.out.println("任务3执行了....");},executor);// acceptEitherAsync 可以获取前面完成的线程的返回结果  自身没有返回结果future1.acceptEitherAsync(future2,(res)->{System.out.println("res = " + res);},executor);// applyToEitherAsync 既可以获取完成任务的线程的返回结果  自身也有返回结果CompletableFuture<String> stringCompletableFuture = future1.applyToEitherAsync(future2, (res) -> {System.out.println("res = " + res);return res + "-->OK";}, executor);// 可以处理异步任务之后的操作System.out.println("获取的线程的返回结果是：" + stringCompletableFuture.get() );}

4.7 多任务组合

allOf：等待所有任务完成
anyOf：只要有一个任务完成

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

/*** @param args* @throws ExecutionException* @throws InterruptedException*/public static void main(String[] args) throws ExecutionException, InterruptedException {CompletableFuture<Object> future1 = CompletableFuture.supplyAsync(() -> {System.out.println("任务1 线程开始了..." + Thread.currentThread().getName());int i = 100 / 5;System.out.println("任务1 线程结束了..." + Thread.currentThread().getName());return i;}, executor);CompletableFuture<Object> future2 = CompletableFuture.supplyAsync(() -> {System.out.println("任务2 线程开始了..." + Thread.currentThread().getName());int i = 100 /10;try {Thread.sleep(5000);} catch (InterruptedException e) {e.printStackTrace();}System.out.println("任务2 线程结束了..." + Thread.currentThread().getName());return i+"";}, executor);CompletableFuture<Object> future3 = CompletableFuture.supplyAsync(() -> {System.out.println("任务3 线程开始了..." + Thread.currentThread().getName());int i = 100 /10;System.out.println("任务3 线程结束了..." + Thread.currentThread().getName());return i+"";}, executor);CompletableFuture<Object> anyOf = CompletableFuture.anyOf(future1, future2, future3);anyOf.get();System.out.println("主任务执行完成..." + anyOf.get());CompletableFuture<Void> allOf = CompletableFuture.allOf(future1, future2, future3);allOf.get();// 阻塞在这个位置，等待所有的任务执行完成System.out.println("主任务执行完成..." + future1.get() + " :" + future2.get() + " :" + future3.get());}

好了，关于【45-线程的实现方式-线程池的创建方式-线程池的执行顺序-CompletableFutrue异步处理】就先学习到这里，后续的内容不断学习中。