解析Java線程池的異常處理機制

darkerXi 發布于2019-08-16 10:39 / 3238人閱讀

摘要：前言今天小伙伴遇到個小問題，線程池提交的任務如果沒有異常，那么會拋到哪里去，之前倒是沒研究過，本著實事求是的原則，看了一下代碼。

前言

今天小伙伴遇到個小問題，線程池提交的任務如果沒有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 ftask = newTaskFor(task, null);
        execute(ftask);
        return ftask;
    }
    protected  RunnableFuture newTaskFor(Runnable runnable, T value) {
        return new FutureTask(runnable, value);
    }
    public FutureTask(Runnable runnable, V result) {
        this.callable = Executors.callable(runnable, result);
        this.state = NEW;       // volatile修飾，包裝多線程下的可見性
    }
    public static  Callable callable(Runnable task, T result) {
        if (task == null)
            throw new NullPointerException();
        return new RunnableAdapter(task, result);
    }
    
    static final class RunnableAdapter implements 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;
        }
    }

接下來就會實際提交到隊列中交給線程池調度處理:

    /**
    * 代碼還是很清爽的，一個很典型的生產者/消費者模型，
    * 這里暫不糾結這些細節，那么如果提交到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 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(),
                "
Exception: %s thrown from the UncaughtExceptionHandler"
                " in thread "%s"
",
                pending_exception()->klass()->external_name(),
                get_thread_name());
          CLEAR_PENDING_EXCEPTION;
        }
      }
    }

可以看到這里最終會去調用Thread#dispatchUncaughtException方法:

    private void dispatchUncaughtException(Throwable e) {
          //默認會調用ThreadGroup的實現
        getUncaughtExceptionHandler().uncaughtException(this, e);
    }

    //ThreadGroup
    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:

總結

對于線程池、包括線程的異常處理推薦一下方式:

直接try/catch，個人基本都是用這種方式

線程直接重寫整個方法:

       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) { }方法

云服務器 GPU云服務器異常處理機制 java線程池的特點 java線程池的應用 java線程池的使用

文章版權歸作者所有，未經允許請勿轉載,若此文章存在違規行為，您可以聯系管理員刪除。

轉載請注明本文地址：http://m.specialneedsforspecialkids.com/yun/70213.html

追蹤解析 ThreadPoolExecutor 源碼

摘要：的前位數用來表示線程的數量，后面三位用來表示線程池的狀態。線程池的狀態有五種，分別是，根據單詞就能猜出大概。并且為了考慮性能問題，線程池的設計沒有使用悲觀鎖關鍵字，而是大量使用了和機制。零前期準備 0 FBI WARNING 文章異常啰嗦且繞彎。 1 版本 JDK 版本 : OpenJDK 11.0.1 IDE : idea 2018.3 2 ThreadPoolExecutor ...

gaomysion 2019-08-16 14:41 評論0 收藏0
淺談Java并發編程系列（六） —— 線程池的使用

摘要：線程池的作用降低資源消耗。通過重復利用已創建的線程降低線程創建和銷毀造成的資源浪費。而高位的部分，位表示線程池的狀態。當線程池中的線程數達到后，就會把到達的任務放到中去線程池的最大長度。默認情況下，只有當線程池中的線程數大于時，才起作用。線程池的作用降低資源消耗。通過重復利用已創建的線程降低線程創建和銷毀造成的資源浪費。提高響應速度。當任務到達時，不需要等到線程創建就能立即執行...

Vicky 2019-08-14 15:52 評論0 收藏0
Java線程池

摘要：中的線程池是運用場景最多的并發框架。才是真正的線程池。存放任務的隊列存放需要被線程池執行的線程隊列。所以線程池的所有任務完成后，它最終會收縮到的大小。飽和策略一般情況下，線程池采用的是，表示無法處理新任務時拋出異常。 Java線程池 1. 簡介系統啟動一個新線程的成本是比較高的，因為它涉及與操作系統的交互，這個時候使用線程池可以提升性能，尤其是需要創建大量聲明周期很短暫的線程時。Ja...

jerry 2019-08-15 18:06 評論0 收藏0
后端ing

摘要：當活動線程核心線程非核心線程達到這個數值后，后續任務將會根據來進行拒絕策略處理。線程池工作原則當線程池中線程數量小于則創建線程，并處理請求。當線程池中的數量等于最大線程數時默默丟棄不能執行的新加任務，不報任何異常。 spring-cache使用記錄 spring-cache的使用記錄，坑點記錄以及采用的解決方案深入分析 java 線程池的實現原理在這篇文章中，作者有條不紊的將 ja...

roadtogeek 2019-08-15 13:49 評論0 收藏0
Java中的線程池

摘要：中的線程池運用場景非常廣泛，幾乎所有的一步或者并發執行程序都可以使用。代碼中如果執行了方法，線程池會提前創建并啟動所有核心線程。線程池最大數量線程池允許創建的線程最大數量。被稱為是可重用固定線程數的線程池。 Java中的線程池運用場景非常廣泛，幾乎所有的一步或者并發執行程序都可以使用。那么線程池有什么好處呢，以及他的實現原理是怎么樣的呢？使用線程池的好處在開發過程中，合理的使用線程...

tomato 2019-08-19 11:46 評論0 收藏0