血的教训,如何正确使用线程池 submit 和 execute 方法
血的教训之背景:使用线程池对存量数据进行迁移,但是总有一批数据迁移失败,无异常日志打印
凶案起因
听说 parallelStream
并行流是个好东西,由于日常开发stream
串行流的场景比较多,这次需要写迁移程序刚好可以用得上,那还不赶紧拿来装*一下,此时不装更待何时。机智的我还知道在 JVM 的后台,使用通用的 fork/join 池来完成上述功能,该池是所有并行流共享的,默认情况,fork/join 池会为每个处理器分配一个线程,对应的变通方案就是创建自己的线程池如
ForkJoinPool pool = new ForkJoinPool(Runtime.getRuntime().availableProcessors());
pool.submit(() -> {
list.parallelStream().collect(Collectors.toList());
});
于是地雷是就从这里埋下的。
submit还是execute
public static void main(String[] args) throws InterruptedException, ExecutionException {
final ExecutorService pool = new ForkJoinPool(Runtime.getRuntime().availableProcessors());
List<Integer> list = Lists.newArrayList(1, 2, 3, null);
//1.使用submit
pool.submit(() -> {
list.parallelStream().map(a -> a.toString()).collect(Collectors.toList());
});
TimeUnit.SECONDS.sleep(3);
//2.使用 execute
pool.execute(() -> {
list.parallelStream().map(a -> a.toString()).collect(Collectors.toList());
});
//3.使用submit,调用get()
pool.submit(() -> {
list.parallelStream().map(a -> a.toString()).collect(Collectors.toList());
}).get();
TimeUnit.SECONDS.sleep(3);
}
读者自行跑一下上面的用例,会发现单独使用 submit
方法的并不会打印出错误日志,而使用execute
方法打印出了错误日志,但是对submit
返回的FutureJoinTask
调用 get()
方法,又会抛出异常。于是真相大白,部分批次中的数据存在脏数据,为null值,遍历到该null值的时候出现了异常,但是异常日志在 submit
方法中给catch住,没有打印出来(心痛的感觉),而被捕获的异常,被包装在返回的结果类 FutureJoinTask
中,并没有再次抛出。
如果不需要异步返回结果,请不要用submit 方法
结论先行,我犯的错误就是,浅显的认为submit
和execute
的区别就只是一个有返回异步结果,一个没有返回一步结果,但是事实是残酷的。在submit()
中逻辑一定包含了将异步任务抛出的异常捕获,而因为使用方法不当而导致该异常没有再次抛出。
现在提出一个问题,ForkJoinPool#submit()
中返回的ForkJoinTask
可以获取异步任务的结果,现这个异步抛出了异常,我们尝试获取该任务的结果会是如何? 我们直接看ForkJoinTask#get()
的源码。
public final V get() throws InterruptedException, ExecutionException {
int s = (Thread.currentThread() instanceof ForkJoinWorkerThread) ?
doJoin() : externalInterruptibleAwaitDone();
Throwable ex;
if ((s &= DONE_MASK) == CANCELLED)
throw new CancellationException();
//这里可以直接看到,异步任务出现异常会在调用get()获取结果的时候,会被包装成ExecutionException再次抛出
if (s == EXCEPTIONAL && (ex = getThrowableException()) != null)
throw new ExecutionException(ex);
return getRawResult();
}
异步任务出现异常会在调用get()获取结果的时候,会被包装成ExecutionException再次抛出,但是异常是在哪里被捕获的呢?万变不离其宗,所有线程的线程都需要重写Thread#run()方法, 投递到ForkJoinPool的线程会被包装成ForkJoinWorkerThread,因此我们看一下ForkJoinWorkerThread#run()的实现.
public void run() {
if (workQueue.array == null) { // only run once
Throwable exception = null;
try {
onStart();
pool.runWorker(workQueue);
} catch (Throwable ex) {
//出现异常,捕获,再次抛出会在调用ForkJoinTask#get()的时候
exception = ex;
} finally {
try {
onTermination(exception);
} catch (Throwable ex) {
if (exception == null)
exception = ex;
} finally {
pool.deregisterWorker(this, exception);
}
}
}
}
上面的分析是基于ForkJoinPool的,是不是所有的线程池的submit和execute方法的实现都是类似这样,我们常用的线程池ThreadPoolThread实现会是怎样的,同样的思路,我们需要找到投递到ThreadPoolThread的异步任务最终被包装为哪个Thread的子类或者是实现java.lang.Runnable#run,答案就是java.util.concurrent.FutureTask
public void run() {
...
try {
Callable<V> c = callable;
if (c != null && state == NEW) {
V result;
boolean ran;
try {
result = c.call();
ran = true;
} catch (Throwable ex) {
//捕获异常
result = null;
ran = false;
setException(ex);
}
if (ran)
set(result);
}
}
....
}
总结
java.util.concurrent.ExecutorService#submit(java.lang.Runnable)为何线程池会有这种设定,实际上我们的思路不应该局限于线程池,而是放在获取异步任务结果,异常是否也是属于异步结果,FutureTask作为JDK提供的并发工具类的实现中,已经给出了很好的答案,即获取异步任务结果,异常也是属于异步结果,如果异步任务出现运行时异常,那么在获取该任务的结果时,该异常会被重新包装抛出。
- 撞库扫号防范
- 分享一个 HTTPS A+ 的 nginx 配置
- Vijos P1785 同学排序【模拟】
- Vijos P1784 数字统计【模拟】
- 网络安全黑暗森林法则:2015 ISC 深度回顾
- Codeforces 626G Raffles(贪心+线段树)
- window.opener.location 安全风险讨论
- Vijos P1497 立体图【模拟】
- Vijos P1127 级数求和【模拟】
- 新型漏洞:利用浏览器Cookie绕过HTTPS并窃取私人信息
- Vijos P1113 不高兴的津津【模拟】
- Linux下MySQL的彻底卸载和安装配置字符集
- Codeforces 626F Group Projects(滚动数组+差分dp)
- Vijos P1103 校门外的树【线段树,模拟】
- 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 数组属性和方法
- php设计模式之适配器模式原理、用法及注意事项详解
- php基于Redis消息队列实现的消息推送的办法
- tp5框架无刷新分页实现方法分析
- PHP与SQL语句写一句话木马总结
- golang实现php里的serialize()和unserialize()序列和反序列办法详解
- php和html的区别点详细总结
- 详解在YII2框架中使用UEditor编辑器发布文章
- 在Laravel的Model层做数据缓存的实现
- Thinkphp5.0 框架使用模型Model添加、更新、删除数据操作详解
- php连接mysql数据库最简单的实现方法
- 解决laravel id非自增 模型取回为0 的问题
- PHP7.0连接DB操作实例分析【基于mysqli】
- Laravel获取当前请求的控制器和方法以及中间件的例子
- laravel中数据显示方法(默认值和下拉option默认选中)
- laravel请求参数校验方法