Java并发实践（七）取消与关闭-白红宇

Java没有提供任何机制来安全的终止线程（虽然Thread.stop和suspend方法提供了这样的机制，但由于存在着一些严重的缺陷，因此应该避免使用）

任务取消

可取消的（Cancellable）：

如果外部代码能在某个操作正常完成之前将其置入“完成”状态，那么这个操作就可以成为可取消的。

取消某个操作的原因

用户取消请求：用户点击图形界面程序中的“取消”按钮，或者通过管理接口来发出取消请求。

有时间限制的操作：例如，某个应用程序需要在有限时间内搜索问题空间，并在这个时间内历最佳的解决方案。当计时器超时时，需要取消所有正在搜索的任务。

应用程序事件：例如，应用程序对某个问题空间进行分解并搜索，从而使不同的任务可以搜索问题空间的中的不同区域。但其中一个任务找到了解决方案时，所有其他仍在搜索的任务都可以将被取消。

错误：网页爬虫程序搜索相关页面，并将页面或摘要数据保存到硬盘。当一个爬虫任务发生操作时（例如，磁盘空间已满），那么所有的搜索任务都会被取消，此时可能会记录他们的当前状态，以便稍后重新启动。

关闭：当一个程序或服务关闭时，必须对正在处理和等待处理的工作执行某种操作。在平缓的关闭过程中，当前正在执行的任务将继续执行直到关闭，而在立即关闭过程中，当前的任务则可能取消。

程序1&2：一个仅运行1秒钟的素数生成器

@ThreadSafepublic class PrimeGenerator implements Runnable {    private static ExecutorService exec = Executors.newCachedThreadPool();    @GuardedBy("this")    private final List
    
      primes            = new ArrayList
     
      ();    private volatile boolean cancelled;    public void run() {        BigInteger p = BigInteger.ONE;        while (!cancelled) {            p = p.nextProbablePrime();            synchronized (this) {                primes.add(p);            }        }    }    public void cancel() {        cancelled = true;    }    public synchronized List
      
        get() {        return new ArrayList
       
        (primes);    }    static List
        
          aSecondOfPrimes() throws InterruptedException {        PrimeGenerator generator = new PrimeGenerator();        exec.execute(generator);        try {            SECONDS.sleep(1);        } finally {            generator.cancel();        }        return generator.get();    }}复制代码

素数生成器，通常不会刚好在运行1秒钟之后停止，因为在请求取消的时刻和run方法循环执行下一次检查之间可能存在延迟。

一个可取消的任务必须拥有可取消策略，在这个策略中将详细的定义取消操作的“How”、“When”以及“What”，即代码如何（How）请求取消任务，任务何时（When）检查是否已经请求了取消，以及在响应取消请求时应该执行哪些（What）操作

中断

PrimeGenerator中的取消机制最终会使得搜索素数的任务退出，但在退出过程中需要花费一定时间。然而，如果使用这种的任务调用了一个阻塞方法，例如BlockingQueue.put，那么可能会产生一个更严重的问题——任务可能永远不会检查取消标志，因此永远不会结束。

class BrokenPrimeProducer extends Thread {    private final BlockingQueue
    
      queue;    private volatile boolean cancelled = false;    BrokenPrimeProducer(BlockingQueue
     
       queue) {        this.queue = queue;    }    public void run() {        try {            BigInteger p = BigInteger.ONE;            while (!cancelled)                queue.put(p = p.nextProbablePrime());        } catch (InterruptedException consumed) {        }    }    public void cancel() {        cancelled = true;    }}void consumePrimes () throws InterruptedException {    BlockingQueue
      
        primes = ...;    BrokenPrimeProducer producer = new BrokenPrimeProducer(primes);    producer.start();    try {        while (needMorePrimes())            consume(pimes.take());    } finally {        producer.cancel();    }}复制代码

在Java的API或语言规则中，并没有将中断与任何取消语义管理起来，但实际上，如果在取消之外的其他操作中使用中断，那么都是不合适的，并且很难支撑起来大的应用。

每个线程都有一个boolean的中断状态，当中断线程时，这个线程的中断状态将被设置为true。在Thread中包含了中断线程以及查询线程中断状态的方法。interrupt方法能中断目标线程，而isInterrupt方法能返回目标线程的中断状态。静态的interrupted方法将清除当前线程的中断状态，并返回它之前的值，这也是清除中断状态的唯一方法。

