Java并发学习笔记(二)-Executor捕获异常机制

学习《java编程思想》的Executor捕获异常的时候，发现代码输出跟书上有出入，于是就研究了一下Executor的机制。

（一）异常捕获实例

1、异常处理类MyUncaughtExceptionHandler

public class MyUncaughtExceptionHandler implements Thread.UncaughtExceptionHandler{

@Override
public void uncaughtException(Thread t, Throwable e) {
System.out.println("I caught: " + e);
}

}

线程抛出异常的时候，会由MyUncaughtExceptionHandler类的uncaughtException方法捕获异常，在方法里可以对异常进行处理，参数Thread为抛出异常的线程，throwable为异常。

2、线程工厂类HandlerThreadFactory

public class HandlerThreadFactory implements ThreadFactory {

@Override
public Thread newThread(Runnable r) {
Thread t = new Thread(r);
t.setUncaughtExceptionHandler(new MyUncaughtExceptionHandler());
System.out.println(1);
return t;
}

}

通过线程工厂创建的线程，都有统一的配置，比如，设置守护线程，异常处理器等。这里，为所有线程设置统一个异常处理器MyUncaughtExceptionHandler

3、任务线程类ExceptionThread

public class ExceptionThread implements Runnable {

@Override
public void run() {
System.out.println(2);
throw new RuntimeException();
}

}

线程的run方法抛出异常。在正常情况下，这个异常是无法捕获的，但是在Executor的帮助下，该异常可以捕获。

4、场景类

public class CaptureUncaughtException {

public static void main(String[] args) throws InterruptedException {
ExecutorService es = Executors.newCachedThreadPool(new HandlerThreadFactory());
es.execute(new ExceptionThread());
}

}

为Executors设置ThreadFactory，这个通过Executors创建的线程都会有统一的配置，在抛出异常时由MyUncaughtExceptionHandler捕获。

按照《java编程思想》里，该代码应该输出如下：

1
2
I caught: java.lang.RuntimeException

但是在我的环境里，输出如下：

1
2
1
I caught: java.lang.RuntimeException

可以看出来，ThreadFactory的newThread执行了两次。

（二）Executor捕获异常机制

这里以ThreadPoolExecutor为例，其execute方法如下：

public void execute(Runnable command) {
if (command == null)
throw new NullPointerException();
int c = ctl.get();
//......
else if (!addWorker(command, false))
reject(command);
}

因为程序第一次启动，线程池没有线程，所以执行了addWorker方法，任务以参数command的形式传递到addWorker里。

private boolean addWorker(Runnable firstTask, boolean core) {
//......

boolean workerStarted = false;
boolean workerAdded = false;
Worker w = null;
try {
final ReentrantLock mainLock = this.mainLock;
//任务当做firstTask传递给worker
w = new Worker(firstTask);
final Thread t = w.thread;
if (t != null) {
mainLock.lock();
try {
// Recheck while holding lock.
// Back out on ThreadFactory failure or if
// shut down before lock acquired.
int c = ctl.get();
int rs = runStateOf(c);

if (rs < SHUTDOWN ||
(rs == SHUTDOWN && firstTask == null)) {
if (t.isAlive()) // precheck that t is startable
throw new IllegalThreadStateException();
//worker创建成功以后，添加到workers里。workers其实就是线程池里的线程
workers.add(w);
int s = workers.size();
if (s > largestPoolSize)
largestPoolSize = s;
workerAdded = true;
}
} finally {
mainLock.unlock();
}
if (workerAdded) {
//启动worker的线程
t.start();
workerStarted = true;
}
}
} finally {
if (! workerStarted)
addWorkerFailed(w);
}
return workerStarted;
}

在worker的构造函数里，进行了一个主要的操作

private final class Worker
extends AbstractQueuedSynchronizer
implements Runnable
{
Worker(Runnable firstTask) {
setState(-1); // inhibit interrupts until runWorker
this.firstTask = firstTask;
this.thread = getThreadFactory().newThread(this);
}

/** Delegates main run loop to outer runWorker  */
public void run() {
runWorker(this);
}

//......其他方法
}

任务成为Worker的成员变量，并以worker自身为参数，以我们自己创建的线程工厂为模板，创建了一个线程，所以打印出“1”，该线程在addWorker方法里启动。

可以看到，worker的run方法调用了runWorker方法

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) {
//.......其他代码
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);
}
}

worker线程里，调用了任务的run方法，该run方法仅仅是一个普通方法，并不是启动线程，所以在我们的ExceptionThread的run方法里抛出异常时，可以被捕获。从中，我们也可以看到，Executor的execute方法，是在Executor自己创建的worker线程里，将我们自己写的runnable任务作为一个普通对象，执行run方法。在这里输出“2”，并捕获了异常。到此，就是所有正常的输出。

我们可以看到最后一个finally里，对work进行了处理

private void processWorkerExit(Worker w, boolean completedAbruptly) {
//......一些逻辑
if (runStateLessThan(c, STOP)) {
if (!completedAbruptly) {
int min = allowCoreThreadTimeOut ? 0 : corePoolSize;
if (min == 0 && ! workQueue.isEmpty())
min = 1;
if (workerCountOf(c) >= min)
return; // replacement not needed
}
addWorker(null, false);
}
}

在我的代码里，抛出异常后，对worker退出时，会重新创建一个worker，这里又回到了worker的创建过程，线程工厂会以模板创建一个worker，newThread又执行了一次，又输出了“1”。