首页 > 代码库 > <<java 并发编程>>第七章:取消和关闭
<<java 并发编程>>第七章:取消和关闭
Java没有提供任何机制来安全地终止线程,虽然Thread.stop和suspend等方法提供了这样的机制,但是存在严重的缺陷,应该避免使用这些方法。但是Java提供了中断Interruption机制,这是一种协作机制,能够使一个线程终止另一个线程的当前工作。
这种协作方式是必要的,我们很少希望某个任务线程或者服务立即停止,因为这种立即停止会时某个共享的数据结构处于不一致的状态。相反,在编写任务和服务的时候可以使用一种协作方式:当需要停止的时候,它们会先清除当前正在执行的工作,然后再结束。
7.1 任务取消
如果外部代码能够在某个操作正常完成之前将其置于 完成 状态,那么这个操作就可以称为可取消的Cancellable
其中一种协作机制是设置一个取消标志Cancellation Requested标志,而任务定期查看该标志。
- @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);
- }
- static List<BigInteger> aSecondOfPrimes() throws InterruptedException {
- PrimeGenerator generator = new PrimeGenerator();
- exec.execute(generator);
- try {
- SECONDS.sleep(1);
- } finally {
- generator.cancel();
- }
- return generator.get();
- }
- }
在Java的API或语言规范中,并没有将中断与任何语义关联起来,但实际上,如果在取消之外的其他操作中使用中断,那么都是不合适的,并且很难支撑起更大的应用。
每个线程都有一个boolean类型的中断状态。但中断线程时,这个线程的中断状态将被设置成true。
Thread中的三个方法:
public void interrupt() 中断一个线程
public boolean isInterrupted() 获取中断标志,判断是否中断
public static boolean interrupted() 清楚中断状态,并返回它之前的状态值
线程在阻塞状态下发生中断的时候会抛出InterruptedException,例如Thread.sleep(), Thread.wait(), Thread.join()等方法。
当线程在非阻塞状态下中断的时候,它的中断状态将被设置,然后根据检查中断状态来判断是否中断。
调用interrupt并不意味着立即停止目标线程正在进行的工作,而只是传递了请求中断的消息,换句话说,仅仅修改了线程的isInterrupted标志字段。
通常,中断时实现取消的最合理方式。
- 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();
- }
- }
7.1.3 响应中断
只有实现了线程中断策略的代码才可以屏蔽中断请求,在常规任务和库代码中都不应该屏蔽中断请求。
两种方法响应中断:
- public class NoncancelableTask {
- public Task getNextTask(BlockingQueue<Task> 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 {
- }
- }
- 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 {
- 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
- }
- }
- }
- 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);
- }
- }
- };
- }
- }
- interface CancellableTask<T> extends Callable<T> {
- void cancel();
- RunnableFuture<T> newTask();
- }
7.2 停止基于线程的服务
应用程序通常会创建拥有多个线程的服务,如果应用程序准备退出,那么这些服务所拥有的线程也需要正确的结束,由于java没有抢占式方法停止线程,因此需要它们自行结束。
正确的封装原则是:除非拥有某个线程,否则不要对该线程进行操控,例如中断线程或者修改优先级等。
线程有个相应的所有者,即创建该线程的类,因此线程池是其工作者线程的所有者,如果要中断线程,那么应该使用线程池去中断。
线程的所有权是不可传递的。服务应该提供生命周期方法Lifecycle Method来关闭它自己以及它所拥有的线程。这样当应用程序关闭该服务的时候,服务就可以关闭所有的线程了。在ExecutorService中提供了shutdown和shutdownNow等方法,同样,在其他拥有线程的服务方法中也应该提供类似的关闭机制。
Tips:对于持有线程的服务,只要服务的存在时间大于创建线程的方法的存在时间,那么就应该提供生命周期方法。
7.2.1 示例:日志服务
我们通常会在应用程序中加入log信息,一般用的框架就是log4j。但是这种内联的日志功能会给一些高容量Highvolume应用程序带来一定的性能开销。另外一种替代方法是通过调用log方法将日志消息放入某个队列中,并由其他线程来处理。
- 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();
- }
- }
- }
- }
7.2.2 通过ExecutorService去关闭
简单的程序可以直接在main函数中启动和关闭全局的ExecutorService,而在复杂程序中,通常会将ExecutorService封装在某个更高级别的服务中,并且该服务提供自己的生命周期方法。下面我们利用ExecutorService重构上面的日志服务:
- public class LogService {
- public void stop() throws InterruptedException {
- try {
- exec.shutdown(); exec.awaitTermination(TIMEOUT, UNIT);
- }
- }
- }
7.2.3 利用Poison Pill对象关闭Producer-Consumer服务
7.2.5 当关闭线程池的时候,保存尚未开始的和开始后取消的任务数据,以备后面重新处理,下面是一个网页爬虫程序,关闭爬虫服务的时候将记录所有尚未开始的和已经取消的所有页面URL:
- public abstract class WebCrawler {
- private volatile TrackingExecutor exec;
- @GuardedBy("this")
- private final Set<URL> urlsToCrawl = new HashSet<URL>();
- private final ConcurrentMap<URL, Boolean> seen = new ConcurrentHashMap<URL, Boolean>();
- 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<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;
- 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;
- }
- }
- }
- public class TrackingExecutor extends AbstractExecutorService {
- private final ExecutorService exec;
- private final Set<Runnable> tasksCancelledAtShutdown =
- Collections.synchronizedSet(new HashSet<Runnable>());
- public TrackingExecutor(ExecutorService exec) {
- this.exec = exec;
- }
- public void shutdown() {
- exec.shutdown();
- }
- public List<Runnable> 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<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);
- }
- }
- });
- }
- }
7.3 处理非正常的线程终止
通过给应用程序提供一个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);
- }
- }
只有通过execute提交的任务,才能将它抛出的异常交给未捕获异常处理器。而通过submit提交的任务,无论是抛出未检查异常还是已检查异常,都将被认为是任务返回状态的一部分
7.4 JVM关闭的时候提供关闭钩子
在正常关闭JVM的时候,JVM首先调用所有已注册的关闭钩子Shutdown Hook。关闭钩子可以用来实现服务或者应用程序的清理工作,例如删除临时文件,或者清除无法由操作系统自动清除的资源。
最佳实践是对所有服务都使用同一个关闭钩子,并且在该关闭钩子中执行一系列的关闭操作。这确保了关闭操作在单个线程中串行执行,从而避免竞态条件的发生或者死锁问题。
- Runtime.getRuntime().addShutdownHook(new Thread() {
- public void run() {
- try{LogService.this.stop();} catch(InterruptedException) {..}
- }
- })
总结:在任务、线程、服务以及应用程序等模块中的生命周期结束问题,可能会增加它们在设计和实现的时候的复杂性。我们通过利用FutureTask和Executor框架,可以帮助我们构建可取消的任务和服务。
<<java 并发编程>>第七章:取消和关闭