深度解析Java線程池的異常處理機制 #3
阿新 • • 發佈:2018-07-20
ima 修飾 lock link pin inf vat time mdict 前言
今天小夥伴遇到個小問題,線程池提交的任務如果沒有catch異常,那麽會拋到哪裏去,之前倒是沒研究過,本著實事求是的原則,看了一下代碼。
正文
小問題
考慮下面這段代碼,有什麽區別呢?你可以猜猜會不會有異常打出呢?如果打出來的話是在哪裏?:
ExecutorService threadPool = Executors.newFixedThreadPool(1); threadPool.submit(() -> { Object obj = null; System.out.println(obj.toString()); }); threadPool.execute(() -> { Object obj = null; System.out.println(obj.toString()); });
源碼解析
我們下面就來看下代碼, 其實就是將我們提交過去的Runnable包裝成一個Future
public Future<?> submit(Runnable task) { if (task == null) throw new NullPointerException(); RunnableFuture<Void> ftask = newTaskFor(task, null); execute(ftask); return ftask; } protected <T> RunnableFuture<T> newTaskFor(Runnable runnable, T value) { return new FutureTask<T>(runnable, value); } public FutureTask(Runnable runnable, V result) { this.callable = Executors.callable(runnable, result); this.state = NEW; // volatile修飾,保證多線程下的可見性,可以看看Java內存模型 } public static <T> Callable<T> callable(Runnable task, T result) { if (task == null) throw new NullPointerException(); return new RunnableAdapter<T>(task, result); } static final class RunnableAdapter<T> implements Callable<T> { final Runnable task; final T result; RunnableAdapter(Runnable task, T result) { this.task = task; this.result = result; } public T call() { task.run(); return result; } }
接下來就會實際提交到隊列中交給線程池調度處理:
/** * 代碼還是很清爽的,一個很典型的生產者/消費者模型, * 這裏暫不糾結這些細節,那麽如果提交到workQueue成功的話,消費者是誰呢? * 明顯在這個newWorker裏搞的鬼,同樣細節有興趣可以自己再去研究,這裏我們會發現 * 核心就是Worker這個內部類 */ public void execute(Runnable command) { if (command == null) throw new NullPointerException(); int c = ctl.get(); if (workerCountOf(c) < corePoolSize) { if (addWorker(command, true)) return; c = ctl.get(); } if (isRunning(c) && workQueue.offer(command)) { int recheck = ctl.get(); if (! isRunning(recheck) && remove(command)) reject(command); else if (workerCountOf(recheck) == 0) addWorker(null, false); } else if (!addWorker(command, false)) reject(command); }
那麽接下來看看線程池核心的流程:
private final class Worker
extends AbstractQueuedSynchronizer
implements Runnable{
/** Delegates main run loop to outer runWorker */
public void run() {
runWorker(this);
}
}
final void runWorker(Worker w) {
Thread wt = Thread.currentThread();
Runnable task = w.firstTask;
w.firstTask = null;
w.unlock(); // allow interrupts
boolean completedAbruptly = true;
try {
//getTask()方法會嘗試從隊列中抓取數據
while (task != null || (task = getTask()) != null) {
w.lock();
if ((runStateAtLeast(ctl.get(), STOP) ||
(Thread.interrupted() &&
runStateAtLeast(ctl.get(), STOP))) &&
!wt.isInterrupted())
wt.interrupt();
try {
//可覆寫此方法打日誌埋點之類的
beforeExecute(wt, task);
Throwable thrown = null;
try {
//簡單明了,直接調用run方法
task.run();
} catch (RuntimeException x) {
thrown = x; throw x;
} catch (Error x) {
thrown = x; throw x;
} catch (Throwable x) {
thrown = x; throw new Error(x);
} finally {
afterExecute(task, thrown);
}
} finally {
task = null;
w.completedTasks++;
w.unlock();
}
}
completedAbruptly = false;
} finally {
processWorkerExit(w, completedAbruptly);
}
}
submit的方式
那麽我們可以這裏是直接調用的run方法,先看submit的方式,我們知道最終傳遞過去的是一個FutureTask,也就是說會調用這裏的run方法,我們看看實現:
public void run() {
if (state != NEW ||
!UNSAFE.compareAndSwapObject(this, runnerOffset,
null, Thread.currentThread()))
return;
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);
}
} finally {
//省略
}
protected void setException(Throwable t) {
if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
outcome = t; //賦給了這個變量
UNSAFE.putOrderedInt(this, stateOffset, EXCEPTIONAL); // final state
finishCompletion();
}
}
可以看到其實類似於直接吞掉了,這樣的話我們調用get()方法的時候會拿到, 比如我們可以重寫afterExecute方法,從而可以得到實際的異常:
protected void afterExecute(Runnable r, Throwable t) {
super.afterExecute(r, t);
if (t == null && r instanceof Future<?>) {
try {
//get這裏會首先檢查任務的狀態,然後將上面的異常包裝成ExecutionException
Object result = ((Future<?>) r).get();
} catch (CancellationException ce) {
t = ce;
} catch (ExecutionException ee) {
t = ee.getCause();
} catch (InterruptedException ie) {
Thread.currentThread().interrupt(); // ignore/reset
}
}
if (t != null){
//異常處理
t.printStackTrace();
}
}
execute的方式
那麽如果是直接exeture的方式有啥不同呢?這樣的話傳遞過去的就直接是Runnable,因此就會直接拋出:
try {
task.run();
} catch (RuntimeException x) {
thrown = x; throw x;
} catch (Error x) {
thrown = x; throw x;
} catch (Throwable x) {
thrown = x; throw new Error(x);
} finally {
afterExecute(task, thrown);
}
那麽這裏的異常到底會拋出到哪裏呢, 我們看看JVM具體是怎麽處理的:
if (!destroy_vm || JDK_Version::is_jdk12x_version()) {
// JSR-166: change call from from ThreadGroup.uncaughtException to
// java.lang.Thread.dispatchUncaughtException
if (uncaught_exception.not_null()) {
//如果有未捕獲的異常
Handle group(this, java_lang_Thread::threadGroup(threadObj()));
{
KlassHandle recvrKlass(THREAD, threadObj->klass());
CallInfo callinfo;
KlassHandle thread_klass(THREAD, SystemDictionary::Thread_klass());
/*
這裏類似一個方法表,實際就會去調用Thread#dispatchUncaughtException方法
template(dispatchUncaughtException_name, "dispatchUncaughtException")
*/
LinkResolver::resolve_virtual_call(callinfo, threadObj, recvrKlass, thread_klass,
vmSymbols::dispatchUncaughtException_name(),
vmSymbols::throwable_void_signature(),
KlassHandle(), false, false, THREAD);
CLEAR_PENDING_EXCEPTION;
methodHandle method = callinfo.selected_method();
if (method.not_null()) {
JavaValue result(T_VOID);
JavaCalls::call_virtual(&result,
threadObj, thread_klass,
vmSymbols::dispatchUncaughtException_name(),
vmSymbols::throwable_void_signature(),
uncaught_exception,
THREAD);
} else {
KlassHandle thread_group(THREAD, SystemDictionary::ThreadGroup_klass());
JavaValue result(T_VOID);
JavaCalls::call_virtual(&result,
group, thread_group,
vmSymbols::uncaughtException_name(),
vmSymbols::thread_throwable_void_signature(),
threadObj, // Arg 1
uncaught_exception, // Arg 2
THREAD);
}
if (HAS_PENDING_EXCEPTION) {
ResourceMark rm(this);
jio_fprintf(defaultStream::error_stream(),
"\nException: %s thrown from the UncaughtExceptionHandler"
" in thread \"%s\"\n",
pending_exception()->klass()->external_name(),
get_thread_name());
CLEAR_PENDING_EXCEPTION;
}
}
}
可以看到這裏最終會去調用Thread#dispatchUncaughtException方法:
private void dispatchUncaughtException(Throwable e) {
//默認會調用ThreadGroup的實現
getUncaughtExceptionHandler().uncaughtException(this, e);
}
?
public void uncaughtException(Thread t, Throwable e) {
if (parent != null) {
parent.uncaughtException(t, e);
} else {
Thread.UncaughtExceptionHandler ueh =
Thread.getDefaultUncaughtExceptionHandler();
if (ueh != null) {
ueh.uncaughtException(t, e);
} else if (!(e instanceof ThreadDeath)) {
//可以看到會打到System.err裏面
System.err.print("Exception in thread \""
+ t.getName() + "\" ");
e.printStackTrace(System.err);
}
}
}
這裏如果環境是tomcat的話最終會打到catalina.out:
總結
對於線程池、包括線程的異常處理推薦一下方式:
1 直接try/catch,個人 基本都是用這種方式
2 線程直接重寫整個方法:
Thread t = new Thread();
t.setUncaughtExceptionHandler(new Thread.UncaughtExceptionHandler() {
public void uncaughtException(Thread t, Throwable e) {
LOGGER.error(t + " throws exception: " + e);
}
});
//如果是線程池的模式:
ExecutorService threadPool = Executors.newFixedThreadPool(1, r -> {
Thread t = new Thread(r);
t.setUncaughtExceptionHandler(
(t1, e) -> LOGGER.error(t1 + " throws exception: " + e));
return t;
});
3 也可以直接重寫protected void afterExecute(Runnable r, Throwable t) { }
方法
深度解析Java線程池的異常處理機制 #3