深度解析Java執行緒池的異常處理機制
阿新 • • 發佈:2018-12-22
前言
今天小夥伴遇到個小問題,執行緒池提交的任務如果沒有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(Runnablerunnable, 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 ( 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) { }
方法