7.取消与关闭

取消与关闭

有时候希望提前结束任务或线程，例如用户取消操作，或者应用程序需要被快速关闭。而java对此没有提供任何机制来安全地终止线程。但它提供了中断，这是一种协作机制，能够使一个线程终止另一个线程的当前工作。
这种协作式的方法是必要的，很少希望某个任务、线程或服务立即停止，因为这种立即停止会使共享的数据结构处于不一致的状态。相反，可以使用一种协作的方式：当需要停止时，首先清除当前正在执行的工作，然后再结束。这提供了更好的灵活性，因为任务本身的代码比发出取消请求的代码更清楚如何执行清除工作。

7.1任务取消

取消某个操作的原因很多：

• 用户请求取消。例如点击图形界面程序中的取消按钮。
• 有时间限制的操作。
• 应用程序事件。例如，启动多个任务查找解决方案，当其中一个找到时，所有其他在搜索的任务都将被取消。
• 应用程序错误。
• 程序或服务关闭。

在java中没有一种安全的抢占式方法来停止线程，只有一些协作式的机制，使请求取消的任务和代码都遵循一种协商好的协议。
其中一种协作机制能设置某个请求取消的标志，而任务将定期地查看该标志。如果设置了这个标志，那么任务将提前结束：

@ThreadSafe
public class PrimeGenerator implements Runnable {
	private static ExecutorService exec = Executors.newCachedThreadPool();

    @GuardedBy("this")
    private final List<BigInteger> primes = new ArrayList<BigInteger>();
    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<BigInteger> get() {
        return new ArrayList<BigInteger>(primes);
    }