public class Thread {    //中断目标线程    public void interrupt () {...}    //返回目标线程的中断状态    public boolean isInterrupted () {...}    //清除当前线程的中断状态并返回之前的值    public static boolen interrupted () {...}}复制代码

阻塞库方法，例如Thread.sleep和Object.wait等，都会检查线程何时中断，并且在发现中断时提前返回。他们在响应中断时执行的操作包括：清除中断状态，抛出InterruptedException，表示阻塞操作由于中断而提前结束。JVM并不能保证阻塞方法检测到中断的速度，但在实际情况中响应速度还是非常快的。

当线程在非阻塞状态下中断时，它的中断状态将被设置，然后根据将被取消的操作来检查中断状态以及判断发生了中断。通过这样的方法，中断操作将变得“有黏性”——如果不触发InterruptedException，那么中断状态将一直保持，直到明确的清除中断状态。

调用interrupt并不意味着立即停止目标线程正在进行的工作，而只是传递了请求中断的消息。

对中断操作的正确理解是：它并不会真正的中断一个正在运行的线程，而只是发出中断请求，然后由线程在下一个合适的时刻中断自己。（这些时刻也被称为取消点）。有些方法，例如wait、sleep和join等，将严格的处理这种请求，当他们收到中断请求或者在开始执行时发现某个已经被设置好的中断状态时，将抛出一个异常。设计良好的方法可以完全忽略这种请求，只要他们能使调用代码对中断请求进行某种处理。设计糟糕的方法可能会屏蔽中断请求，从而导致调用栈中的其他代码无法对中断请求作出响应。

在使用静态的interrupted时应该小心，因为他会清除当前线程的中断状态。如果在调用interrupted是返回了true，那么除非你想屏蔽这个中断，否则必须处理——可以抛出InterruptedException，或者通过再次调用interrupt来恢复中断状态。

通常，中断是实现取消的最合理方式

程序5：通过中断来取消

public class PrimeProducer extends Thread {    private final BlockingQueue
    
      queue;    PrimeProducer(BlockingQueue
     
       queue) {        this.queue = queue;    }    public void run() {        try {            BigInteger p = BigInteger.ONE;            while (!Thread.currentThread().isInterrupted())                queue.put(p = p.nextProbablePrime());        } catch (InterruptedException consumed) {            /* Allow thread to exit */        }    }    public void cancel() {        interrupt();    }}复制代码

BrokenPrimeProducer中的问题很容解决：使用中断而不是boolean标志来请求取消。在每次循环迭代中，有两个位置可以检查出中断：在阻塞的put方法调用中，以及在循环开始出查询中断状态时。

中断策略

中断策略规定线程如何解释某个中断请求——当发现中断请求时，应该做哪些工作，哪些工作单元对中断来说是原则操作，以及以多快的速度来响应中断。

最合理的中断策略是以某种形式的线程级取消操作或服务级取消操作：尽快退出，在必要时进行清理，通知某个所有者该线程已经退出。

任务不会在自己拥有的线程中执行，而是在某个服务（例如线程池）拥有的线程中执行。对于非线程所有者的代码来说，应该小心的保存中断状态，这样拥有线程的代码才能对中断做出响应。

由于每个线程拥有各自的中断策略，因此除非知道中断对该线程的含义，否则，不应该中断这个线程。

响应中断

两种实用策略可用于处理InterruptedException：

传递异常（可能在执行某个特定域任务的清除操作之后），从而使你的方法也成为可中断的阻塞方法。

恢复中断状态，从而使调用栈中的上层代码能够对其进行处理。

程序6：将InterruptedException传递给调用者

BlockingQueue
    
