并发（8）：终结任务（上）

ゝ一世哀愁。 2023-03-12 07:20 9阅读 0赞

在前面的某些示例中，cancel()和isCanceled()方法被放到了一个所有任务都可以看到的类中。这些任务通过检查isCanceled()来确定何时终止它们自己，对于这个问题来说，这是一种合理方式。但是，在某些情况下，任务必须更加突然地终止。本节将学习到有关这种终止的各类话题和问题。

首先，让我们观察一个示例，它不仅演示了终止问题，而且还是一个资源共享的示例。

### 一、装饰性花园 ###

在这个仿真程序中，花园委员会希望了解每天通过多个大门进入公园的总人数。每个大门都有一个十字转门或某种其他形式的计数器，并且任何一个十字转门的计数值递增时，就表示公园中的总人数的共享计数值也会递增。

package concurrency;
    
    import java.util.ArrayList;
    import java.util.List;
    import java.util.Random;
    import java.util.concurrent.ExecutorService;
    import java.util.concurrent.Executors;
    import java.util.concurrent.TimeUnit;
    
    class Count {
    	private int count = 0;
    	private Random r = new Random();
    
    	// 删除synchronized关键字以查看计数失败
    	public synchronized int increment() {
    		int temp = count;
    		if (r.nextBoolean())
    			Thread.yield();
    		return (count = ++temp);
    	}
    
    	public synchronized int value() {
    		return count;
    	}
    }
    
    /**
     * 入口
     */
    class Entrance implements Runnable {
    
    	private static Count count = new Count();
    	private static List<Entrance> entrances = new ArrayList<>();
    	// 不需要同步读取
    	private int number = 0;
    	private final int id;
    	// 在不稳定的字段上的原子操作
    	private static volatile boolean canceled = false;
    
    	// 取消
    	public static void cancel() {
    		canceled = true;
    	}
    
    	public Entrance(int id) {
    		this.id = id;
    		// 将此任务保存在列表中
    		entrances.add(this);
    	}
    
    	@Override
    	public void run() {
    		while (!canceled) {
    			synchronized (this) {
    				++number;
    			}
    			System.out.println(this + " Total: " + count.increment());
    			try {
    				TimeUnit.MILLISECONDS.sleep(100);
    			} catch (InterruptedException e) {
    				e.printStackTrace();
    			}
    		}
    		System.out.println("Stopping " + this);
    	}
    
    	public synchronized int getValue() {
    		return number;
    	}
    
    	public String toString() {
    		return "Entrance " + id + ": " + getValue();
    	}
    
    	public static int getTotalCount() {
    		return count.value();
    	}
    
    	public static int sumEntrances() {
    		int sum = 0;
    		for (Entrance entrance : entrances) {
    			sum += entrance.getValue();
    		}
    		return sum;
    	}
    }
    
    /**
     * 装饰花园
     */
    public class OrnamentalGarden {
    	public static void main(String[] args) throws InterruptedException {
    		ExecutorService exec = Executors.newCachedThreadPool();
    		for (int i = 0; i < 5; i++) {
    			exec.execute(new Entrance(i));
    		}
    		// 运行一段时间，然后停止并收集数据
    		TimeUnit.SECONDS.sleep(3);
    		Entrance.cancel();
    		exec.shutdown();
    		if (!exec.awaitTermination(250, TimeUnit.MILLISECONDS))
    			System.out.println("有些任务没有终止");
    		System.out.println("Total: " + Entrance.getTotalCount());
    		System.out.println("Sum of Entrances: " + Entrance.sumEntrances());
    	}
    }

这里使用单个的Count对象来跟踪花园参观者的主计数值，并且将其当作Entrance类中的一个静态域进行存储。Count.increment()和Count.value()都是synchronized的，用来控制对count域的访问。increment()方法使用了Random对象，目的是在从把count读取到temp中，到递增temp并将其存储回count的这段时间里，有大约一半的时间产生让步。如果你将increment()上的synchronized关键字注释掉，那么这个程序就会崩溃，因为多个任务将同时访问并修改count（yield()会使问题更快的发生）。

每个Entrance任务都维护着一个本地值number，它包含通过某个特定入口进入的参观者的数量。这提供了对count对象的双重检查，以确保其记录的参观者数量是正确的。Entrance.run()只是递增number和count对象，然后休眠100毫秒。

因为Entrance.canceled是一个volatile布尔标识，而它只会被读取和赋值（不会与其他域组合在一起被读取），所以不需要同步对其的访问，就可以安全地操作它。如果你对诸如此类的情况有任何疑惑，那么最好总是使用synchronized。

这个程序在以稳定的方式关闭所有事物方面还有一些小麻烦，其部分原因是为了说明在终止多线程程序时你必须相当小心，而另一部分原因是为了演示interrupt()的值，稍后你将学习有关这个值的知识。

