摘要:接下來就是會把任務提交到隊列中給線程池調度處理因為主要關心的是這個線程怎么執行,異常的拋出和處理,所以我們暫時不解析多余的邏輯。
前言
今天遇到了一個bug,現象是,一個任務放入線程池中,似乎“沒有被執行”,日志也沒有打。
經過本地代碼調試之后,發現在任務邏輯的前半段,拋出了NPE,但是代碼外層沒有try-catch,導致這個異常被吃掉。
這個問題解決起來是很簡單的,外層加個try-catch就好了,但是這個異常如果沒有被catch,線程池內部邏輯是怎么處理這個異常的呢?這個異常最后會跑到哪里呢?
帶著疑問和好奇心,我研究了一下線程池那一塊的源碼,并且做了以下的總結。
源碼分析項目中出問題的代碼差不多就是下面這個樣子
ExecutorService threadPool = Executors.newFixedThreadPool(3); threadPool.submit(() -> { String pennyStr = null; Double penny = Double.valueOf(pennyStr); ... })
先進到newFixedThreadPool這個工廠方法中看生成的具體實現類,發現是ThreadPoolExecutor
public static ExecutorService newFixedThreadPool(int nThreads) { return new ThreadPoolExecutor(nThreads, nThreads, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue()); }
再看這個類的繼承關系,
再進到submit方法,這個方法在ExecutorService接口中約定,其實是在AbstractExectorService中實現,ThreadPoolExecutor并沒有override這個方法。
public Future> submit(Runnable task) { if (task == null) throw new NullPointerException(); RunnableFutureftask = newTaskFor(task, null); execute(ftask); return ftask; } protected RunnableFuture newTaskFor(Runnable runnable, T value) { return new FutureTask (runnable, value); }
對應的FutureTask對象的構造方法
public FutureTask(Runnable runnable, V result) { this.callable = Executors.callable(runnable, result); this.state = NEW; // state由volatile 修飾 保證多線程下的可見性 }
對應Callable 對象的構造方法
public staticCallable callable(Runnable task, T result) { if (task == null) throw new NullPointerException(); return new RunnableAdapter (task, result); }
對應RunnableAdapter 對象的構造方法
/** * A callable that runs given task and returns given result * 一個能執行所給任務并且返回結果的Callable對象 */ static final class RunnableAdapterimplements Callable { final Runnable task; final T result; RunnableAdapter(Runnable task, T result) { this.task = task; this.result = result; } public T call() { task.run(); return result; } }
總結上面的,newTaskFor就是把我們提交的Runnable 對象包裝成了一個Future。
接下來就是會把任務提交到隊列中給線程池調度處理:
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); }
因為主要關心的是這個線程怎么執行,異常的拋出和處理,所以我們暫時不解析多余的邏輯。很容易發現,如果任務要被執行,肯定是進到了addWorker方法當中,所以我們再進去看,鑒于addWorker方法的很長,不想列太多的代碼,我就摘了關鍵代碼段:
private boolean addWorker(Runnable firstTask, boolean core) { ... boolean workerStarted = false; boolean workerAdded = false; Worker w = null; try { // 實例化一個worker對象 w = new Worker(firstTask); final Thread t = w.thread; if (t != null) { final ReentrantLock mainLock = this.mainLock; mainLock.lock(); try { int rs = runStateOf(ctl.get()); if (rs < SHUTDOWN || (rs == SHUTDOWN && firstTask == null)) { if (t.isAlive()) // precheck that t is startable throw new IllegalThreadStateException(); workers.add(w); int s = workers.size(); if (s > largestPoolSize) largestPoolSize = s; workerAdded = true; } } finally { mainLock.unlock(); } if (workerAdded) { // 從Worker對象的構造方法看,當這個thread對象start之后, // 之后實際上就是調用Worker對象的run() t.start(); workerStarted = true; } } } finally { if (! workerStarted) addWorkerFailed(w); } return workerStarted; } // Worker的構造方法 Worker(Runnable firstTask) { setState(-1); // inhibit interrupts until runWorker this.firstTask = firstTask; this.thread = getThreadFactory().newThread(this); }
我們再看這個ThreadPoolExecutor的內部類Worker對象:
private final class Worker extends AbstractQueuedSynchronizer implements Runnable { ... /** Delegates main run loop to outer runWorker */ public void run() { runWorker(this); } ... }
看來真正執行任務的是在這個外部的runWorker當中,讓我們再看看這個方法是怎么消費Worker線程的。
final void runWorker(Worker w) { Thread wt = Thread.currentThread(); Runnable task = w.firstTask; w.firstTask = null; w.unlock(); // allow interrupts boolean completedAbruptly = true; try { 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; // ==== 關鍵代碼 start ==== 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); } // ==== 關鍵代碼 end ==== } finally { task = null; w.completedTasks++; w.unlock(); } } completedAbruptly = false; } finally { processWorkerExit(w, completedAbruptly); } }
終于走到底了,可以看到關鍵代碼中的try-catch block代碼塊中,調用了本次執行任務的run方法。
// ==== 關鍵代碼 start ==== 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); } // ==== 關鍵代碼 end ====
可以看到捕捉了異常之后,會再向外拋出,只不過再finally block 中有個afterExecute()方法,似乎在這里是可以處理這個異常信息的,進去看看
protected void afterExecute(Runnable r, Throwable t) { }
可以看到ThreadPoolExecutor#afterExecute()方法中,是什么都沒做的,看來是讓使用者通過override這個方法來定制化任務執行之后的邏輯,其中可以包括異常處理。
那么這個異常到底是拋到哪里去了呢。我在一個大佬的文章找到了hotSpot 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(), " Exception: %s thrown from the UncaughtExceptionHandler" " in thread "%s" ", pending_exception()->klass()->external_name(), get_thread_name()); CLEAR_PENDING_EXCEPTION; } } }
代碼是C寫的,有興趣可以去全文,根據英文注釋能稍微看懂一點
http://hg.openjdk.java.net/jd...
可以看到這里最終會去調用Thread#dispatchUncaughtException方法:
/** * Dispatch an uncaught exception to the handler. This method is * intended to be called only by the JVM. */ private void dispatchUncaughtException(Throwable e) { getUncaughtExceptionHandler().uncaughtException(this, e); }
/** * Called by the Java Virtual Machine when a thread in this * thread group stops because of an uncaught exception, and the thread * does not have a specific {@link Thread.UncaughtExceptionHandler} * installed. * */ 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); } } }
jdk的注釋也說明的很清楚了,當一個線程拋出了一個未捕獲的異常,JVM會去調用這個方法。如果當前線程沒有聲明UncaughtExceptionHandler成員變量并且重寫uncaughtException方法的時候,就會看線程所屬的線程組(如果有線程組的話)有沒有這個類,沒有就會打到System.err里面。
IBM這篇文章也提倡我們使用ThreadGroup 提供的 uncaughtException 處理程序來在線程異常終止時進行檢測。
https://www.ibm.com/developer...總結 (解決方法)
從上述源碼分析中可以看到,對于本篇的異常“被吃掉”的問題,有以下幾種方法
用try-catch 捕捉,一般都是用這種
線程或者線程組對象設置UncaughtExceptionHandler成員變量
Thread t = new Thread(r); t.setUncaughtExceptionHandler( (t1, e) -> LOGGER.error(t1 + " throws exception: " + e)); return t;
override 線程池的afterExecute方法。
本篇雖然是提出問題的解決方法,但主旨還是分析源碼,了解了整個過程中異常的經過的流程,希望能對您產生幫助。
參考https://www.jcp.org/en/jsr/de...
https://www.ibm.com/developer...
http://ifeve.com/%E6%B7%B1%E5...
文章版權歸作者所有,未經允許請勿轉載,若此文章存在違規行為,您可以聯系管理員刪除。
轉載請注明本文地址:http://specialneedsforspecialkids.com/yun/74301.html
摘要:先看寫的簡略的代碼線程池中發現異常,被中斷線程池中發現異常,被中斷我這是一個訂單處理流程,主要用到了一個方法,就是。好了,以上就是對線程池異常捕捉的一個記錄。 開發自己的項目有一段時間了,因為是個長時間跑的服務器端程序,所以異常處理顯得尤為重要。 對于異常的抓取和日志(狹義上的日志)的分析一點都不能落下。 我們使用了Java自帶的Executor模塊,我只是稍微看了下Executor...
摘要:進程線程與協程它們都是并行機制的解決方案。選擇是任意性的,并在對實現做出決定時發生。線程池的大小一旦達到最大值就會保持不變,如果某個線程因為執行異常而結束,那么線程池會補充一個新線程。此線程池支持定時以及周期性執行任務的需求。 并發與并行的概念 并發(Concurrency): 問題域中的概念—— 程序需要被設計成能夠處理多個同時(或者幾乎同時)發生的事件 并行(Parallel...
摘要:本文對多線程基礎知識進行梳理,主要包括多線程的基本使用,對象及變量的并發訪問,線程間通信,的使用,定時器,單例模式,以及線程狀態與線程組。源碼采用構建,多線程這部分源碼位于模塊中。通知可能等待該對象的對象鎖的其他線程。 本文對多線程基礎知識進行梳理,主要包括多線程的基本使用,對象及變量的并發訪問,線程間通信,lock的使用,定時器,單例模式,以及線程狀態與線程組。 寫在前面 花了一周時...
摘要:基礎問題的的性能及原理之區別詳解備忘筆記深入理解流水線抽象關鍵字修飾符知識點總結必看篇中的關鍵字解析回調機制解讀抽象類與三大特征時間和時間戳的相互轉換為什么要使用內部類對象鎖和類鎖的區別,,優缺點及比較提高篇八詳解內部類單例模式和 Java基礎問題 String的+的性能及原理 java之yield(),sleep(),wait()區別詳解-備忘筆記 深入理解Java Stream流水...
閱讀 2076·2023-04-25 19:03
閱讀 1221·2021-10-14 09:42
閱讀 3399·2021-09-22 15:16
閱讀 946·2021-09-10 10:51
閱讀 1545·2021-09-06 15:00
閱讀 2401·2019-08-30 15:55
閱讀 485·2019-08-29 16:22
閱讀 893·2019-08-26 13:49