什么事网站建设,上海公共招聘网新版,北京网络推广公司,公司企业文化墙设计方案近期热推文章#xff1a; 1、springBoot对接kafka,批量、并发、异步获取消息,并动态、批量插入库表; 2、SpringBoot用线程池ThreadPoolTaskExecutor异步处理百万级数据; 3、基于Redis的Geo实现附近商铺搜索(含源码) 4、基于Redis实现关注、取关、共同关注及消息推送(含源码) 5…近期热推文章 1、springBoot对接kafka,批量、并发、异步获取消息,并动态、批量插入库表; 2、SpringBoot用线程池ThreadPoolTaskExecutor异步处理百万级数据; 3、基于Redis的Geo实现附近商铺搜索(含源码) 4、基于Redis实现关注、取关、共同关注及消息推送(含源码) 5、SpringBoot整合多数据源并支持动态新增与切换详细教程 6、基于Redis实现点赞及排行榜功能
一、多任务组合回调 备注源码获取方式在文底。
1.1、AND组合关系 thenCombine / thenAcceptBoth / runAfterBoth都表示将两个CompletableFuture组合起来只有这两个都正常执行完了才会执行某个任务。也即当任务一和任务二都完成再执行任务三(异步任务)。
区别在于 1、runAfterBoth不会把执行结果当做方法入参且没有返回值。 2、thenAcceptBoth会将两个任务的执行结果作为方法入参传递到指定方法中且无返回值。 3、thenCombine会将两个任务的执行结果作为方法入参传递到指定方法中且有返回值。
代码案例
/** * 功能描述:多任务组合回调:AND组合关系 * MethodName: testCompleteAnd * MethodParam: [] * Return: void * Author: yyalin * CreateDate: 2023/10/11 17:30 */ public void testCompleteAnd() throws ExecutionException, InterruptedException { //创建线程池 ExecutorService executorService Executors.newFixedThreadPool(10); long startTime System.currentTimeMillis(); //1、使用自定义线程池开启异步任务01 CompletableFutureInteger supplyAsyncRes01CompletableFuture.supplyAsync(()-{ int res1; try { //执行任务1 开始执行任务01,当前线程为12 log.info(开始执行任务01,当前线程为Thread.currentThread().getId()); //执行具体的事务 Thread.sleep(600); res1; //模拟加1 } catch (InterruptedException e) { e.printStackTrace(); } //返回结果 return res; },executorService); //2、使用自定义线程池开启异步任务02 CompletableFutureInteger supplyAsyncRes02CompletableFuture.supplyAsync(()-{ int res1; try { //执行任务02 开始执行任务02,当前线程为13 log.info(开始执行任务02,当前线程为Thread.currentThread().getId()); //执行具体的事务 Thread.sleep(600); res2; //模拟加2 } catch (InterruptedException e) { e.printStackTrace(); } //返回结果 return res; }); //3、任务02将任务1与任务2开始任务组合 CompletableFutureInteger thenCombineAsyncRessupplyAsyncRes01.thenCombineAsync(supplyAsyncRes02,(res01, res02)-{ //始执行任务03,当前线程为14 log.info(开始执行任务03,当前线程为Thread.currentThread().getId()); log.info(任务01返回值res01); log.info(任务02返回值res02); //任务组合返回值 可以拿到任务01和任务02的返回结果进行相关操作然后统一返回结果 return res01res02; },executorService); //4、最终返回结果 log.info(最终返回结果为thenCombineAsyncRes.get()); log.info(总共用时 (System.currentTimeMillis() - startTime) ms); }
运行结果 1.2、OR组合关系 将两个CompletableFuture组合起来只要其中一个执行完了就会执行某个任务。两个任务只要有一个任务完成就执行任务三
区别在于 1、runAfterEither不会把执行结果当做方法入参且没有返回值。 2、acceptEither: 会将已经执行完成的任务作为方法入参传递到指定方法中且无返回值。 3、applyToEither会将已经执行完成的任务作为方法入参传递到指定方法中且有返回值。(个人推荐)
参考代码 /** * 功能描述:OR组合关系 * MethodName: testCompleteOr * MethodParam: [] * Return: void * Author: yyalin * CreateDate: 2023/10/11 18:14 */ public void testCompleteOr(){ //创建线程池 ExecutorService executorService Executors.newFixedThreadPool(10); long startTime System.currentTimeMillis(); //1、使用自定义线程池开启异步任务01 CompletableFutureInteger supplyAsyncRes01CompletableFuture.supplyAsync(()-{ int res1; try { //执行任务1 开始执行任务01,当前线程为12 log.info(开始执行任务01,当前线程为Thread.currentThread().getId()); //执行具体的事务 Thread.sleep(600); res2; //模拟加1 } catch (InterruptedException e) { e.printStackTrace(); } //返回结果 return res; },executorService); //2、使用自定义线程池开启异步任务02 CompletableFutureInteger supplyAsyncRes02CompletableFuture.supplyAsync(()-{ int res1; try { //执行任务02 开始执行任务02,当前线程为13 log.info(开始执行任务02,当前线程为Thread.currentThread().getId()); //执行具体的事务 Thread.sleep(600); res3; //模拟加2 } catch (InterruptedException e) { e.printStackTrace(); } //返回结果 return res; },executorService); //3、任务组合or supplyAsyncRes01.acceptEitherAsync(supplyAsyncRes02,(res)-{ try { log.info(开始执行任务03,当前线程为Thread.currentThread().getId()); //执行具体的事务 Thread.sleep(600); log.info(上一个任务返回值res); log.info(总共用时 (System.currentTimeMillis() - startTime) ms); } catch (InterruptedException e) { e.printStackTrace(); } },executorService); }
返回结果 若将异步任务02中的Thread.sleep(600)改为300将输出的结果为 从结果中不难对比发现任务03的参数是任务01和任务02中执行最快的返回结果。
注意若把核心线程数量改为1会是什么样的呢
ExecutorService executorService Executors.newFixedThreadPool(1);
运行结果 从上面看出改为1就变成单线程执行了。
1.3、多任务组合(allOf\anyOf) 1.allOf等待所有任务都执行完成后才会执行 allOf 返回的CompletableFuture。如果任意一个任务异常allOf的CompletableFuture执行get方法会抛出异常。(等待所有任务完成才会执行) 2.anyOf:任意一个任务执行完就执行anyOf返回的CompletableFuture。如果执行的任务异常anyOf的CompletableFuture执行get方法会抛出异常。(只要有一个任务完成)
参考案例
public void testAllOfOrAnyOf() throws ExecutionException, InterruptedException { //创建线程池 ExecutorService executorService Executors.newFixedThreadPool(10); long startTime System.currentTimeMillis(); //1、使用自定义线程池开启异步任务01 CompletableFutureInteger supplyAsyncRes01CompletableFuture.supplyAsync(()-{ int res1; try { //执行任务1 开始执行任务01,当前线程为12 log.info(开始执行任务01,当前线程为Thread.currentThread().getId()); //执行具体的事务 Thread.sleep(600); res3; //模拟加1 } catch (InterruptedException e) { e.printStackTrace(); } //返回结果 return res; },executorService); //2、使用自定义线程池开启异步任务02 CompletableFutureInteger supplyAsyncRes02CompletableFuture.supplyAsync(()-{ int res1; try { //执行任务02 开始执行任务02,当前线程为13 log.info(开始执行任务02,当前线程为Thread.currentThread().getId()); //执行具体的事务 Thread.sleep(600); res4; //模拟加2 } catch (InterruptedException e) { e.printStackTrace(); } //返回结果 return res; },executorService); //3、使用自定义线程池开启异步任务03 CompletableFutureInteger supplyAsyncRes03CompletableFuture.supplyAsync(()-{ int res1; try { //执行任务02 开始执行任务02,当前线程为13 log.info(开始执行任务03,当前线程为Thread.currentThread().getId()); //执行具体的事务 Thread.sleep(600); res5; //模拟加2 } catch (InterruptedException e) { e.printStackTrace(); } //返回结果 return res; },executorService); //4、开始任务组合 CompletableFutureVoid allOfResCompletableFuture.allOf(supplyAsyncRes01,supplyAsyncRes02,supplyAsyncRes03); //等待所有任务完成 log.info(所有任务执行完成,组合后返回结果为allOfRes.get()); //获取所有任务的返回结果 log.info(任务01返回值supplyAsyncRes01.get()); log.info(任务02返回值supplyAsyncRes02.get()); log.info(任务03返回值supplyAsyncRes03.get()); log.info(总共用时 (System.currentTimeMillis() - startTime) ms); } 结果返回 从结果中看出等待所有任务都执行完成后才会执行 allOf 返回的CompletableFuture。
同理anyOf只需要调整代码 CompletableFutureObject allOfResCompletableFuture.anyOf(supplyAsyncRes01,supplyAsyncRes02,supplyAsyncRes03);
运行结果 1.4、thenCompose
thenCompose方法会在某个任务执行完成后将该任务的执行结果作为方法入参去执行指定的方法。该方法会返回一个新的CompletableFuture实例。
1、如果该CompletableFuture实例的result不为null则返回一个基于该result新的CompletableFuture实例
2、如果该CompletableFuture实例为null然后就执行这个新任务。
代码案例 /** * 功能描述:thenCompose * MethodName: testThenCompose * MethodParam: [] * Return: void * Author: yyalin * CreateDate: 2023/10/12 9:38 */ public void testThenCompose() throws ExecutionException, InterruptedException { CompletableFutureString res01CompletableFuture.completedFuture(任务01); ExecutorService executor Executors.newSingleThreadExecutor(); //第二个任务 在某个任务执行完成后将该任务的执行结果作为方法入参去执行指定的方法, // 该方法会返回一个新的CompletableFuture实例。 CompletableFutureString futureRes CompletableFuture.supplyAsync(()- 第二个任务02 ,executor).thenComposeAsync(data-{ log.info(data数据为data); return res01; },executor); log.info(最终返回futureRes.get()); executor.shutdown(); }
结果 二、使用注意点
CompletableFuture 使异步编程更加便利的、代码更加优雅的同时也要关注使用的一些注意点。 2.1、Future需要获取返回值才能获取异常信息
代码案例
/** * 功能描述:使用注意点 * MethodName: testFuture * MethodParam: [] * Return: void * Author: yyalin * CreateDate: 2023/10/12 9:54 */ public void testFuture() throws ExecutionException, InterruptedException { //自定义线程池 ExecutorService executorService new ThreadPoolExecutor( 5, 10, 5L, TimeUnit.SECONDS, new ArrayBlockingQueue(10)); //创建任务 CompletableFutureVoid res01CompletableFuture.supplyAsync(()-{ int sum1/0; return 分母不能为0; },executorService).thenAccept((res)-{ //3、异常捕获 log.info(系统出现异常需要处理res); }); log.info(返回结果res01.get()); }
输出结果 Future需要获取返回值(res01.get())才能获取到异常信息。如果不加 get()/join()方法看不到异常信息。使用的时候注意一下考虑是否加try…catch…或者使用exceptionally方法。
若改成exceptionally方法无需get或join也可以捕获异常信息 CompletableFutureString res01CompletableFuture.supplyAsync(()-{ int sum1/0; return 分母不能为0; },executorService).exceptionally((throwable)-{ //3、异常捕获 log.info(系统出现异常需要处理throwable.getMessage()); return 00; });// log.info(返回结果res01.get());
结果 2.2、CompletableFuture的get()方法是阻塞的
CompletableFuture的get()方法是阻塞的如果使用它来获取异步调用的返回值需要添加超时时间。 推荐使用
log.info(返回结果res01.get(5,TimeUnit.SECONDS));
2.3、建议使用自定义线程池不要使用默认的
CompletableFuture代码中使用了默认的线程池处理的线程个数是电脑CPU核数-1。在大量请求过来的时候处理逻辑复杂的话响应会很慢。一般建议使用自定义线程池优化线程池配置参数。
参考案例 //自定义线程池ExecutorService executorService new ThreadPoolExecutor( 5, 10, 5L, TimeUnit.SECONDS, new ArrayBlockingQueue(10));
但是如果线程池拒绝策略是DiscardPolicy或者DiscardOldestPolicy当线程池饱和时会直接丢弃任务不会抛弃异常。因此建议CompletableFuture线程池策略最好使用AbortPolicy然后耗时的异步线程做好线程池隔离。
/*** 参数信息* int corePoolSize 核心线程大小* int maximumPoolSize 线程池最大容量大小* long keepAliveTime 线程空闲时线程存活的时间* TimeUnit unit 时间单位* BlockingQueueRunnable workQueue 任务队列。一个阻塞队列* AbortPolicy默认直接抛弃*/ThreadPoolExecutor pool new ThreadPoolExecutor(4, 4, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingDeque(10), new ThreadPoolExecutor.AbortPolicy());
说明
AbortPolicy默认直接抛弃
CallerRunsPolicy用调用者的线程执行任务
DiscardOldestPolicy抛弃队列中最久的任务
DiscardPolicy抛弃当前任务。
三、源码获取方式 更多优秀文章请关注个人微信公众号或搜索“程序猿小杨”查阅。然后回复源码可以获取对应的源码开箱即可使用。 如果大家对相关文章感兴趣可以关注微信公众号程序猿小杨会持续更新优秀文章!欢迎大家 分享、收藏、点赞、在看您的支持就是我坚持下去的最大动力谢谢! 参考网站
https://blog.csdn.net/ThinkWon/article/details/123390393
https://mp.weixin.qq.com/s/shjANruBk6VL492JaWLTEg