在3秒钟之后，main()向Entrance发送static cancel()消息，然后调用exec对象的shutdown()方法，之后调用exec上的awaitTermination()方法。ExecutorService.awaitTermination()等待每个任务结束，如果所有的任务在超时时间到达之前全部结束，则返回true，否则返回false，表示不是所有任务都已经结束了。尽管这会导致每个任务都退出其run()方法，并因此作为任务而终止，但是Entrance对象仍旧是有效的，因为在构造器中，每个Entrance对象都存储在称为entrances的静态List<Entrance>中。因此，sumEntrances()仍旧可以作用于这些有效的Entrance对象。

当这个程序运行时，你将看到，在人们通过十字转门时，将显示总人数和通过每个入口的人数。如果移除Count.increment()上面的synchronized声明，你将会注意到总人数与你的期望有差异，每个十字转门统计的人数将与count中的值不同。只要用互斥来同步对Count的访问，问题就可以解决了。请记住，Count.increment()通过使用temp和yield()，增加了失败的可能性。在真正的线程问题中，失败的可能性从统计学角度看可能非常小，因此你可能很容易就掉进了轻信所有事物都将正确工作的陷阱里。就像在上面的示例中，有些还未发生的问题就有可能会隐藏起来，因此在复审并发代码时，要格外地仔细。

### 二、在阻塞时终结 ###

前面的示例中的Entrance.run()在其循环中包含对sleep()的调用。我们知道，sleep()最终将唤醒，而任务也将返回循环的开始开始部分，去检查canceled标志，从而决定是否跳出循环。但是，sleep()一种情况，它使任务从执行状态变为阻塞状态，而有时你必须终止被阻塞的任务。

**（1）线程状态**

一个线程可以处于以下四种状态之一：

1.  新建（new）：当线程被创建时，它只会短暂的处于这种状态。此时它已经分配了必需的系统资源，并执行了初始化。此刻线程已经有资格获得CPU时间了，之后调度器将把这个线程转变为可运行状态或阻塞状态。
2.  就绪（Runnable）：这种状态下，只要调度器把时间分配给线程，线程就可以运行。也就是说，在任意时刻，线程可以运行也可以不运行。只要调度器能分配时间片段给线程，它就可以运行；这不同于死亡和阻塞状态。
3.  阻塞（Blocked）：线程能够运行，但有某个条件阻止它的运行。当线程处于阻塞状态时，调度器将忽略线程，不会分配给线程任何CPU时间。直到线程重新进入了就绪状态，它才有可能执行操作。
4.  死亡（Dead）：处于死亡或终止状态的线程将不再是可调度的，并且再也不会得到CPU时间，它的任务以结束，或不再是可运行的。任务死亡的通常方式是从run()方法返回，但是任务的线程还可以被中断，你将要看到这一点。

**（2）进入阻塞状态**

一个任务进入阻塞状态，可能有如下原因：

1.  通过调用sleep(milliseconds)使任务进入休眠状态，在这种情况下，任务在指定的时间内不会运行。
2.  你通过调用wait()使线程挂起。直到线程得到了notify()或notifyAll()消息（或者在java SE5的java.util.concurrent类库中等价的signal()或signalAll()消息），线程才会进入就绪状态。我们将在稍后的小节中验证这一点。
3.  任务在等待某个输入/输出完成。
4.  任务试图在某个对象上调用同步控制方法，但是对象锁不可用，因为另一个任务已经获取了这个锁。

在较早的代码中，也可能会看到用suspend()和resume()来阻塞和唤醒线程，但是在现代java中这些方法被废止了（因为可能导致死锁），所以将不再讨论这些内容。stop()方法也已经被废止了，因为它不释放线程获得的锁，并且如果线程处于不一致的状态（受损状态），其他任务可以在这种状态下浏览并修改它们。这样所产生的问题是微妙而难以被发现的。

现在我们需要查看的问题是：有时你希望能够终止处于阻塞状态的任务。如果对于处于阻塞状态的任务，你不能等待其到达代码中可以检查其状态值的某一点，因为决定让它主动的终止，那么你就必须强制这个任务跳出阻塞状态。

### 三、中断 ###

正如你所想象的，在Runnable.run()方法的中间打断它，与等待该方法到达对cancel标志的测试，或者到达程序员准备好离开该方法的其他一些地方相比，要棘手得多。当你打断被阻塞的任务时，可能需要清理资源。正因为这一点，在任务的run()方法中间打断，更像是抛出的异常，因此在java线程中的这种类型的异常中断中用到了异常（这会滑向异常的不恰当用法，因为这意味着你经常用它们来控制执行流程）。为了在以这种方式终止任务时，返回众所周知的良好状态，你必须仔细考虑代码的执行路径，并仔细编写catch子句以正确清除所有事物。

