线程池提交任务方法
时间:2021-09-13
本文章向大家介绍线程池提交任务方法,主要包括线程池提交任务方法使用实例、应用技巧、基本知识点总结和需要注意事项,具有一定的参考价值,需要的朋友可以参考一下。
excute方法: 源码
submit方法通过提交参数构造FutrueTask,然后执行excute(FutrueTask)方法,返回一个future对象
public Future<?> submit(Runnable task) { if (task == null) throw new NullPointerException(); RunnableFuture<Void> ftask = newTaskFor(task, null);//new FutureTask<T>(runnable, null) execute(ftask); return ftask; }
public <T> Future<T> submit(Runnable task, T result) { if (task == null) throw new NullPointerException(); RunnableFuture<T> ftask = newTaskFor(task, result);//new FutureTask<T>(runnable, value) execute(ftask); return ftask; }
public <T> Future<T> submit(Callable<T> task) { if (task == null) throw new NullPointerException(); RunnableFuture<T> ftask = newTaskFor(task); execute(ftask); return ftask; }
任务批量提交
超时时间:执行invokeAll或者invokeAny的时间
会等待所有任务完成,如果异常终止(包括超时停止),取消所有任务
public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks) throws InterruptedException { if (tasks == null) throw new NullPointerException(); ArrayList<Future<T>> futures = new ArrayList<Future<T>>(tasks.size()); boolean done = false; try { for (Callable<T> t : tasks) { //构造FutureTask,添加进futures集合里 RunnableFuture<T> f = newTaskFor(t); futures.add(f); execute(f);//执行任务 } for (int i = 0, size = futures.size(); i < size; i++) { Future<T> f = futures.get(i); if (!f.isDone()) {//任务已经完成 try { f.get();//这里只是起到阻塞的作用,如果线程中断,则跳到finally } catch (CancellationException ignore) { } catch (ExecutionException ignore) {//不处理任务执行抛出 } } } //如果没走到这里,就说明抛出了异常,可能是线程中断异常或其他异常。 done = true; return futures; } finally { //未正常结束,取消所有任务,通过future get时抛出CancellationException( if (!done) for (int i = 0, size = futures.size(); i < size; i++) futures.get(i).cancel(true); } }
public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks, long timeout, TimeUnit unit) throws InterruptedException { if (tasks == null) throw new NullPointerException(); long nanos = unit.toNanos(timeout); ArrayList<Future<T>> futures = new ArrayList<Future<T>>(tasks.size()); boolean done = false; try { for (Callable<T> t : tasks) futures.add(newTaskFor(t)); final long deadline = System.nanoTime() + nanos; final int size = futures.size(); // Interleave time checks and calls to execute in case // executor doesn't have any/much parallelism. for (int i = 0; i < size; i++) { execute((Runnable)futures.get(i)); nanos = deadline - System.nanoTime(); if (nanos <= 0L)//超时返回 return futures; } for (int i = 0; i < size; i++) { Future<T> f = futures.get(i); //如果任务未完成,则进去阻塞到任务完成 if (!f.isDone()) { //超时返回 if (nanos <= 0L) return futures; try { f.get(nanos, TimeUnit.NANOSECONDS); } catch (CancellationException ignore) { } catch (ExecutionException ignore) { } catch (TimeoutException toe) { return futures;//get超时返回 } nanos = deadline - System.nanoTime(); } } done = true;//未走到这里,标识抛出异常了,在finally里取消所有任务 return futures; } finally { //进而取消所有任务,取消的任务不能再执行了 if (!done) for (int i = 0, size = futures.size(); i < size; i++) futures.get(i).cancel(true); } }
只要有任意一个任务完成,就会返回其结果值,如果超时,则抛出 TimeoutException异常,最终都会取消所有任务
public <T> T invokeAny(Collection<? extends Callable<T>> tasks) throws InterruptedException, ExecutionException { try { return doInvokeAny(tasks, false, 0); } catch (TimeoutException cannotHappen) { assert false; return null; } }
public <T> T invokeAny(Collection<? extends Callable<T>> tasks, long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException { return doInvokeAny(tasks, true, unit.toNanos(timeout)); }
真正执行invokeAny的方法
private <T> T doInvokeAny(Collection<? extends Callable<T>> tasks, boolean timed, long nanos) throws InterruptedException, ExecutionException, TimeoutException { if (tasks == null) throw new NullPointerException(); int ntasks = tasks.size(); if (ntasks == 0) throw new IllegalArgumentException(); ArrayList<Future<T>> futures = new ArrayList<Future<T>>(ntasks); //可以优先获取到已经完成的任务 ExecutorCompletionService<T> ecs = new ExecutorCompletionService<T>(this); // For efficiency, especially in executors with limited // parallelism, check to see if previously submitted tasks are // done before submitting more of them. This interleaving // plus the exception mechanics account for messiness of main // loop. try { // Record exceptions so that if we fail to obtain any // result, we can throw the last exception we got. //任务执行时抛出的异常 ExecutionException ee = null; final long deadline = timed ? System.nanoTime() + nanos : 0L; Iterator<? extends Callable<T>> it = tasks.iterator(); // Start one task for sure; the rest incrementally futures.add(ecs.submit(it.next())); --ntasks; int active = 1;//正在执行的任务数量 for (;;) { //获取已完成的任务, 这里是通过ExecutorCompletionService的方法来获取 Future<T> f = ecs.poll(); //没有已经完成的任务 if (f == null) { if (ntasks > 0) {//当前未提交的任务 --ntasks;//未提交任务-1 futures.add(ecs.submit(it.next()));//提交任务 ++active; } else if (active == 0) //如果没有任务正在执行 break; else if (timed) {//如果设置了超时时间 //阻塞获取 f = ecs.poll(nanos, TimeUnit.NANOSECONDS); //获取不到,就抛出超时异常 if (f == null) throw new TimeoutException(); nanos = deadline - System.nanoTime(); } else f = ecs.take();//阻塞获取 } if (f != null) {//有任务完成了 --active;//任务正在执行数量-1 try { return f.get();//直接返回结果值 } catch (ExecutionException eex) { ee = eex; } catch (RuntimeException rex) { ee = new ExecutionException(rex); } } } if (ee == null) ee = new ExecutionException(); throw ee; } finally { //最终取消所有任务 for (int i = 0, size = futures.size(); i < size; i++) futures.get(i).cancel(true); } }
原文地址:https://www.cnblogs.com/shuiyingyuan/p/15262979.html
- 16.3 安装Tomcat
- 16.2 安装jdk
- UITabBarController实现Tab切换
- React Native库版本升级与降级
- Java并发学习之Volatile及内存模型探究
- Java并发学习之CountDownLatch实现原理及使用姿势
- Linux基础(day58)
- 携程Android App插件化和动态加载实践
- 15.5 使用pure-ftpd搭建ftp服务
- JDK容器学习之Queue: ArrayBlockingQueue
- 手机APP安装包缩减方案
- react-native添加redux支持
- Java并发学习之ReentrantLock的工作原理及使用姿势
- Linux基础(day63)
- JavaScript 教程
- JavaScript 编辑工具
- JavaScript 与HTML
- JavaScript 与Java
- JavaScript 数据结构
- JavaScript 基本数据类型
- JavaScript 特殊数据类型
- JavaScript 运算符
- JavaScript typeof 运算符
- JavaScript 表达式
- JavaScript 类型转换
- JavaScript 基本语法
- JavaScript 注释
- Javascript 基本处理流程
- Javascript 选择结构
- Javascript if 语句
- Javascript if 语句的嵌套
- Javascript switch 语句
- Javascript 循环结构
- Javascript 循环结构实例
- Javascript 跳转语句
- Javascript 控制语句总结
- Javascript 函数介绍
- Javascript 函数的定义
- Javascript 函数调用
- Javascript 几种特殊的函数
- JavaScript 内置函数简介
- Javascript eval() 函数
- Javascript isFinite() 函数
- Javascript isNaN() 函数
- parseInt() 与 parseFloat()
- escape() 与 unescape()
- Javascript 字符串介绍
- Javascript length属性
- javascript 字符串函数
- Javascript 日期对象简介
- Javascript 日期对象用途
- Date 对象属性和方法
- Javascript 数组是什么
- Javascript 创建数组
- Javascript 数组赋值与取值
- Javascript 数组属性和方法
- 搭建简易的物联网服务端和客户端-第四次增补(二十六)
- 实习第十一周,第十二周
- Cordova安装
- django-ckeditor本地图片上传功能
- ESP32 windows下编译环境搭建
- Android hex十六进制转String
- “Ajax请求后台,后台两次session不一致”问题解决
- Angularjs动态加载ECharts(一)
- NodeMCU连接wifi语句报错:“stdin:1: bad argument #1 to 'config' (config table not found!)”
- Angularjs动态加载ECharts(二)
- 【LeetCode】每日一题(8.3)
- LeetCode精选好题(四)
- 【Android 直播软件开发:音视频硬解码篇】
- 知更鸟主题修改CSS:让凌乱的标签变得整洁清爽
- 浅谈 Android Dex 文件