Java多线程编程模式实战指南（三）：Two-phase Termination模式

停止线程是一个目标简单而实现却不那么简单的任务。首先，Java没有提供直接的API用于停止线程。此外，停止线程时还有一些额外的细节需要考虑，如待停止的线程处于阻塞（等待锁）或者等待状态（等待其它线程）、尚有未处理完的任务等。本文介绍的Two-phase Termination模式提供了一种通用的用于优雅地停止线程的方法。

Two-phase Termination模式简介

Java并没有提供直接的API用于停止线程。Two-phase Termination模式通过将停止线程这个动作分解为准备阶段和执行阶段这两个阶段，以应对停止线程过程中可能存在的问题。

准备阶段。该阶段主要动作是“通知”目标线程（欲停止的线程）准备进行停止。这一步会设置一个标志变量用于指示目标线程可以准备停止了。但是，由于目标线程可能正处于阻塞状态（等待锁的获得）、等待状态（如调用Object.wait）或者I/O（如InputStream.read）等待等状态，即便设置了这个标志，目标线程也无法立即“看到”这个标志而做出相应动作。因此，这一阶段还需要通过调用目标线程的interrupt方法，以期望目标线程能够通过捕获相关的异常侦测到该方法调用，从而中断其阻塞状态、等待状态。对于能够对interrupt方法调用作出响应的方法（参见表1），目标线程代码可以通过捕获这些方法抛出的InterruptedException来侦测线程停止信号。但也有一些方法（如InputStream.read）并不对interrupt调用作出响应，此时需要我们手工处理，如同步的Socket I/O操作中通过关闭socket，使处于I/O等待的socket抛出java.net.SocketException。

表 1. 能够对Thread.interrupt作出响应的一些方法

　　　　　　　　　　　　　　方法　　　　　　　　　　　　　　　　　　　　　　　　　　　　响应interrupt调用抛出的异常

Object.wait() 、 Object.wait(long timeout) 、Object.wait(long timeout, int nanos)　　　　 InterruptedException

Thread.sleep(long millis) 、Thread.sleep(long millis, int nanos)　　　　　　　　　　　　　　InterruptedException

Thread.join()、Thread.join(long millis) 、Thread.join(long millis, int nanos)　　　　　　　 InterruptedException

java.util.concurrent.BlockingQueue.take()　　　　　　　　　　　　　　　　　　　　　　　　InterruptedException

java.util.concurrent.locks.Lock.lockInterruptibly()　　　　　　　　　　　　　　　　　　　　 InterruptedException

java.nio.channels.InterruptibleChannel　　　　　　　　　　　　　　　　　　　　　　　　　 java.nio.channels.ClosedByInterruptException

执行阶段。该阶段的主要动作是检查准备阶段所设置的线程停止标志和信号，在此基础上决定线程停止的时机，并进行适当的“清理”操作。

Two-phase Termination模式的架构

Two-phase Termination模式的主要参与者有以下几种。其类图如图1所示。

图 1. Two-phase Termination模式的类图

public class ProducerConsumerStop {
class SampleConsumer<P> {
private final BlockingQueue<P> queue = new LinkedBlockingQueue<P>();

private AbstractTerminatableThread workThread
= new AbstractTerminatableThread() {
@Override
protected void doRun() throws Exception {
terminationToken.reservations.decrementAndGet();
P product = queue.take();
// ...
System.out.println(product);
}

};

public void placeProduct(P product) {
if (workThread.terminationToken.isToShutdown()) {
throw new IllegalStateException("Thread shutdown");
}
try {
queue.put(product);
workThread.terminationToken.reservations.incrementAndGet();
} catch (InterruptedException e) {

}
}

public void shutdown() {
workThread.terminate();
}

public void start() {
workThread.start();
}
}

public void test() {
final SampleConsumer<String> aConsumer = new SampleConsumer<String>();

AbstractTerminatableThread aProducer = new AbstractTerminatableThread() {
private int i = 0;

@Override
protected void doRun() throws Exception {
aConsumer.placeProduct(String.valueOf(i));
}

@Override
protected void doCleanup(Exception cause) {
// 生产者线程停止完毕后再请求停止消费者线程
aConsumer.shutdown();
}

};

aProducer.start();
aConsumer.start();
}
}

View Code

隐藏而非暴露可停止的线程

为了保证可停止的线程不被其它代码误停止，一般我们将可停止线程隐藏在线程拥有者背后，而使系统中其它代码无法直接访问该线程，正如本案例代码（见清单1）所展示：AlarmMgr定义了一个private字段alarmSendingThread用于引用告警发送线程（可停止的线程），系统中的其它代码只能通过调用AlarmMgr的shutdown方法来请求该线程停止，而非通过引用该线程对象自身来停止它。

总结

本文介绍了Two-phase Termination模式的意图及架构。并结合笔者工作经历提供了一个实际的案例用于展示一个可复用的Two-phase Termination模式实现代码，在此基础上对该模式进行了评价并分享在实际运用该模式时需要注意的事项。