Thread类包含interrupt()方法，因此你可以终止被阻塞的任务，这个方法将设置线程的中断状态。如果一个线程已经被阻塞，或者试图执行一个阻塞操作，那么设置这个线程的中断状态将抛出InterruptedException。当抛出该异常或者该任务调用Thread.interrupted()时，中断状态将被复位。正如你将看到的，Thread.interrupted()提供了离开run()循环而不抛出异常的第二种方式。

为了调用interrupt()，你必须持有Thread对象。你可能已经注意到了，新的concurrent类库似乎在避免对Thread对象的直接操作，转而尽量通过Executor来执行所有操作。如果你在Executor上调用shutDownNow()，那么它将发送一个interrupt()调用给它启动的所有线程。这么做是有意义的，因为当你完成工程中的某个部分或者整个程序时，通常会希望同时关闭某个特定Executor的所有任务。然而，你有时也会希望只中断某个单一任务。如果使用Executor，那么通过调用submit()而不是executor()来启动任务，就可以持有该任务的上下文。submit()将返回一个泛型Future<?>，其中有一个未修饰的参数，因为你永远都不会在其上调用get()——持有这中Future的关键在于你可以在其上调用cancel()，并因此可以使用它来中断某个特定任务。如果你将true传递给cancel()，那么它就会拥有在该线程上调用interrupt()以停止这个线程的权限。因此，cancel()是一种中断由Executor启动的单个线程的方式。

下面的示例用Executor展示了基本的interrupt()用法：

package concurrency;
    
    import java.io.IOException;
    import java.io.InputStream;
    import java.util.concurrent.ExecutorService;
    import java.util.concurrent.Executors;
    import java.util.concurrent.Future;
    import java.util.concurrent.TimeUnit;
    
    class SleepBlocked implements Runnable {
    	@Override
    	public void run() {
    		try {
    			TimeUnit.SECONDS.sleep(100);
    		} catch (InterruptedException e) {
    			System.out.println("InterruptedException");
    		}
    		System.out.println("Exiting sleepBlocked.run()");
    	}
    }
    
    class IOBlocked implements Runnable {
    
    	private InputStream in;
    
    	public IOBlocked(InputStream is) {
    		in = is;
    	}
    
    	@Override
    	public void run() {
    		System.out.println("Waiting for read():");
    		try {
    			in.read();
    		} catch (IOException e) {
    			if (Thread.currentThread().isInterrupted()) {
    				System.out.println("Interrupted from blocked I/O");
    			} else {
    				throw new RuntimeException(e);
    			}
    		}
    		System.out.println("Exiting IOBlocked.run()");
    	}
    }
    
    class SynchronizedBlocked implements Runnable {
    
    	public synchronized void f() {
    		while (true) // 不释放锁
    			Thread.yield();
    	}
    
    	public SynchronizedBlocked() {
    		new Thread() {
    			public void run() {
    				f(); // 这个线程获取的锁
    			}
    		}.start();
    	}
    
    	@Override
    	public void run() {
    		System.out.println("尝试调用f()");
    		f();
    		System.out.println("Exiting synchronizedBlock.run()");
    	}
    }
    
    public class Interrupting {
    	private static ExecutorService exec = Executors.newCachedThreadPool();
    
    	static void test(Runnable r) throws InterruptedException {
    		Future<?> f = exec.submit(r);
    		TimeUnit.MILLISECONDS.sleep(100);
    		System.out.println("Interrupting " + r.getClass().getName());
    		f.cancel(true); // 如果中断运行
    		System.out.println("中断发送到 " + r.getClass().getName());
    	}
    
    	public static void main(String[] args) throws InterruptedException {
    		test(new SleepBlocked());
    		test(new IOBlocked(System.in));
    		test(new SynchronizedBlocked());
    		TimeUnit.SECONDS.sleep(3);
    		System.out.println("Aborting with System.exit(0)");
    		System.exit(0);
    	}
    }

上面的每个任务都表示了一种不同类型的阻塞。SleepBlock是可中断的阻塞示例，而IOBlocked和SynchronizedBlocked是不可中断的阻塞示例。这个程序证明I/O和在synchronized块上的等待是不可中断的，但是通过浏览代码，你也可以预见到这一点——无论是I/O还是尝试调用synchronized方法，都不需要任何InterruptedException处理器。

前两个类很简单直观：在第一个类中run()方法调用了sleep()，而在第二个类中调用了read()。但是，为了演示SynchronizedBlock，我们必须首先获取锁。这是通过在构造器中创建匿名的Thread类的实例来实现的，这个匿名Thread类的对象通过调用f()获取了对象锁（这个线程必须有别于SynchronizedBlock驱动run()的线程，因为一个线程可以多次获得某个对象锁）。由于f()永远都不返回，因此这个锁永远不会释放，而SynchronizedBlock.run()在试图调用f()，并阻塞以等待这个锁释放。