	/**
	 * 素数生成器运行1秒后取消，通常并不会刚好在运行1秒后停止，因此在请求取消的时刻和run方法中循环执行
	 * 下一次检查之间可能存在延迟。
	 */
	static List<BigInteger> aSecondOfPrimes() throws InterruptedException {
        PrimeGenerator generator = new PrimeGenerator();
        exec.execute(generator);
        try {
            TimeUnit.SECONDS.sleep(1);
        } finally {
        	// cancel方法由finally块调用，从而确保即使在调用sleep时被中断也能取消。如果cancel
        	// 没有被调用，那么搜索素数的线程将永远运行下去，不断消耗CPU，并使得JVM不能正常退出。
            generator.cancel();
        }
        return generator.get();
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43

一个可取消的任务必须拥有取消策略，即其他代码如何请求取消该任务，任务在何时检查是否已经请求了取消，以及在响应取消请求时应该执行哪些操作。

7.1.1中断

/**
 * 生产者线程生成素数，并将它们放入一个阻塞队列。
 * 
 * 如果生产者的速度超过了消费者的处理速度，队列将被填满，put方法也会阻塞。而如果此时消费者希望取消
 * 生产者任务，虽然可以调用cancel方法来设置cancelled标志，但此时生产者却永远不能检查这个标志，因为
 * 它无法从阻塞的put方法中恢复过来(消费者此时已经停止从队列中取出素数，所以put方法将一直保持阻塞状态)。
 */
class BrokenPrimeProducer extends Thread {
    private final BlockingQueue<BigInteger> queue;
    private volatile boolean cancelled = false;

    BrokenPrimeProducer(BlockingQueue<BigInteger> 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;
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30

线程中断是一种协作机制，线程可以通过这种机制来通知另一个线程，告诉它在合适的或者可能的情况下停止当前工作，并转而执行其他的工作。
每个线程都有一个boolean类型的中断状态。当中断线程时，这个线程的中断状态将被设置为true。在Thread中包含了中断线程以及查询线程中断状态的方法。

public class Thread {
	// 中断目标线程
	public void interrupt() { ... }
	// 返回目标线程的中断状态
	public boolean isInterrupted() { ... }
	// 清除当前线程的中断状态，并返回它之前的值，这也是清除中断状态的唯一方法
	public static boolean interrupted() { ... }
}
1
2
3
4
5
6
7
8

当线程在非阻塞状态下中断时，它的中断状态将被设置，然后根据将被取消的操作来检查中断状态以判断发生了中断。如果不触发InterruptedException，那么中断状态将一直保持，直到明确地清除中断状态。
对中断操作的正确理解是：它并不会真正地中断一个正在运行的线程，而只是发出中断请求，然后由线程在下一个合适的时刻中断自己。有些方法，例如wait、sleep和join等，将严格地处理这种请求，当它们收到中断请求或者在开始执行时发现某个已经被设置好的中断状态时，将清除中断状态并抛出一个异常。设计良好的方法可以对中断请求进行某种处理。设计糟糕的方法可能会屏蔽中断请求，从而导致调用栈中的其他代码无法对中断请求作出响应。
在使用静态的interrupted时应该小心，因为它会清除当前线程的中断状态。如果在调用interrupted时返回了true，那么除非想屏蔽这个中断，否则必须对它进行处理：可以抛出InterruptedException，或者通过再次调用interrupt来恢复中断状态。

如果任务代码能够响应中断，那么可以使用中断作为取消机制，并且利用许多库类中提供的中断支持。通常，中断是实现取消的最合理方式。

public class PrimeProducer extends Thread {
    private final BlockingQueue<BigInteger> queue;

    PrimeProducer(BlockingQueue<BigInteger> 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();
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

7.1.2中断策略

线程应该包含中断策略，其规定线程在发现中断请求时，应该做哪些工作(如果需要的话)，哪些工作单元对于中断来说是原子操作，以及以多快的速度来响应中断。
最合理的中断策略是某种形式的线程级取消操作或服务级取消操作：尽快退出，在必要时进行清理，通知某个所有者该线程已经退出。此外还可以建立其他的中断策略，例如暂停服务或重新开始服务。

区分任务和线程对中断的反应是很重要的。一个中断请求可以有一个或多个接收者：中断线程池中的某个工作者线程，同时意味着取消当前任务和关闭工作者线程。
任务不会在其自己拥有的线程中执行，而是在某个服务(例如线程池)拥有的线程中执行。对于非线程所有者的代码来说(例如，对于线程池而言，任何在线程池实现以外的代码)，应该小心地保存中断状态，这样拥有线程的代码才能对中断做出响应，即使非所有者代码也可以做出响应。
这就是为什么大多数可阻塞的库函数都只是抛出InterruptedException作为中断响应。它们永远不会在某个由自己拥有的线程中运行，因此它们为任务或库代码实现了最合理的取消策略：尽快退出执行流程，并把中断信息传递给调用者，从而使调用栈中的上层代码可以采取进一步的操作。

当检查到中断请求时，任务并不需要放弃所有的操作，它可以推迟处理中断请求，并直到某个更合适的时刻。因此需要记住中断请求，并在完成当前任务后抛出InterruptedException或者表示已收到中断请求。这项技术能够确保在更新过程中发生中断时，数据结构不会被破坏。
任务不应该对执行该任务的线程的中断策略做出任何假设，除非该任务被专门设计为在服务中运行，并且在这些服务中包含特定的中断策略。无论任务把中断视为取消，还是其他某个中断响应操作，都应该小心地保存执行线程的中断状态。如果除了将InterruptedException传递给调用者外还需要执行其他操作，那么应该在捕获InterruptedException之后恢复中断状态：

Thread.currentThread().interrupt();
1

正如任务代码不应该对其执行所在的线程的中断策略做出假设，执行取消操作的代码也不应该对线程的中断策略做出假设。线程应该只能由其所有者中断，所有者可以将线程的中断策略信息封装到某个合适的取消机制中，例如关闭(shutdown)方法。
由于每个线程拥有各自的中断策略，因此除非知道中断对该线程的含义，否则就不应该中断这个线程。

7.1.3响应中断

当调用可中断的阻塞函数时，例如Thread.sleep或BlockingQueue.put等，有两种实用策略可用于处理InterruptedException：

• 传递异常(可能在执行某个特定于任务的清除操作之后)，从而使自己的方法也成为可中断的阻塞方法。
• 恢复中断状态，从而使调用栈中的上层代码能够对其进行处理。

如果不想或无法传递InterruptedException(或许通过Runnable来定义任务)，可以通过再次调用interrupt来恢复中断状态。不能屏蔽InterruptedException，例如在catch块中捕获到异常却不做任何处理，除非自己的代码中实现了线程的中断策略。由于大多数代码并不知道它们将在哪个线程中运行，因此应该保存中断状态。

对于一些不支持取消但仍可以调用可中断阻塞方法的操作，它们必须在循环中调用这些方法，并在发现中断后重新尝试。在这种情况下，它们应该在本地保存中断状态，并在返回前恢复状态而不是在捕获InterruptedException时恢复状态：

/**
 * 如果过早地设置中断状态，就可能引起无限循环，因为大多数可中断的阻塞方法都会在入口处检查中断状态，
 * 并且当发现该状态已被设置时会立即抛出InterruptedException(通常，可中断的方法会在阻塞或进行重要
 * 的工作前首先检查中断，从而尽快地响应中断)。
 */
public Task getNextTask(BlockingQueue<Task> queue) {
	boolean interrupted = false;

	try {
		while (true) {
			try {
				return queue.take();
			} catch (InterruptedException e) {
				interrupted = true;
				// 重新尝试
			}
		}
	} finally {
		if (interrupted) {
			Thread.currentThread().interrupt();
		}
	}
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

如果代码不会调用可中断的阻塞方法，那么仍然可以通过在任务代码中轮询当前线程的中断状态来响应中断。要选择合适的轮询频率，就需要在效率和响应性之间进行权衡。如果响应性要求较高，那么不应该调用那些执行时间较长并且不响应中断的方法，从而对可调用的库代码进行一些限制。
在取消过程中可能涉及除了中断状态之外的其他状态。中断可以用来获得线程的注意，并且由中断线程保存的信息，可以为中断的线程提供进一步的指示(当访问这些信息时，要确保使用同步)。
例如，当一个由ThreadPoolExecutor拥有的工作者线程检测到中断时，它会检查线程池是否正在关闭。如果是，它会在结束之前执行一些线程池清理工作，否则它可能创建一个新线程将线程池恢复到合理的规模。

7.1.4示例：计时运行

/**
 * 在指定时间内运行一个任意的Runnable的示例。它在调用线程中运行任务，并安排了一个取消任务，
 * 在运行指定的时间间隔后中断它。这解决了从任务中抛出未检查异常的问题，因为该异常会被timedRun
 * 的调用者捕获。
 */
private static final ScheduledExecutorService cancelExec = ...;

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);
	// 任务抛出的未检查异常会被timedRun的调用者捕获
	r.run();
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

这是一种非常简单的方法，但却破坏了以下规则：在中断线程之前，应该了解它的中断策略。由于timedRun可以从任意一个线程中调用，因此它无法知道这个调用线程的中断策略。如果任务在超时之前完成，那么中断timedRun所在线程的取消任务将在方法返回到调用者之后启动，在这种情况下不知道将运行什么代码，但结果一定是不好的。
而且，如果任务不响应中断，那么timedRun会在任务结束时才返回，此时可能已经超过了指定的时限(或者还没有超过时限)。如果某个限时运行的服务没有在指定的时间内返回，那么将对调用者带来负面影响。

/**
 * 执行任务的线程拥有自己的执行策略，即使任务不响应中断，限时运行的方法仍能返回到它的调用者。
 * 在启动任务线程之后，timedRun将执行一个限时的join方法。在join返回后，它将检查任务中是否
 * 有异常抛出，如果有的话，则会在调用timedRun的线程中再次抛出该异常。由于Throwable将在两个
 * 线程之间共享，因此该变量被声明为volatile类型，从而确保安全地将其从任务线程发布到timedRun
 * 线程。
 */
public static void timedRun(final Runnable r, long timeout, TimeUnit unit) throws InterruptedException {
	class RethrowableTask implements Runnable {
		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();
	cancelExec.schedule(new Runnable() {
		public void run() {
			// 并不会打断调用者的线程
			taskThread.interrupt();
		}
	}, timeout, unit);

	taskThread.join(unit.toMillis(timeout));
	task.rethrow();
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39

存在着另外一个问题：由于依赖于一个限时的join，因此无法知道执行控制时因为线程正常退出而返回还是因为join超时而返回。

7.1.5通过Future来实现取消

ExecutorService.submit将返回一个Future来描述任务。Future拥有一个cancel方法，该方法带有一个boolean类型的参数mayInterruptIfRunning，表示取消操作是否成功(这只是表示任务是否能够接收中断，而不是表示任务是否能检测并处理中断)。如果mayInterruptIfRunning为true并且任务当前正在某个线程中执行，那么这个线程能被中断。如果这个参数为false，那么意味着若任务还没有启动，就不要运行它，这种方式应该用于那些不处理中断的任务中。
那么在什么情况下调用cancel可以将参数指定为true：执行任务的线程是由标准的Executor创建的，它实现了一种中断策略使得任务可以通过中断被取消，所以如果任务在标准Executor中运行，并通过它们的Future来取消任务，那么可以设置mayInterruptIfRunning。当尝试取消某个任务时，不宜直接中断线程池，因为并不知道当中断请求到达时正在运行什么任务，只能通过任务的Future来实现取消。

/**
 * 将任务提交给一个ExecutorService，并通过一个定时的Future.get来获得结果。如果get在返回时抛出了一个
 * TimeoutException，那么任务将通过它的Future来取消(为了简化代码，这个版本的timedRun在finally块中
 * 将直接调用Future.cancel，因为取消一个已完成的任务不会带来任何影响)。如果任务在被取消前就抛出一个
 * 异常，那么该异常将被重新抛出以便由调用者来处理。
 */
public static void timedRun(Runnable r, long timeout, TimeUnit unit) throws InterruptedException {
	Future<?> task = taskExec.submit(r);
	try {
		task.get(timeout, unit);
	} catch (TimeoutException e) {
		// 接下来任务将被取消
	} catch (ExecutionException e) {
		// 如果在任务中抛出了异常，那么重新抛出该异常
		throw launderThrowable(e.getCause());
	} finally {
		// 如果任务已经结束，那么执行取消操作也不会带来任何影响
		task.cancel(true);	// 如果任务正在运行，那么将被中断
	}
 }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

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

7.1.6处理不可中断的阻塞

在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并关闭链路(这还会使得其他在这条链路上阻塞的线程同样抛出该异常)。当关闭一个InterruptibleChannel时，将导致所有在链路操作上阻塞的线程都抛出AsynchronousCloseException。大多数标准的Channel都实现了InterruptibleChannel。
• Selector的异步I/O。如果一个线程在调用Selector.select方法时阻塞了，那么调用close或wakeup方法会使线程抛出ClosedSelectorException并提前返回。
• 获取某个锁。如果一个线程由于等待某个内置锁而阻塞，那么将无法响应中断，因为线程认为它肯定会获得锁，所以将不会理会中断请求。但是，在Lock类中提供了lockInterruptibly方法，该方法允许在等待一个锁的同时仍能响应中断。

/**
 * 封装非标准的取消操作。ReaderThread管理了一个套接字连接，并从中读取数据，同时将接收到的数据
 * 传递给processBuffer。为了结束某个用户的连接或者关闭服务器，ReaderThread改写了interrupt方法，
 * 使其既能处理标准的中断，也能关闭底层的套接字。因此，无论ReaderThread线程是在read方法中阻塞
 * 还是在某个可中断的阻塞方法中阻塞，都可以被中断并停止执行当前的工作。
 */
public class ReaderThread extends Thread {
    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.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) { /* 允许线程退出 */
        }
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39

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

当把一个Callable提交给ExecutorService时，submit方法会返回一个Future，可以通过这个Future来取消任务。newTaskFor是一个工厂方法，它将创建Future来代表任务。newTaskFor还能返回一个RunnableFuture接口，该接口扩展了Future和Runnable(并由FutureTask实现)。
通过定制表示任务的Future可以改变Future.cancel的行为。例如，定制的取消代码可以实现日志记录或者收集取消操作的统计信息，以及取消一些不响应中断的操作。

/**
 * 扩展了Callable，并增加了一个cancel方法和一个newTask工厂方法来构造RunnableFuture。
 */
interface CancellableTask <T> extends Callable<T> {
    void cancel();

    RunnableFuture<T> newTask();
}

/**
 * 扩展了ThreadPoolExecutor，并通过改写newTaskFor使得CancellingExecutor可以创建自己的Future。
 */
@ThreadSafe
class CancellingExecutor extends ThreadPoolExecutor {
	// ...
    
    protected <T> RunnableFuture<T> newTaskFor(Callable<T> callable) {
        if (callable instanceof CancellableTask)
            return ((CancellableTask<T>) callable).newTask();
        else
            return super.newTaskFor(callable);
    }
}

/**
 * SocketUsingTask实现了CancellableTask，并定义了Future.cancel来关闭套接字和调用super.cancel。
 * 如果SocketUsingTask通过其自己的Future来取消，那么底层的套接字将被关闭并且线程将被中断。因此
 * 它提高了任务对取消操作的响应性：不仅能够在调用可中断方法的同时确保响应取消操作，而且还能调用可阻塞
 * 的套接字I/O方法。
 */
public abstract class SocketUsingTask<T> implements CancellableTask<T> {
    @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<T> newTask() {
        return new FutureTask<T>(this) {
            public boolean cancel(boolean mayInterruptIfRunning) {
                try {
                    SocketUsingTask.this.cancel();
                } finally {
                    return super.cancel(mayInterruptIfRunning);
                }
            }
        };
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58

7.2停止基于线程的服务

应用程序通常会创建拥有多个线程的服务，例如线程池，并且这些服务的生命周期通常比创建它们的方法的生命周期更长。如果应用程序准备退出，那么这些服务所拥有的线程也需要结束。
线程有一个相应的所有者，即创建该线程的类。因此线程池是其工作者线程的所有者，如果要中断这些线程，那么应该使用线程池。
与其他封装对象一样，线程的所有权是不可传递的：应用程序可以拥有服务，服务也可以拥有工作者线程，但应用程序并不能拥有工作者线程，因此应用程序不能直接停止工作者线程。相反，服务应该提供生命周期方法来关闭它自己以及它所拥有的线程。这样，当应用程序关闭该服务时，服务就可以关闭所有的线程了。在ExecutorService中提供了shutdown和shutdownNow等方法。同样，在其他拥有线程的服务中也应该提供类似的关闭机制。
对于持有线程的服务，只要服务的存在时间大于创建线程的方法的存在时间，那么就应该提供生命周期方法。

7.2.1示例：日志服务

/**
 * 在大多数服务器应用程序中都会用到日志，其中日志操作在单独的日志线程中执行。产生日志消息的线程并不会
 * 将消息直接写入输出流，而是由LogWriter通过BlockingQueue将消息提交给日志线程，并由日志线程写入。
 * 这是一种多生产者单消费者的设计方式：每个调用log的操作都相当于一个生产者，而后台的日志线程则相当于
 * 消费者。如果消费者的处理速度低于生产者的生成速度，那么BlockingQueue将阻塞生产者，直到日志线程有
 * 能力处理新的日志消息。
 */
public class LogWriter {
    private final BlockingQueue<String> queue;
    private final LoggerThread logger;
    private static final int CAPACITY = 1000;

    public LogWriter(Writer writer) {
        this.queue = new LinkedBlockingQueue<String>(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();
            }
        }
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44

为了能发挥实际的作用，还需要实现一种终止日志线程的方法，从而避免使JVM无法正常关闭。要停止日志线程是很容易的，因为它会反复调用take，而take能响应中断。如果将日志线程修改为当捕获到InterruptedException时退出，那么只需中断日志线程就能停止服务。
然而，如果只是使日志线程退出，那么还不是一种完备的关闭机制。这种直接关闭的做法会丢失那些正在等待被写入到日志的信息，不仅如此，其他线程将在调用log时被阻塞，因为此时日志消息队列可能是满的，因此这些线程将无法解除阻塞状态。
当取消一个生产者/消费者操作时，需要同时取消生产者和消费者。在中断日志线程时会处理消费者，但在这个示例中，由于生产者并不是专门的线程，因此要取消它们将非常困难。

另外一种关闭LogWriter的方法是：设置某个已请求关闭标志，以避免进一步提交日志消息：

public void log(String msg) throws InterruptedException {
	if (!shutdownRequested) {
		queue.put(msg);	
	} else {
		throw new IllegalStateException("logger is shut down");
	}
}
1
2
3
4
5
6
7

然而，在这个方法中存在着竞态条件问题，使得该方法并不可靠。log的实现是一种先判断再运行的代码序列：生产者发现该服务还没有关闭，因此在关闭服务后仍然会将日志消息放入队列，这同样会使得生产者可能在调用log时阻塞并且无法解除阻塞状态。

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

public class LogService {
    private final BlockingQueue<String> 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<String>();
        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();
            }
        }
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59

7.2.2关闭ExecutorService

ExecutorService提供了两种关闭方法：使用shutdown正常关闭，以及使用shutdownNow强行关闭。在进行强行关闭时，shutdownNow首先关闭当前正在执行的任务，然后返回所有尚未启动的任务清单。
强行关闭的速度更快，但风险也更大，因为任务很可能在执行到一半时被结束；而正常关闭虽然速度慢，但却更安全，因为ExecutorService会一直等到队列中的所有任务都执行完成后才关闭。在其他拥有线程的服务中也应该考虑提供类似的关闭方式以供选择。
在复杂程序中，通常会将ExecutorService封装在某个更高级别的服务中，并且该服务能提供自己的生命周期方法：

public class LogService {
	private final ExecutorService exec = new SingleThreadExecutor();

	// ...

	public void start() {}

	public void stop() throws InterruptedException {
		try {
			exec.shutdown();
			exec.awaitTermination(TIMEOUT, UNIT);
		} finally {
			writer.close();
		}
	}
	
	public void log(String msg) {
		try {
			exec.execute(new WriteTask(msg));
		} catch (RejectedExecutionException ignored) {}
	}
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

7.2.3毒丸对象

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

/**
 * 一个单生产者/单消费者的桌面搜索示例，其中使用毒丸对象来关闭服务。
 */
public class IndexingService {
	// 毒丸对象
    private static final File POISON = new File("");
    private final IndexerThread consumer = new IndexerThread();
    private final CrawlerThread producer = new CrawlerThread();
    private final BlockingQueue<File> queue;
    private final FileFilter fileFilter;
    private final File root;

	public void start() {
        producer.start();
        consumer.start();
    }

    public void stop() {
        producer.interrupt();
    }

	public void awaitTermination() throws InterruptedException {
        consumer.join();
    }

	/**
	 * 生产者线程
	 */
    class CrawlerThread extends Thread {
        public void run() {
            try {
                crawl(root);
            } catch (InterruptedException e) { /* 发生异常 */
            } finally {
                while (true) {
                    try {
                        queue.put(POISON);
                        break;
                    } catch (InterruptedException e1) { /* 重新尝试 */
                    }
                }
            }
        }

        private void crawl(File root) throws InterruptedException {
            // ...
        }
    }

	/**
	 * 消费者线程
	 */
    class IndexerThread extends Thread {
        public void run() {
            try {
                while (true) {
                    File file = queue.take();
                    if (file == POISON) {
						break;
					} else {
						indexFile(file);
					} 
                }
            } catch (InterruptedException consumed) {
            }
        }
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68

只有在生产者和消费者的数量都已知的情况下，才可以使用毒丸对象。在IndexingService中采用的解决方案可以扩展到多个生产者：只需每个生产者都向队列中放入一个毒丸对象，并且消费者仅当在接收到N_producers个毒丸对象时才停止。这种方法也可以扩展到多个消费者的情况，只需生产者将N_consumers个毒丸对象放入队列。然而，当生产者和消费者的数量较大时，这种方法将变得难以使用。只有在无界队列中，毒丸对象才能可靠地工作。

7.2.4示例：只执行一次的服务

如果某个方法需要处理一批任务，并且当所有任务都处理完成后才返回，那么可以通过一个私有的Executor来简化服务的生命周期管理，其中该Executor的生命周期是由这个方法来控制的(在这种情况下，invokeAll和invokeAny等方法通常会起较大的作用)。

/**
 * checkMail方法能在多台主机上并行地检查新邮件，它创建一个私有的Executor，并向每台主机提交一个任务。
 * 然后，当所有邮件检查任务都执行完成后，关闭Executor并等待结束。
 */
boolean checkMail(Set<String> hosts, long timeout, Timeunit unit) throws InterruptedException {
	ExecutorService exec = Executors.newCachedThreadPool();
	// 之所以采用AtomicBoolean来代替volatile类型的boolean，是因为能从内部的
	// Runnable中访问hasNewMail标志，因此它必须是final类型以免被修改
	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 {
		exec.shutdown();
		exec.awaitTermination(timeout, unit);
	}

	return hasNewMail.get();
}

private boolean checkMail(String host) {
	// 检查邮件
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31

7.2.5shutdownNow的局限性

无法通过常规方法来找出哪些任务已经开始但尚未结束。这意味着无法在关闭过程中知道正在执行的任务的状态，除非任务本身会执行某种检查。要知道哪些任务还没有完成，不仅需要知道哪些任务还没有开始，而且还需要知道当Executor关闭时哪些任务正在执行。
然而，在关闭过程中只会返回尚未开始的任务，而不会返回正在执行的任务。如果能返回所有这两种类型的任务，那么就不需要这种不确定的中间状态。

/**
 * 如何在关闭过程中判断正在执行的任务。通过封装ExecutorService并使得execute(类似地还有sumbt，
 * 在这里没有给出)记录哪些任务是在关闭后取消的，TrackingExecutor可以找出哪些任务已经开始但还没
 * 有正常完成。在Executor结束后，getCancelledTasks返回被取消的任务清单。要使这项技术能发挥作用，
 * 任务在返回时必须维持线程的中断状态，在所有设计良好的任务中都会实现这个功能。
 */
public class TrackingExecutor extends AbstractExecutorService {
    private final ExecutorService exec;
    private final Set<Runnable> tasksCancelledAtShutdown =
            Collections.synchronizedSet(new HashSet<Runnable>());

    public void shutdown() {
        exec.shutdown();
    }

    // 将ExecutorService的其他方法委托给exec

    public List<Runnable> getCancelledTasks() {
        if (!exec.isTerminated()) {
			throw new IllegalStateException(/*...*/);
		}  
        return new ArrayList<Runnable>(tasksCancelledAtShutdown);
    }

    public void execute(final Runnable runnable) {
        exec.execute(new Runnable() {
            public void run() {
                try {
                    runnable.run();
                } finally {
                	// 存在一个不可避免的竞态条件
                    if (isShutdown() && Thread.currentThread().isInterrupted()) {
						tasksCancelledAtShutdown.add(runnable);
					} 
                }
            }
        });
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39

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

/**
 * 网页爬虫程序的工作通常是无穷尽的，因此当爬虫程序必须关闭时，通常希望保存它的状态，
 * 以便稍后重新启动。当爬虫程序关闭时，无论是还没有开始的任务，还是那些被取消的任务，
 * 都将记录它们的URL，因此当爬虫程序重新启动时，就可以将这些URL的页面抓取任务加入到
 * 任务队列中。
 */
public abstract class WebCrawler {
    private volatile TrackingExecutor exec;
    @GuardedBy("this")
    private final Set<URL> urlsToCrawl = new HashSet<URL>();

	// ...

    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<URL> processPage(URL url);

    private void saveUncrawled(List<Runnable> 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;

		// ...

        public void run() {
            for (URL link : processPage(url)) {
                if (Thread.currentThread().isInterrupted()) {
					return;
				}
                submitCrawlTask(link);
            }
        }

        public URL getPage() { return url; }
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62

7.3处理非正常的线程终止

并发程序中的某个线程发生故障，问题通常并不会很明显。虽然可能会输出日志以及控制台信息，但没有人会时刻观察。此外，当线程发生故障时，应用程序可能看起来仍然在工作，所以这个失败很可能会被忽略。幸运的是，有可以监测并防止在程序中遗漏线程的方法。

导致线程提前死亡的最主要原因就是RuntimeException。由于这些异常表示出现了某种编程错误或者其他不可修复的错误，因此它们通常不会被捕获。它们不会在调用栈中逐层传递，而是默认地在控制台中输出栈追踪信息，并终止线程。线程非正常退出的后果可能是良性的，也可能是恶性的，这要取决于线程在应用程序中的作用
任何代码都可能抛出一个RuntimeException。每当调用另一个方法时，都要对它的行为保持怀疑，不要盲目地认为它一定会正常返回，或者一定会抛出在方法原型中声明的某个已检查异常。对调用的代码越不熟悉，就越应该对其代码行为保持怀疑。
不安全的线程应该在try-catch代码块中调用这些任务，这样就能捕获那些未检查的异常了，或者也可以使用try-finally代码块来确保框架能够知道线程非正常退出的情况，并做出正确的响应。在这种情况下，或许会考虑捕获RuntimeException，即当通过Runnable这样的抽象机制来调用未知的和不可信的代码时。

public void run() {
	Throwable thrown = null;

	try {
		while (!isInterrupted()) {
			runTask(getTaskFromWorkQueue);
		}
	} catch (Throwable e) {
		thrown = e;
	} finally {
		threadExited(this, thrown);
	}
}
1
2
3
4
5
6
7
8
9
10
11
12
13

如何在线程池内部构件一个工作者线程。如果任务抛出了一个未检查异常，那么它将使线程终结，但会首先通知框架该线程已经终结。然后，框架可能会用新的线程来代替这个工作线程，也可能不会，因为线程池正在关闭，或者当前已有足够多的线程能满足需要。
ThreadPoolExecutor和Swing都通过这项技术来确保行为糟糕的任务不会影响到后续任务的执行。当编写一个向线程池提交任务的工作者线程类时，或者调用不可信的外部代码时(例如动态加载的插件)，使用这些方法中的某一种可以避免某个编写得糟糕的任务或插件不会影响调用它的整个线程。

7.3.1未捕获异常的处理

在Thread API中同样提供了UncaughtExceptionHandler，它能检测出某个线程由于未捕获的异常而终结的情况。结合前面提到的主动方法来解决未检查异常，就能有效地防止线程泄漏的问题。
当一个线程由于未捕获异常而退出时，JVM会把这个事件报告给应用程序提供的UncaughtExceptionHandler异常处理器。如果没有提供任何异常处理器，那么默认的行为是将栈追踪信息输出到System.err。

public interface UncaughtExceptionHandler {
	void uncaughtException(Thread t, Throwable e);
}
1
2
3

异常处理器如何处理未捕获异常，取决于对服务质量的需求。最常见的响应方式是将一个错误信息以及相应的栈追踪信息写入应用程序日志中：

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);
		}
	}
}
1
2
3
4
5
6
7

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

要为线程池中的所有线程设置一个UncaughtExceptionHandler，需要为ThreadPoolExecutor的构造函数提供一个ThreadFactory(与所有的线程操控一样，只有线程的所有者能够改变线程的UncaughtExceptionHandler)。标准线程池允许当发生未捕获异常时结束线程，但由于使用了一个try-finally块来接收通知，因此当线程结束时，将有新的线程来代替它。如果没有提供捕获异常处理器或者其他的故障通知机制，那么任务会悄悄失败，从而导致极大的混乱。如果希望在任务由于发生异常而失败时获得通知，并且执行一些特定于任务的恢复操作，那么可以将任务封装在能捕获异常的Runnable或Callable中，或者改写ThreadPoolExecutor的afterExecute方法。
令人困惑的是，只有通过execute提交的任务，才能将它抛出的异常交给未捕获异常处理器，而通过submit提交的任务，无论是抛出的未检查异常还是已检查异常，都将被认为是任务返回状态的一部分。如果一个由submit提交的任务由于抛出了异常而结束，那么这个异常将被Future.get封装在ExecutionException中重新抛出。

7.4JVM关闭

7.4.1关闭钩子

在正常关闭中，JVM首先调用所有已注册的关闭钩子。关闭钩子是指通过Runtime.addShutdownHook注册的但尚未开始的线程。JVM并不能保证关闭钩子的调用顺序。
在关闭应用程序线程时，如果有(守护或非守护)线程仍然在运行，那么这些线程接下来将与关闭进程并发执行。当所有的关闭钩子都执行结束时，如果runFinalizersOnExit为true，那么JVM将运行终结器，然后再停止。JVM并不会停止或中断任何在关闭时仍然运行的应用程序线程。当JVM最终结束时，这些线程将被强行结束。如果关闭钩子或终结器没有执行完成，那么正常关闭进程挂起并且JVM必须被强行关闭。当被强行关闭时，只是关闭JVM，而不会运行关闭钩子。
关闭钩子应该是线程安全的：它们在访问共享数据时必须使用同步机制，并且小心地避免发生死锁，这与其他并发代码的要求相同。而且，关闭钩子不应该对应用程序的状态(例如，其他服务是否已经关闭，或者所有的正常线程是否已经执行完成)或者JVM的关闭原因做出任何假设，因此在编写关闭钩子的代码时必须考虑周全。最后，关闭钩子必须尽快退出，因为它们会延迟JVM的结束时间，而用户可能希望JVM能尽快终止。
关闭钩子可以用于实现服务或应用程序的清理工作，例如删除临时文件，或者清除无法由操作系统自动清除的资源：

public void start() {
	Runtime.getRuntime().addShutdownHook(new Thread() {
		public void run() {
			try {
				LogService.this.stop();
			} catch (InterruptedException ignored) {
			}
		}
	});
}
1
2
3
4
5
6
7
8
9
10

由于关闭钩子将并发执行，因此在关闭日志文件时可能导致其他需要日志服务的关闭钩子产生问题。为了避免这种情况，关闭钩子不应该依赖那些可能被应用程序或其他关闭钩子关闭的服务。实现这种功能的一种方式是对所有服务使用同一个关闭钩子(而不是每个服务使用一个不同的关闭钩子)，并且在该关闭钩子中执行一系列的关闭操作。这确保了关闭操作在单个线程中串行执行，从而避免了再关闭操作之间出现竞态条件或死锁等问题。
无论是否使用关闭钩子，都可以使用这项技术，通过将各个关闭操作串行执行而不是并行执行，可以消除许多潜在的故障。当应用程序需要维护多个服务之间的显式依赖信息时，这项技术可以确保关闭操作按照正确的顺序执行。

7.4.2守护线程

线程可分为两种：普通线程和守护线程。在JVM启动时创建的所有线程中，除了主线程以外，其他的线程都是守护线程(例如垃圾回收器以及其他执行辅助工作的线程)。当创建一个新线程时，新线程将继承创建它的线程的守护状态，因此在默认情况下，主线程创建的所有线程都是普通线程。
普通线程与守护线程之间的差异仅在于当线程退出时发生的操作。当一个线程退出时，JVM会检查其他正在运行的线程，如果这些线程都是守护线程，那么JVM会正常退出操作。当JVM停止时，所有仍然存在的守护线程都将被抛弃，既不会执行finally代码块，也不会执行回卷栈，而JVM只是直接退出。
应尽可能少地使用守护线程，很少有操作能够在不进行清理的情况下被安全地抛弃。特别是，如果在守护线程中执行可能包含I/O操作的任务，那么将是一种危险的行为。守护线程最好用于执行内部任务，例如周期性地从内存的缓存中移除逾期的数据。

7.4.3终结器

当不再需要内存资源时，可以通过垃圾回收器来回收它们，但对于其他一些资源，例如文件句柄或套接字句柄，当不再需要它们时，必须显式地交还给操作系统。为了实现这个功能，垃圾回收器对那些定义了finalize方法的对象会进行特殊处理：在回收器释放它们后，调用它们的finalize方法，从而保证一些持久化的资源被释放。
由于终结器可以在某个由JVM管理的线程中运行，因此终结器访问的任何状态都可能被多个线程访问，这样就必须对其访问操作进行同步。终结器并不能保证它们将在何时运行甚至是否会运行，并且复杂的终结器通常还会再对象上产生巨大的性能开销。要编写正确的终结器是非常困难的。在大多数情况下，通过使用finally代码块和显式的close方法，能够比使用终结器更好地管理资源。唯一的例外情况在于：当需要管理对象，并且该对象持有的资源是通过本地方法获得的。基于这些原因以及其他一些原因，要尽量避免编写或使用包含终结器的类(除非是平台库中的类)。