门户网站建设需求,湛江网站建设团队,短视频拍摄价目表,珠海网站运营文章目录一、处理异步任务并获取返回值——CompletionService二、线程池三、Callable 与 Future四、通过回调方式处理可组合编排任务——CompletableFuture一、处理异步任务并获取返回值——CompletionService
特点描述#xff1a; 对于比较复杂的计算#xff0c;把…
文章目录一、处理异步任务并获取返回值——CompletionService二、线程池三、Callable 与 Future四、通过回调方式处理可组合编排任务——CompletableFuture一、处理异步任务并获取返回值——CompletionService
特点描述 对于比较复杂的计算把任务进行提交并发执行哪个任务先执行完get()方法就会获取到相应的任务结果。
范式 1、 假设有一组针对某个问题的任务solvers需要实现Callable接口任务的具体逻辑就在其call方法里每个任务都返回一个类型为Result的值并且想要并发地运行它们处理每个返回一个非空值的结果在某些方法使用
void solve(Executor e,CollectionCallableResult solvers)throws InterruptedException, ExecutionException {CompletionServiceResult ecs new ExecutorCompletionServiceResult(e);for (CallableResult s : solvers)ecs.submit(s);int n solvers.size();for (int i 0; i n; i) {Result r ecs.take().get();if (r ! null)use(r);}}2、 假设想使用任务集的第一个非空结果忽略任何遇到异常的任务并在第一个任务准备好时取消所有其他任务比如多仓库文件/镜像下载从最近的服务中心下载后终止其他下载过程
void solve(Executor e,CollectionCallableResult solvers)throws InterruptedException {CompletionServiceResult ecs new ExecutorCompletionServiceResult(e);int n solvers.size();ListFutureResult futures new ArrayListFutureResult(n);Result result null;try {for (CallableResult s : solvers)futures.add(ecs.submit(s));for (int i 0; i n; i) {try {Result r ecs.take().get();if (r ! null) {result r;break;}} catch (ExecutionException ignore) {}}}finally {for (FutureResult f : futures)// 注意这里的参数给的是 true详解同样在前序 Future 源码分析文章中f.cancel(true);}if (result ! null)use(result);}
总得来说分两步 1、提交异步任务 submit方法submit最终会委托给内部的 executor 去执行任务 2、从队列中拿取并移除元素 take如果队列为空那么调用 take() 方法的线程会被阻塞/poll…不会被阻塞返回null/poll带超时参数获取并移除阻塞队列中的第一个元素如果超时时间到而队列还是空该方法返回null 方法
实现原理 将异步任务的生产、任务完成结果的消费进行解耦类似mq哪个任务先执行完就把结果放到队列中。
唯一实现类 ExecutorCompletionService阻塞队列默认是 LinkedBlockingQueue
二、线程池
为什么要用线程池 ∵ 手动创建线程的缺点 1、不受控系统资源有限每个人如果都创建的话标准不一样线程疯狂抢占资源.混乱… 2、开销大创建一个线程需要调用操作系统内核API然后操作系统要为线程分配一系列资源创建个线程啥也不干大概需要1M左右大小。 线程池可以统一管理、控制最大并发数并实现拒绝策略、隔离线程环境当执行大量异步任务时线程池里的线程能复用不用频繁创建和销毁能够提供好的性能。 Java并发包里的线程池——ThreadPoolExecutor 接口是ExecutorService Spring对线程池的封装——ThreadPoolTaskExecutor 关于线程池核心线程数的设置 CPU是时间片轮转机制来让线程占用的也就是说程序表面上是同时进行的实际上是切换执行的CPU每个时刻只能由一个线程占用比如 4核CPU只能同时跑4个线程。 对于CPU密集型程序如运算、逻辑判断等I/O操作可以在短时间完成但CPU运算比较多 ——最佳线程数量CPU核数1这个1可以理解为替补如果某个线程因为发生错误或其他原因暂停了这个线程可以继续工作。 对于I/O密集型如涉及网络、磁盘、内存等 ——最佳线程数CPU核心数 * (1/CPU利用率)CPU核心数 * (1 (I/O耗时/CPU耗时))如果几乎都是I/O耗时可取2N11为替补 p.s.线程数不是越多越好线程上下文切换开销不小
三、Callable 与 Future Runnable接口的方法没有返回值Callable 是泛型接口可以返回指定类型的结果。 当提交一个Callable 任务后会同时获得一个Future对象然后在主线程某个时刻调用Future对象的get() 方法就可以获得异步执行的结果。在调用get()时如果异步任务已经完成就直接获得结果。如果异步任务还没有完成那么get()会阻塞直到任务完成后才返回结果。
四、通过回调方式处理可组合编排任务——CompletableFuture
特点描述 CompletableFuture是由 Java 8 引入的在 Java 8之 前一般通过 Future 实现异步CompletableFuture 对 Future 进行了扩展可以通过设置回调的方式处理计算结果同时也支持组合操作比如步骤1、2、3存在依赖关系支持对步骤进一步的编排降低依赖之间的阻塞。
使用 如上图所示这里描绘的是一个业务接口的流程其中包括 CF1\CF2\CF3\CF4\CF5 共5个步骤并描绘了这些步骤之间的依赖关系每个步骤可以是一次 RPC 调用、一次数据库操作或者是一次本地方法调用等在使用 CompletableFuture 进行异步化编程时图中的每个步骤都会产生一个 CompletableFuture 对象最终结果也会用一个 CompletableFuture 来进行表示。只看第一层的话 好像跟 CompletionService 效果差不多… 都可以异步执行批量任务并拿到结果…
1、零依赖CompletableFuture 的创建 比如图中所示的 CF1、CF2可以有以下方式 // 1、使用 runAsync 或 supplyAsync 发起异步调用// 线程池ExecutorService executorService Executors.newFixedThreadPool(5);CompletableFutureString CF1 CompletableFuture.supplyAsync(() - {return CF1 result;}, executorService);// 2、CompletableFuture.completedFuture() 直接创建一个已完成状态的 CompletableFutureCompletableFutureString CF2 CompletableFuture.completedFuture(CF2 result);// 3、先初始化一个未完成的 CompletableFutureCompletableFutureString CF3 new CompletableFuture();// 然后通过complete()、completeExceptionally()完成该CompletableFutureCF3.complete(CF3result);2、一元依赖依赖一个 CompletableFuture 比如图中所示的 CF3、CF5可以用 thenApply、thenAccept、thenCompose 等方法来实现
// result为CF1的结果CompletableFutureString CF3CF1.thenApply(result-{return CF3result;});3、二元依赖依赖两个 CompletableFuture 比如图中所示的 CF4这种二元依赖可以通过 thenCombine 等回调来实现 // result1、result2分别为CF1、CF2的结果CompletableFutureString CF4 CF1.thenCombine(CF2, (result1, result2) - {return CF4result;});4、多元依赖依赖多个 CompletableFuture 比如图中所示的 CF6依赖于三个步骤CF3、CF4、CF5这种多元依赖可以通过 allOf 或 anyOf 方法来实现区别是当需要多个依赖全部完成时使用allOf当多个依赖中的任意一个完成即可时使用anyOf CompletableFutureVoid CF6 CompletableFuture.allOf(CF3, CF4, CF5);CompletableFutureString result CF6.thenApply(v -{// 这里的 join是完成任务后用来获取结果的并不会阻塞// 因为传给 thenApply 的函数都是在 CF3、CF4、CF5 全都完成时才会执行String result3 CF3.join();String result4 CF4.join();String result5 CF5.join();// 根据 result3、result4、result5组装最终 resultreturn result3 result4 result5;});如果只用一层的话异步执行批量任务并拿到总的结果参考api里 allOf
代码示例 // 任务入参集合ArrayListString paramList new ArrayList();// 用于汇总所有结果ArrayListString resultList new ArrayList();CompletableFuture.allOf(paramList.stream().map(string -CompletableFuture.supplyAsync(() -// 这里返回了本身实际上也可以是具体的方法string,asyncServiceExecutor)// thenApply是对结果做简单映射类似于Stream.maplist-list就是原样往下传递这里不使用thenApply也行.thenApply(list - list).whenComplete((result, e) - {// 对异常结果的处理if (e ! null) System.out.println(exception);// 汇总结果resultList.add(result);})).toArray(CompletableFuture[]::new)// 完成后返回结果值如果异常完成则抛出未经检查异常相当于一个等待任务完成的动作).join(); 参考文档 https://dayarch.top/p/how-many-threads-should-be-created.html https://segmentfault.com/a/1190000023129592?utm_sourcesf-similar-article https://segmentfault.com/a/1190000023587881 https://tech.meituan.com/2022/05/12/principles-and-practices-of-completablefuture.html