从输出中可以看到，你能够中断对sleep()的调用（或者任何要求抛出InterruptedException的调用）。但是，你不能中断正在试图获取synchronized锁或者试图执行I/O操作的线程。这有点令人烦恼，特别是在创建执行I/O的任务时，因为这意味着I/O具有锁住你的多线程程序的潜在可能。特别是对于基于Web的程序，这更是关乎利害。

对于这类问题，有一个略显笨拙但是有时确实行之有效的解决方案，即关闭任务在其上发生阻塞的底层资源：

package concurrency;
    
    import java.io.IOException;
    import java.io.InputStream;
    import java.net.ServerSocket;
    import java.net.Socket;
    import java.util.concurrent.ExecutorService;
    import java.util.concurrent.Executors;
    import java.util.concurrent.TimeUnit;
    
    public class CloseResource {
    	@SuppressWarnings("resource")
    	public static void main(String[] args) throws IOException, InterruptedException {
    		ExecutorService exec = Executors.newCachedThreadPool();
    		ServerSocket server = new ServerSocket(8080);
    		InputStream socketInput = new Socket("localhost", 8080).getInputStream();
    		exec.execute(new IOBlocked(socketInput));
    		exec.execute(new IOBlocked(System.in));
    		TimeUnit.MILLISECONDS.sleep(100);
    		System.out.println("关闭所有线程");
    		exec.shutdownNow();
    		TimeUnit.SECONDS.sleep(1);
    		System.out.println("Closing " + socketInput.getClass().getName());
    		socketInput.close(); // 释放阻塞的线程
    		TimeUnit.SECONDS.sleep(1);
    		System.out.println("Closing " + System.in.getClass().getName());
    		System.in.close(); // 释放阻塞的线程
    	}
    }

在shutDownNow()被调用之后以及在两个输入流上调用close()之前的延迟强调的是一旦底层资源被关闭，任务将解除阻塞。请注意，有一点很有趣，interrupt()看起来发生在关闭Socket而不是关闭System.in的时刻。

幸运的是，在I/O系统章节中介绍的各种nio类提供了更人性化的I/O中断。被阻塞的nio通道会自动地响应中断：

package concurrency;
    
    import java.io.IOException;
    import java.net.InetSocketAddress;
    import java.net.ServerSocket;
    import java.nio.ByteBuffer;
    import java.nio.channels.AsynchronousCloseException;
    import java.nio.channels.ClosedByInterruptException;
    import java.nio.channels.SocketChannel;
    import java.util.concurrent.ExecutorService;
    import java.util.concurrent.Executors;
    import java.util.concurrent.Future;
    import java.util.concurrent.TimeUnit;
    
    class NIOBlocked implements Runnable {
    
    	private final SocketChannel sc;
    
    	public NIOBlocked(SocketChannel sc) {
    		this.sc = sc;
    	}
    
    	@Override
    	public void run() {
    		System.out.println("Waiting for read() in " + this);
    		try {
    			sc.read(ByteBuffer.allocate(1));
    		} catch (ClosedByInterruptException e) {
    			System.out.println("ClosedByInterruptException");
    		} catch (AsynchronousCloseException e) {
    			System.out.println("AsynchronousCloseException");
    		} catch (IOException e) {
    			throw new RuntimeException(e);
    		}
    		System.out.println("Exiting NIOBlocked.run() " + this);
    	}
    }
    
    public class NIOInterruption {
    	public static void main(String[] args) throws IOException, InterruptedException {
    		ExecutorService exec = Executors.newCachedThreadPool();
    		ServerSocket server = new ServerSocket(8080);
    		InetSocketAddress isa = new InetSocketAddress("localhost", 8080);
    		SocketChannel sc1 = SocketChannel.open(isa);
    		SocketChannel sc2 = SocketChannel.open(isa);
    		Future<?> f = exec.submit(new NIOBlocked(sc1));
    		exec.execute(new NIOBlocked(sc2));
    		exec.shutdown();
    		TimeUnit.SECONDS.sleep(1);
    		// 通过取消产生一个中断
    		f.cancel(true);
    		TimeUnit.SECONDS.sleep(1);
    		sc2.close();
    	}
    }

如你所见，你还可以关闭底层资源以释放锁，尽管这种做法一般不是必需的。注意，使用execute()来启动两个任务，并调用e.shutdownNow()可以很容易地终止所有事物，而对于捕获上面示例中的Future，只有在将中断发送给一个线程，同时不发送给另一个线程时才是必需的。

**如果本文对您有很大的帮助，还请点赞关注一下。**