Java 中 InterruptedException 的最佳实

引子

我们经常会遇到在一个 Thread 中执行阻塞操作 (blocking method) 时，IDE 会提示我们需要 catch InterruptedException。我在工程实践中，经常看到大家的写法是

try {
    // do something blocking
} catch (InterruptedException e) {
    e.printStackTrace();
}

这就引发了我的思考，因为我记得在《Effective Java》中有说过，不要忽略任何异常，即使你只是把异常 log 下来也好。

• 那么对于 InterruptedException 有没有什么最佳实践呢？

最近我在看一个 Websocket 的 Java 源码实现，发现里面有如下写法：

try {
    while( !Thread.interrupted() ) {
        ByteBuffer buffer = engine.outQueue.take();
        ostream.write( buffer.array(), 0, buffer.limit() );
        ostream.flush();
    }
} catch ( InterruptedException e ) {
    for (ByteBuffer buffer : engine.outQueue) {
        ostream.write( buffer.array(), 0, buffer.limit() );
        ostream.flush();
    }
    Thread.currentThread().interrupt();
}

上面的代码块是在一个被 Thread 执行的 Runnable 中。try 中的 while(!Thread.interrupted()) 是一个很常见的写法：它表示只要这个线程没有被打断，就会一直死循环下去，因此线程不会退出。我注意到下面 catch 代码块的写法：

catch ( InterruptedException e ) {
    ...
    Thread.currentThread().interrupt();
}

这里就很奇怪了：为什么该线程明明已经被 interrupted 了，还要显式地去执行 Thread.currentThread().interrupt() 呢？

Blocking methods

当一个方法抛出 InterruptedException 时，它其实是想告诉你，它是一个 blocking method，并且它尝试 unblock 后提前返回 (return early)。
阻塞方法不同于普通方法的一个点就在于，任务的完成不完全取决于自身，还取决于一些外部事件：I/O 操作，其他线程的行为（释放锁、设置标志位、把一个任务加入到队列等）。这就导致阻塞方法的完成是不可预测的，因此一个阻塞方法应该是可取消的（cancelable）。这就是 interruption 机制， Thread.sleep(), Object.wait() 都是提供这样的机制。

Thread interruption

每一个线程都拥有一个 interrupted status，初始值是 false。当一个线程被另一个线程调用 Thread.interrupt() 而中止时，jvm 默默地做了一下事情：

1. 如果被中止线程正阻塞在 Thread.sleep()，Thread.join()，或者 Object.wait() 这样的阻塞方法上，那么它将 unblock 并抛出 InterruptedException 异常。
2. 否则 interrupt() 方法只是将 interrupted status 置为 true。运行在这个线程的阻塞方法之后的代码块，就可以通过 Thread.isInterrupted() 来检查该线程是否已被 interrupted，来决定执行哪部分代码逻辑（因为程序运行到这部分代码块可能是因为被 interrupted 而提前往下执行到了这里，也可能是因为阻塞方法被成功执行完了而往下执行到了这里）
   所以本质上，当一个线程通过调用 Thread.interrupt() 方法去中止另一个线程时，另一个线程并不是马上中止（主要看线程对 interruption 的处理，即这个线程意识到 interruption 了，是否提前退出 (return early) 取决于它自己）。阻塞方法通过捕获 InterruptionException 异常来意识到 interruption 事件，而非阻塞的任务则通过轮询 interrupted status 来意识是 interruption 事件。

处理 InterruptedException

• 自己不处理善后工作，直接向上抛异常

public class TaskQueue {
    private static final int MAX_TASKS = 1000;
 
    private BlockingQueue<Task> queue 
        = new LinkedBlockingQueue<Task>(MAX_TASKS);
 
    public void putTask(Task r) throws InterruptedException { 
        // put 是阻塞方法
        queue.put(r);
    }
 
    public Task getTask() throws InterruptedException { 
        // take 是阻塞方法
        return queue.take();
    }
}

适用于，当你意识到 interruption 时，你没有什么需要 “善后收尸” 的工作需要处理，因此仅仅需要告诉调用者 interruption 已经发生。

• 自己处理善后工作，之后向上抛异常

public class PlayerMatcher {
    private PlayerSource players;
 
    public PlayerMatcher(PlayerSource players) { 
        this.players = players; 
    }
 
    public void matchPlayers() throws InterruptedException { 
        Player playerOne, playerTwo;
         try {
             while (true) {
                 playerOne = playerTwo = null;
                 // Wait for two players to arrive and start a new game
                 playerOne = players.waitForPlayer(); // could throw IE
                 playerTwo = players.waitForPlayer(); // could throw IE
                 startNewGame(playerOne, playerTwo);
             }
         }
         catch (InterruptedException e) {  
             // If we got one player and were interrupted, put that player back
             if (playerOne != null)
                 players.addFirst(playerOne);
             // Then propagate the exception
             throw e;
         }
    }
}

matchPlayers() 等待两位玩家到达后开始游戏，如果一个玩家已经到了，并在等待第二个玩家，此时线程被中止，那么线程在捕获异常后需要做如下 “善后收尸” 工作：将已到达玩家压栈，以保留这个玩家的信息，待系统恢复后重建这个玩家的状态，然后将该 InterruptedException 向上抛出。

• 不能向上抛异常时，怎么通知上层异常发生呢？

有时，你不能显式地把异常向上传递。

public class TaskRunner implements Runnable {
    private BlockingQueue<Task> queue;
 
    public TaskRunner(BlockingQueue<Task> queue) { 
        this.queue = queue; 
    }
 
    public void run() { 
        try {
             while (true) {
                 Task task = queue.take(10, TimeUnit.SECONDS);
                 task.execute();
             }
         }
         catch (InterruptedException e) { 
             // Restore the interrupted status
             Thread.currentThread().interrupt();
         }
    }
}

阻塞方法在 Runnable.run() 中，不能重新向上抛出 InterruptedException。而 catch （InterruptedException e) 时，jvm 认为你已经意识到 interruption 事件了，所以将 interrupted status 置为 false。此时你要做的就是显式调用 Thread.currentThread().interrupt(); 来重置状态位，这样上层调用者可以通过 Thread.isInterrupted() 来检查 interruption 事件状态。

• 最不该做的

最不该做的就是忽略这个异常

// Don't do this 
public class TaskRunner implements Runnable {
    private BlockingQueue<Task> queue;
 
    public TaskRunner(BlockingQueue<Task> queue) { 
        this.queue = queue; 
    }
 
    public void run() { 
        try {
             while (true) {
                 Task task = queue.take(10, TimeUnit.SECONDS);
                 task.execute();
             }
         }
         catch (InterruptedException swallowed) { 
             /* DON'T DO THIS - RESTORE THE INTERRUPTED STATUS INSTEAD */
         }
    }
}

只有一种情况可以忽略这个异常：你显式地知道线程的退出完全由你操作，即线程不直接暴露给外部调用者，而只暴露给调用者 cancel 它的接口。

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(); }
}

• 并不是所有的阻塞方法都会抛出 InterruptedException，比如 stream 的相关阻塞方法（read() 等）就只会抛出 IOException。

• 不可取消的任务

有些情况下，你可能不希望你的任务被中止，这种就是不可取消的任务。但是这种情况同样需要保留 interrupted exception。

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();
    }
}