      queue;...public Task getNextTask() throws InterruptedException {    return queue.take();}复制代码

如果不想或无法传递InterruptedException（或许通过Runnable来定义任务），那么需要寻找另一种方式来保存中断请求。一种标准的方法就是再次调用interrupt来恢复中断状态。

只有实现了线程中断策略的代码才可以屏蔽中断请求，在常规的任务和库代码中，都不应该屏蔽中断请求。

对于一些不支持取消但仍可以调用中断阻塞方法的操作，他们必须在循环中调用这些方法，并在发现中断后重新尝试。

程序7：不可取消的任务在退出前恢复中断

public class NoncancelableTask {    public Task getNextTask(BlockingQueue
    
      queue) {        boolean interrupted = false;        try {            while (true) {                try {                    return queue.take();                } catch (InterruptedException e) {                    interrupted = true;                    // fall through and retry                }            }        } finally {            if (interrupted)                Thread.currentThread().interrupt();        }    }    interface Task {    }}复制代码

示例：计时运行

许多问题永远也无法解决（例如，枚举所有的素数），而某些问题，能很快得到答案，也可能永远得不到答案。这些情况下，如果能够指定“最多花10分钟搜索答案”或者“枚举出在10分钟内能找到的答案”，将是非常有用的。

在程序2中的aSecondOfPrimes方法将启动一个PrimeGenerator，并在1秒钟后中断。尽管PrimeGenerator可能需要超过1秒的时间才能停止，但它最终会发现中断，然后停止，并使线程结束。在执行任务时的另一个方面是，你希望知道在任务执行过程中是否抛出异常。如果PrimeGenerator在指定时限内抛出了一个未检查的异常，那么这个异常可能会被忽略，因为素数生成器在另一个独立的线程中运行，而这个线程不会显示的处理异常。

程序8给出了在指定时间内运行一个任意的Runnable的示例。在调用线程中运行任务，并安排了一个取消任务，在运行指定的时间间隔后中断它。解决了任务中抛出未检查异常的问题，因为该异常会被timedRun的调用者捕获。

程序8：在外部线程中安排中断（不要这么做）

public class TimedRun1 {    private static final ScheduledExecutorService cancelExec = Executors.newScheduledThreadPool(1);    public static void timedRun(Runnable r,                                long timeout, TimeUnit unit) {        final Thread taskThread = Thread.currentThread();        cancelExec.schedule(new Runnable() {            public void run() {                taskThread.interrupt();            }        }, timeout, unit);        r.run();    }}复制代码

这是一种非常简单的方法，但却破坏了以下规则：在中断线程之前，应该了解它的中断策略。由于timedRun可以从任意一个线程中调用，因此它无法知道这个调用线程的中断策略。如果任务在超时之前完成，那么中断timedRun所在线程的取消任务将在timedRun返回到调用者之后启动。

而且，如果任务不响应中断，那么timedRun会在任务结束时才返回，此时可能已经超过了指定的时限（或者还没有超过时限）。如果某个限时运行的任务没有在指定的时间内返回，那么将对调用者带来负面影响。

程序9：在专门的线程中中断任务

public class TimedRun2 {    private static final ScheduledExecutorService cancelExec = newScheduledThreadPool(1);    public static void timedRun(final Runnable r,                                long timeout, TimeUnit unit)            throws InterruptedException {        class RethrowableTask implements Runnable {            //通过一个volatile变量，来存储线程是否异常            private volatile Throwable t;            public void run() {                try {                    r.run();                } catch (Throwable t) {                    this.t = t;                }            }            void rethrow() {                if (t != null)                    throw launderThrowable(t);            }        }        RethrowableTask task = new RethrowableTask();        final Thread taskThread = new Thread(task);        taskThread.start();        //延时timeout个unit单位后 执行线程中断        cancelExec.schedule(new Runnable() {            public void run() {                taskThread.interrupt();            }        }, timeout, unit);        //无论如何都等待；如果线程不响应中断，那么通过join等待任务线程timeout时间后 不再等待，回到调用者线程        taskThread.join(unit.toMillis(timeout));        //如果 任务线程中有异常，则抛出        task.rethrow();    }复制代码

程序9中解决了aSecondOfPrimes的异常处理问题以及之前解决方案中的问题。执行任务的线程拥有自己的执行策略，即使任务不响应中断，限时运行的方法仍能返回到它的调用者。在启动任务线程之后，timedRun将执行一个限时的join方法。在join返回后，它将检查任务中是否有异常抛出，如果有的话，则会在调用timedRun的线程中再次抛出该异常。由于Throwable将在两个线程之间共享，因此该变量被声明为volatile类型，从而确保安全的将其从任务线程发布到timedRun线程。

该代码解决了前面示例中的问题，但由于它依赖于一个限时的join，因此存在join的不足：无法知道执行控制室因为线程正常退出而返回还是因为join超时而返回。

通过Future来实现取消

程序10：通过Future来取消任务

public class TimedRun {    private static final ExecutorService taskExec = Executors.newCachedThreadPool();    public static void timedRun(Runnable r,                                long timeout, TimeUnit unit)            throws InterruptedException {        Future
     task = taskExec.submit(r);        try {            //通过Futrue.get(超时时间)，捕获相应的异常来处理计时运行和取消任务            task.get(timeout, unit);        } catch (TimeoutException e) {            // task will be cancelled below        } catch (ExecutionException e) {            // exception thrown in task; rethrow            throw launderThrowable(e.getCause());        } finally {            // Harmless if task already completed            task.cancel(true); // interrupt if running        }    }}复制代码

我们已经使用了一种抽象机制来管理任务的生命周期，处理异常，以及实现取消，即Future。通常，使用现有库中的类比自行编写更好，因此我们将继续使用Future和任务执行框架来构建timedRun。

ExecutorService.submit将返回一个Future来描述任务。Future拥有一个cancel方法，该方法带有一个boolean类型的参数mayInterruptIfRunning，表示取消操作是否成功（仅表示任务能够接收中断，不表示任务是否检测并处理中断）。如果mayInterruptIfRunning为true并且任务当前正在某个线程中运行，那么这个线程可以被中断。如果mayInterruptIfRunning为false，那么意味着“若任务还没有启动，就不要运行它”，这种方式应该用于那些不处理中断的任务中。

程序10给出了另一个版本的timedRun：将任务提交给一个ExecutorService，并通过一个定时的Future.get来获得结果。如果get在返回时抛出了一个TimeoutException，那么任务将同通过Future来取消（为了简化代码，这个版本的timedRun在finally块中将直接调用Future.cancel，因为取消一个已经完成的任务，不会带来任何影响）。如果任务在取消前就抛出一个异常，那么该异常将被重新抛出以便由调用者来处理。

当Futrue.get抛出InterruptedException或TimeoutException时，如果你知道不再需要结果，那么就可以调用Future.cancel来取消任务。

处理不可中断的阻塞

在Java库中，许多可阻塞的方法都是通过提前返回或者抛出InterruptedException来响应中断请求，从而使开发人员更容易构建出能响应取消请求的任务。如果一个线程由于执行同步的Socket I/O或者等待获得内置锁而阻塞，那么中断请求只能设置线程的中断状态，除此之外，没有其他任何作用。对于那些由于执行不可中断操作而阻塞的线程，可以使用类似于中断的手段来停止这些线程，但这要求我们必须知道线程阻塞的原因。

Java.io包中的同步Socket I/O：在服务器应用程序中，最常见的阻塞I/O形式就是对套接字进行读取和写入。虽然InputStream和OutputStream中的read和write等方法都不会响应中断，但通过关闭底层的套接字，可以使得由于执行read或write等方法而被阻塞的线程抛出一个SocketException。

Java.io包中的同步I/O：当中断一个正在InterruptibleChannel上等待的线程时，将抛出ClosedByInterruptException并关闭链路（这还会使得其他在这条链路上阻塞的线程同样抛出CloaseByInterruptException）。当关闭一个InterrptibleChannel时，将导致所有在链路操作上阻塞的线程都抛出AsynchronousClosedException。

Selector的异步I/O：如果一个线程调用Selector.select方法（java.nio.channels中）时阻塞了，那么调用close或wakeup方法，会使线程抛出ClosedSelectorException并提前返回。

获取某个锁：如果一个线程由于等待某个内置所而阻塞，那么将无法响应中断，因为线程认为它肯定获得锁，所以将不会理会中断请求。

程序11：通过改写interrupt方法将非标准的取消操作封装在Thread中

public class ReaderThread extends Thread {    private static final int BUFSZ = 512;    private final Socket socket;    private final InputStream in;    public ReaderThread(Socket socket) throws IOException {        this.socket = socket;        this.in = socket.getInputStream();    }    @Override    public void interrupt() {        try {            //中断前关闭socket            socket.close();        } catch (IOException ignored) {        } finally {            super.interrupt();        }    }    public void run() {        try {            byte[] buf = new byte[BUFSZ];            while (true) {                int count = in.read(buf);                if (count < 0)                    break;                else if (count > 0)                    processBuffer(buf, count);            }        } catch (IOException e) { /* Allow thread to exit */        }    }    public void processBuffer(byte[] buf, int count) {    }}复制代码

程序11中的ReaderThread给出了如何封装非标准的取消操作。ReaderThread管理了一个套接字链接，它采用同步方式从该套接字中读取数据，并将接收到的数据传递给processBuffer。为了结束某个用户的连接或关闭服务器，ReaderThread改写了interrupt方法，使其既能处理标准的中断，也能关闭底层的套接字。因此，无论ReaderThread线程是在read方法中阻塞还是在某个可中断的阻塞方法中阻塞，都可以被中断并停止执行当前的工作。

采用newTaskFor来封装非标准的取消

Java6在ThreadPoolExecutor中的新增功能。

当把一个Callable提交给ExecutorService时，submit方法会返回一个Future，我们可以通过这Future来取消任务。newTaskFor是一个工厂方法，它将创建Future来代表任务。newTaskFor还能返回一个RunnableFuture接口，该接口扩展了Future和Runnable（并由FutureTask实现）

程序12：通过newTaskFor将非标准的取消操作封装在一个任务中

public abstract class SocketUsingTask 
    
      implements CancellableTask
     
       {    @GuardedBy("this") private Socket socket;    protected synchronized void setSocket(Socket s) {        socket = s;    }    public synchronized void cancel() {        try {            if (socket != null)                socket.close();        } catch (IOException ignored) {        }    }    public RunnableFuture
      
        newTask() {        return new FutureTask
       
        (this) {            public boolean cancel(boolean mayInterruptIfRunning) {                try {                    SocketUsingTask.this.cancel();                } finally {                    return super.cancel(mayInterruptIfRunning);                }            }        };    }}interface CancellableTask 
        
          extends Callable
         
           { void cancel(); RunnableFuture
          
            newTask();}@ThreadSafeclass CancellingExecutor extends ThreadPoolExecutor { public CancellingExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue
           
             workQueue) { super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue); } public CancellingExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue
            
              workQueue, ThreadFactory threadFactory) { super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory); } public CancellingExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue
             
               workQueue, RejectedExecutionHandler handler) { super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, handler); } public CancellingExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue
              
                workQueue, ThreadFactory threadFactory, RejectedExecutionHandler handler) { super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory, handler); } protected 
               
                 RunnableFuture
                
                  newTaskFor(Callable
                 
                   callable) { if (callable instanceof CancellableTask) return ((CancellableTask
                  
                   ) callable).newTask(); else return super.newTaskFor(callable); }}复制代码

程序12的CancellableTask中定义了一个CancellableTask接口，该接口扩展了Callable，并增加了一个cancel方法和一个newTask工厂方法来构造RunnableFuture。CancellingExecutor扩展了ThreadPoolExecutor，并通过改写newTaskFor使得CancellableTask可以创建自己的Future。

SocketUsingTask实现了CancellableTask，并定义了Future.cancel来关闭套接字和调用super.cancel。如果SocketUsingTask通过其自己的Future来取消，那么底层的套接字将被关闭并且线程将被中断。因此它提高了任务队取消操作的响应性：不仅能够在调用可中断方法的同时确保响应取消操作，而且还能调用可阻调的套接字I/O方法。

停止基于线程的服务

应用程序通常会创建拥有多个线程的服务，例如线程池，并且这些服务的生命周期通常比创建它们的方法的生命周期更长。如果应用程序准备退出，那么这些服务所拥有的线程也需要结束。由于无法通过抢占式的方法停止线程，因此他们需要自行结束。

正确的封装原则是：除非拥有某个线程，否则不能对该线程进行操控。

与其他封装对象一样，线程的所有权是不可传递的：应用程序可以拥有服务，服务也可以拥有工作者线程，但应用程序并不能拥有工作者线程，因此应用程序不能直接停止工作者线程。

对于持有线程的服务，只要服务的存在时间大于创建者线程的方法的存在时间，那么就应该提供生命周期方法

示例：日志服务

程序13：不支持关闭的生产者-消费者日志服务

public class LogWriter {    private final BlockingQueue
    
      queue;    private final LoggerThread logger;    private static final int CAPACITY = 1000;    public LogWriter(Writer writer) {        this.queue = new LinkedBlockingQueue
     
      (CAPACITY);        this.logger = new LoggerThread(writer);    }    public void start() {        logger.start();    }    public void log(String msg) throws InterruptedException {        queue.put(msg);    }    private class LoggerThread extends Thread {        private final PrintWriter writer;        public LoggerThread(Writer writer) {            this.writer = new PrintWriter(writer, true); // autoflush        }        public void run() {            try {                while (true)                    writer.println(queue.take());            } catch (InterruptedException ignored) {            } finally {                writer.close();            }        }    }}复制代码

程序13中的LogWriter给出了一个简单的日志服务示例，其中日志操作在单独的日志线程中执行。产生的日志消息的线程并不会将消息直接写入输出流，而是由LogWriter通过BlockingQueue将消息提交给日志线程，并由日志线程写入。这是一种多生产者单消费者的设计方式：每个调用log的操作都相当于一个生产者，而后台的日志线程则相当于消费者。如果消费者的处理速度低于生产者的生成速度，那么BlockingQueue将阻塞生产者，直到日志线程有能力处理新的日志消息。

为了使像LogWriter这样的服务在软件产品中能发挥实际的作用，还需要实现一种终止日志线程的方法，从而避免使JVM无法正常关闭。要停止日志线程是很容易的，因为他会反复调用take，而take能响应中断。如果将日志线程修改为当捕获到InterruptedException时退出，那么只需在日志线程就能停止服务。

然而，如果只是使日志线程退出，那么还不是一种完备的关闭机制。这种直接关闭的做法会丢失那些正在等待被写入到日志的信息。不仅如此，其他线程将在调用log时被阻塞，因为日志消息队列是满的，因此这些线程将无法解除阻塞状态。当取消一个生产者-消费者操作时，需要同时取消生产者和消费者。在中断日志线程时会处理消费者，但这个示例中，由于生产者并不是专门的线程，因此要取消他们将非常困难。

另一种关闭LogWriter的方法是：设置某个“已请求关闭”的标志，以避免进一步提交日志消息，如程序14所示。在收到关闭请求后，消费者把队列中的所有消息写入日志，并解除所有在调用log时阻塞的生产者。然而，这个方法存在着竞态条件问题，使得该方法并不可靠。log的实现是一种“先判断再运行”的代码序列：生产者发现该服务还没有关闭，因此在关闭服务后仍然会将日志消息放入队列，这同样会是的生产者可能在调用log时阻塞并且无法接触阻塞状态。

程序14：通过一种不可靠的方式为日志增加关闭支持

public void log(String msg) throws InterruptedException {    if (!shutdownRequested)        queue.put(msg);    else        throw new IllegalStateException("logger is shut down");}复制代码

为LogWriter提供可靠关闭操作的方法是解决竞态条件问题，因而要使日志消息的提交造作变成原子操作。然而，我们不希望在消息加入队列时去持有一个锁，因为put方法本身就可以阻塞。采用的方法是：通过原子操作来检查关闭请求，并且有条件的递增一个计数器来“保持”提交消息的权利，如程序15的LogService所示。

程序15：向LogWriter添加可靠的取消操作

public class LogService {    private final BlockingQueue
    
      queue;    private final LoggerThread loggerThread;    private final PrintWriter writer;    @GuardedBy("this") private boolean isShutdown;    @GuardedBy("this") private int reservations;    public LogService(Writer writer) {        this.queue = new LinkedBlockingQueue
     
      ();        this.loggerThread = new LoggerThread();        this.writer = new PrintWriter(writer);    }    public void start() {        loggerThread.start();    }    public void stop() {        synchronized (this) {            isShutdown = true;        }        loggerThread.interrupt();    }    public void log(String msg) throws InterruptedException {        synchronized (this) {            if (isShutdown)                throw new IllegalStateException(/*...*/);            ++reservations;        }        queue.put(msg);    }    private class LoggerThread extends Thread {        public void run() {            try {                while (true) {                    try {                        synchronized (LogService.this) {                            if (isShutdown && reservations == 0)                                break;                        }                        String msg = queue.take();                        synchronized (LogService.this) {                            --reservations;                        }                        writer.println(msg);                    } catch (InterruptedException e) { /* retry */                    }                }            } finally {                writer.close();            }        }    }}复制代码

关闭ExecutorService

在上节中，ExecutorService提供了两种关闭方法：shutdown正常关闭，shutdownNow强行关闭。在进行强行关闭时，shutdownNow首先关闭当前正在执行的任务，然后返回所有尚未启动的任务清单。

这两种关闭方式的差别在于各自的安全性和响应性：强行关闭的速度更快，但风险更大，因为任务很可能在执行到一半时被结束；而正常关闭虽然速度慢，但却更安全，因为ExecutorService会一直等到队列中的所有任务都执行完成后才关闭。

程序16：使用ExecutorService的日志服务

public class LogService {    private final ExecutorService exec = new SingleThreadExecutor();    ...    public void start() {}    public void stop() {        try{            exec.shutdown();            exec.awaitTermination(TIMEOUT, UNIT);        }finally{            writer.close()；        }    }    public void log(String msg){        try{            exec.execute(new WriterTask(msg));        }catch(RejectedExecutionException ignored){}    }}复制代码

“毒丸”对象

另一种关闭生产者-消费者服务的方式就是使用“毒丸”对象，“毒丸”是指一个放在队列上的对象，其含义是：“当得到这个对象时，立即停止”。在FIFO队列中，“毒丸”对象将确保消费者在关闭之前首先完成队列中的所有工作，在提交“毒丸”对象之前提交的工作都会被处理，而生产者提交了“毒丸”对象之后，将不会再提交任何工作。

程序17：通过“毒丸”对象来关闭服务

public class IndexingService {    private static final int CAPACITY = 1000;    private static final File POISON = new File("");    private final IndexerThread consumer = new IndexerThread();    private final CrawlerThread producer = new CrawlerThread();    private final BlockingQueue
    
      queue;    private final FileFilter fileFilter;    private final File root;    public IndexingService(File root, final FileFilter fileFilter) {        this.root = root;        this.queue = new LinkedBlockingQueue
     
      (CAPACITY);        this.fileFilter = new FileFilter() {            public boolean accept(File f) {                return f.isDirectory() || fileFilter.accept(f);            }        };    }    private boolean alreadyIndexed(File f) {        return false;    }    class CrawlerThread extends Thread {        public void run() {            try {                crawl(root);            } catch (InterruptedException e) { /* fall through */            } finally {                while (true) {                    try {                        queue.put(POISON);                        break;                    } catch (InterruptedException e1) { /* retry */                    }                }            }        }        private void crawl(File root) throws InterruptedException {            File[] entries = root.listFiles(fileFilter);            if (entries != null) {                for (File entry : entries) {                    if (entry.isDirectory())                        crawl(entry);                    else if (!alreadyIndexed(entry))                        queue.put(entry);                }            }        }    }    class IndexerThread extends Thread {        public void run() {            try {                while (true) {                    File file = queue.take();                    if (file == POISON)                        break;                    else                        indexFile(file);                }            } catch (InterruptedException consumed) {            }        }        public void indexFile(File file) {            /*...*/        };    }    public void start() {        producer.start();        consumer.start();    }    public void stop() {        producer.interrupt();    }    public void awaitTermination() throws InterruptedException {        consumer.join();    }}复制代码

只有生产者和消费者数量都已知的情况下，才可以使用“毒丸”对象。

在IndexingService中采用的解决方案可以扩展到多个生产者：只需每个生产者都向队列中放一个“毒丸”对象，并且消费者仅当接收到第N个生产者的“毒丸”对象时才停止。这种方法也可以扩展到多个消费者的情况，只需生产者将第N个消费者“毒丸”对象放入队列。

然而，当生产者和消费者数量较大时，这种方法变的难以使用，只有在无界队列中，“毒丸”对象才能可靠的工作。

示例：只执行一次的服务

如果某个方法需要处理一批任务，并且当所有任务都处理完成后才返回，那么可以通过一个私有的Executor来简化服务的生命周期管理，其中该Executor的生命周期是由这个方法来控制的。

程序20中的checkMail方法能在多台主机上并行的检查新邮件。他创建一个私有的Executor，并向每台主机提交一个任务。然后，当所有的邮件检查任务都执行完成后，关闭Executor并等待结束。

程序20：使用私有的Executor，并且该Executor的生命周期受限于方法调用

public class CheckForMail {    public boolean checkMail(Set
    
      hosts, long timeout, TimeUnit unit)            throws InterruptedException {        ExecutorService exec = Executors.newCachedThreadPool();        //这里不能使用 volatile hasNewMail，因为还需要在匿名内中修改        final AtomicBoolean hasNewMail = new AtomicBoolean(false);        try {            //循环检索每台主机            for (final String host : hosts)                exec.execute(new Runnable() {                    public void run() {                        if (checkMail(host))                            hasNewMail.set(true);                    }                });        } finally {            //因为ExecutorService只在这个方法中服务，所以完成后即可关闭            exec.shutdown();            //等待任务的完成，如果超时还未完成也会返回            exec.awaitTermination(timeout, unit);        }        return hasNewMail.get();    }    private boolean checkMail(String host) {        // Check for mail        return false;    }}复制代码

shutdownNow的局限性

当通过shutdownNow来强行关闭ExecutorService时，它会尝试取消正在执行的任务，并返回所有已提交但尚未开始的任务，从而将这些任务写入日志或者保存起来以便之后进行处理。

然而，我们无法通过常规方法来找出哪些任务已经开始但尚未结束。这意味着我们无法在关闭过程中知道正在执行的任务的状态，除非任务本身会执行某种检查。要知道那些任务还没完成，你不仅需要知道那些任务还没有开始，还需要知道当Executor关闭时那些任务正在执行。

程序21的TrackingExecutor中给出了如果在关闭过程中判断正在执行的任务。通过封装ExecutorService并使得execute记录哪些任务是在关闭后取消的，TrackingExecutor可以找出哪些任务已经开始但还没有正常完成。在Executor结束后，getCancelledTasks返回被取消的任务清单。要使这项技术能发挥作用，任务在返回时必须维持线程的中断状态，在所有设计良好的任务中，都会实现这个功能。

任务21：在ExecutorService中跟踪在关闭之后被取消的任务

public class TrackingExecutor extends AbstractExecutorService {    private final ExecutorService exec;    private final Set
    
      tasksCancelledAtShutdown =            Collections.synchronizedSet(new HashSet
     
      ());    public TrackingExecutor(ExecutorService exec) {        this.exec = exec;    }    public void shutdown() {        exec.shutdown();    }    public List
      
        shutdownNow() {        return exec.shutdownNow();    }    public boolean isShutdown() {        return exec.isShutdown();    }    public boolean isTerminated() {        return exec.isTerminated();    }    public boolean awaitTermination(long timeout, TimeUnit unit)            throws InterruptedException {        return exec.awaitTermination(timeout, unit);    }    //返回被取消的任务    public List
       
         getCancelledTasks() {        //如果shutdownNow未调用或调用未完成时        if (!exec.isTerminated())            throw new IllegalStateException(/*...*/);        return new ArrayList
        
         (tasksCancelledAtShutdown);    }    public void execute(final Runnable runnable) {        exec.execute(new Runnable() {            public void run() {                try {                    runnable.run();                } finally {                    //如果调用了shutdownNow且运行的任务被中断                    if (isShutdown()                            && Thread.currentThread().isInterrupted())                        //记录被取消的任务                        tasksCancelledAtShutdown.add(runnable);                }            }        });    }}复制代码

在程序22的WebCrawler中给出了TrackingExecutor的用法。网页爬虫程序的工作通常是无穷尽的，因此当爬虫程序必须关闭时，我们通常希望保存它的状态，以便稍后重新启动。CrawlTask提供了一个getPage方法，该方法能找出正在处理的页面。当爬虫程序关闭时，无论是还没有开始的任务，还是那些被取消的任务，都将记录他们的URL，因此，当爬虫程序重新启动时，就可以将这些URL的页面抓取任务加入到任务队列中。

程序22：使用TrackingExecutorService来保存未完成的任务以备后续执行

public abstract class WebCrawler {    private volatile TrackingExecutor exec;    @GuardedBy("this") private final Set
    
      urlsToCrawl = new HashSet
     
      ();    private final ConcurrentMap
      
        seen = new ConcurrentHashMap
       
        ();    private static final long TIMEOUT = 500;    private static final TimeUnit UNIT = MILLISECONDS;    public WebCrawler(URL startUrl) {        urlsToCrawl.add(startUrl);    }    public synchronized void start() {        exec = new TrackingExecutor(Executors.newCachedThreadPool());        for (URL url : urlsToCrawl) submitCrawlTask(url);        urlsToCrawl.clear();    }    public synchronized void stop() throws InterruptedException {        try {            saveUncrawled(exec.shutdownNow());            if (exec.awaitTermination(TIMEOUT, UNIT))                saveUncrawled(exec.getCancelledTasks());        } finally {            exec = null;        }    }    protected abstract List
        
          processPage(URL url);    private void saveUncrawled(List
         
           uncrawled) { for (Runnable task : uncrawled) urlsToCrawl.add(((CrawlTask) task).getPage()); } private void submitCrawlTask(URL u) { exec.execute(new CrawlTask(u)); } private class CrawlTask implements Runnable { private final URL url; CrawlTask(URL url) { this.url = url; } private int count = 1; boolean alreadyCrawled() { return seen.putIfAbsent(url, true) != null; } void markUncrawled() { seen.remove(url); System.out.printf("marking %s uncrawled%n", url); } public void run() { for (URL link : processPage(url)) { if (Thread.currentThread().isInterrupted()) return; submitCrawlTask(link); } } public URL getPage() { return url; } }}复制代码

在TrackingExecutor中存在一个不可避免的竞态条件，从而产生“误报”问题：一些被任务已取消的任务实际上已经执行完成。这个问题的原因在于，在任务执行最后一条指令以及线程池将任务记录为“结束”的两个时刻之间，线程池可能被关闭。如果是幂等（即将任务执行两次与执行一次会得到相同的结果），那么这不会存在问题，在网页爬虫程序中就是这种情况。否则，在应用程序中必须考虑这种风险，并对“误报”问题做好准备。

处理非正常的线程终止

当单线程的控制程序由于发生了一个未捕获的异常而终止时，程序将停止运行，并产生于程序正常输出非常不同的栈追踪信息，这种情况很容易理解。然而，如果并非程序中的某个线程发生故障，那么通常并不会如此明显。在控制台可能会输出栈追踪信息，但没人会观察控制台。此外，当线程发生故障时，应用程序可能看起来仍然在工作，所以这个是失败可能被忽略。

导致线程提前死亡的最主要原因就是RuntimeException。由于这些异常表示出现了某种编程错误或者其他不可修复的错误，因此他们通常不会被捕获。他们不会再调用栈中逐层传递，而是默认的在控制台中输入栈追踪信息，并终止线程。

线程非正常退出的后果可能是良性的，也可能是恶性的，这要取决于线程在应用程序中的作用。虽然在线程池中丢失一个线程可能会性能带来一定影响，但如果程序能在包含50个线程的线程池上运行良好，那么在包含49个线程的线程池上通常也能运行良好。

程序23中给出了如果在线程池内部构建一个工作者线程。如果任务抛出了一个未检查异常，那么它将使线程终结，但会首先通知框架该线程已经终结。然后，框架可能会用新的线程来代替这个工作线程，也可能不会，因为线程池正在关闭，或者当前已经有足够多的线程满足需要。

程序23：典型的线程池工作者线程结构

public void run(){    Throwable thrown = null;    try{        while(!isInterrupted())            runTask(getTaskFromWorkQueue());    }catch(Throwable e){        thrown = e;    }finally{        threadExited(this, thrown);    }}复制代码

未捕获异常的处理

上一节介绍了主动方法来解决未检查异常。Thread API中提供了UncaughtExceptionHandler，它能检测出某个线程由于未捕获的异常而终结的情况。这两种方法是互补的，通过结合，可以有效的防止线程泄露问题。

当一个线程由于未捕获异常而退出时，JVM会把这个事件报告给应用程序提供的UncaughtExceptionHandler异常处理器。如果没有提供任何异常处理器，那么默认的行为将是栈追踪信息输入到System.err

程序24：UncaughtExceptionHandler接口

public interface UncaughtExceptionHandler {    void uncaughtException(Thread t, Throwable e);}复制代码

异常处理器如何处理未捕获异常，取决于对服务质量的需求。最常见的响应方式是将一个错误信息以及对应的栈追钟信息写入应用程序日志中。异常处理器还可以采取更直接的响应，例如重新启动线程，关闭应用程序，或者执行其他修复或诊断操作。

程序25：将异常写入日志的UncaughtExceptionHandler

public class UEHLogger implements Thread.UncaughtExceptionHandler {    public void uncaughtException(Thread t, Throwable e) {        Logger logger = Logger.getAnonymousLogger();        logger.log(Level.SEVERE, "Thread terminated with exception: " + t.getName(), e);    }}复制代码

在运行时间较长的应用程序中，通过会为所有线程的未捕获异常制定同一个
异常处理器，并且该处理器至少会将异常记录到日志中。

JVM关闭

JVM既可以正常关闭，也可以强行关闭。正常关闭的触发方法有多种，包括：当最终一个正常线程结束时，或者调用了System.exit时，或者通过其他平台的方法关闭时。虽然可以通过标准的方法来正常关闭JVM，但也可以通过调用Runtime.halt或者在操作系统杀死JVM进程来强行关闭JVM。

关闭钩子

在正常关闭中，JVM首先调用所有已注册的关闭钩子。关闭钩子指通过Runtime.addShutdownHook注册的但尚未开始的线程。JVM不能保证关闭钩子的调用顺序。在关闭应用程序线程时，如果有线程仍然在运行，那么这些线程接下来将与关闭进程并发执行。当所有的关闭钩子都执行结束时，如果runFinalizersOnExit为true，那么JVM将运行终结器，然后停止。JVM并不会停止或者中断任何在关闭时仍然运行的应用程序线程。当JVM最终结束时，这些线程将被强行结束。如果关闭钩子或者终结器没有执行完成，那么正常关闭进程挂起，并且JVM必须强制关闭。当被强制关闭时，只是关闭了JVM，而不会运行关闭钩子。

关闭钩子应该是线程安全的：他们在访问共享数据时必须使用同步机制，并且小心的避免发生思索，这与其他并发代码的要求相同。并且，关闭钩子不应该对应用程序的状态或者JVM的关闭原因做出任何加持，因此在编写关闭钩子的代码时，必须考虑周全。最后，关闭钩子必须尽快退出，因为他们会延迟JVM的结束时间，而用户可能希望JVM尽快终止。

程序26：通过注册一个关闭钩子来停止服务

public void start(){    Runtime.getRuntime().addShutdownHook(new Thread(){        public void run(){            try{LogServie.this.stop();}            catch(InterruptedException ignored){}        }    });}复制代码

程序26给出了如何使程序16中的LogService在其start方法中注册一个关闭钩子，从而确保在退出时关闭日志文件。

守护线程

线程分两种：普通线程和守护线程。在JVM启动时创建的所有线程中，除了主线程以外，其他的线程都是守护线程。当创建一个新线程时，新线程将继承创建它的线程的守护状态，因为，默认情况下，主线程创建的所有线程，都是普通线程。

普通线程与守护线程之前的差异仅在于当线程退出时发生的操作：当一个线程退出时，JVM会检查其他正在运行的线程，如果这些线程都是守护线程，那么JVM会正常退出，当JVM停止时，所有仍然存在的守护线程都会将被抛弃——既不会执行finally代码，也不会执行回卷栈，而JVM只是直接退出。

守护线程通常不能用来替代应用程序管理各个服务的生命周期。

终结器

当不再需要内存资源时，可以通过垃圾回收器来回收他们，但对于其他一些资源，如文件句柄或者套接字句柄，当不在需要他们时，需要显示的交还给操作系统。为了实现这个功能，垃圾回收器对那些定义了finalize方法的对象会进行特殊处理：在回收器释放他们后，调用他们的finalize方法，从而保证一些持久化的资源被释放掉。

由于终结器可以在某个JVM管理的线程中运行，因此终结器访问的任何状态都可能被多线程访问，这样就必须对其访问操作进行同步。终结器不能保证他们将在何时运行，甚至是否会运行，并且复杂的终结器通常还会在对象上产生巨大的性能开销。

避免使用终结器

小结

在任务、线程、服务以及应用程序等模块的生命周期结束问题，可能会增加他们的设计和实现时的复杂性。Java并没有提供某种抢占机制来取消操作或者终结线程。相反，它提供了一种协作式的中断机制来实现取消操作，但这要依赖于如何构建取消操作的协议，以及能否始终遵守这个协议。通过FutureTask和Executor框架，可以帮助我们构建可取消的任务和服务。