Java线程中断interrupt详解

摘要: 介绍Java线程interrupt的使用

Java线程中断interrupt详解

Java多线程编程中，中断一直是一个难以理解的点。

中断状态标志The Interrupt Status Flag

每个Java线程（Thread对象）都拥有一个标志位，即中断状态interrupt status，有两个值，true和false。默认上情况下，interrupt status值为false，即本线种没有任何中断；当调用了线程对象的interrupt()方法后，interrupt status值为true，表示线程收到中断请求。

程序可以通过isInterrupted()方法，读取中断标志位的值；同时Thread.interrupted()静态方法也能读取中断标志，与isInterrupted()不同的是，还会同时清除中断标志。

中断的用途

线程收到中断时，通常表示应该停止当前的工作，比如退出循环、停止监听、从wait/sleep/join状态立即结束等。具体处理方式完全由程序决定。

1、一个正常运行的线程，收到中断后，不会有任何问题，可以不需要做任何处理，当然程序可以通过显式的判断isInterrupted，来决定下一步动作，比如下面这样的：

// 其他代码
while (!Thread.currentThread().isInterrupted()) {
// 循环做某件事情
}

2、当线程被Object.wait、Thread.join和Thread.sleep三种方法之一阻塞时，这时如果收到中断请求，则会抛出InterruptedException并清除中断标志。这是中断非常有用的一点，可以提前从阻塞状态返回，稍后用代码示范。

3、做为一种约定，Java API里任何声明为抛出InterruptedException的方法，在抛出异常之前，都会先清除掉中断标志。

示例

我们通过几个例子来演示一下如何使用中断。

中断的基本应用示例

简单的中断示例，包括对sleep/wait/join状态线程的中断操作。

package org.alive.learn.thread;

/**
* <p>
* 测试Java线程interrupt
*
* @author myumen
* @date 2017.08.30
*/
public class InterruptTest {

public static void main(String[] args) {
int pause = 2000;
testBasic();
sleep(pause);
testSleep();
sleep(pause);
testWait();
sleep(pause);
testJoin();
}

// 基本的中断方法调用
public static void testBasic() {
System.out.println("InterruptTest.testBasic()");
Thread t = Thread.currentThread();
String name = t.getName();
System.out.println(name + " " + t.isInterrupted()); // false
// 调用interrupt，设置中断状态(interrupt status)为true
t.interrupt();
System.out.println(name + " " + t.isInterrupted()); // true

// Thread.interrupted()，返回当前线程的中断状态，同时清除设置的中断状态
System.out.println(name + " " + Thread.interrupted()); // true
System.out.println(name + " " + t.isInterrupted()); // false
}

// 中断sleep状态的线程
public static void testSleep() {
System.out.println("InterruptTest.testSleep()");
SleepThread t = new SleepThread("SleepThread");
t.start();
try {
Thread.sleep(7 * 1000);
} catch (InterruptedException e) {
// ignore
}

// 调用interrupt，SleepThread从sleep状态抛出InterruptedException，处理后结束线程
t.interrupt();
}

static class SleepThread extends Thread {

public SleepThread(String name) {
super(name);
}

@Override
public void run() {
while (true) {
try {
System.out.printf("%s loop...", getName()).println();
Thread.sleep(3 * 1000);
} catch (InterruptedException e) {
// 收到中断后退出循环，结束线程；某些时候需要将中断状态继续向上层暴露时，则需要再次调用interrupt进行中断。
// Thread.currentThread().interrupt();
System.out.printf("%s 中断，退出loop...", getName()).println();
break;
}
}
System.out.printf("%s 结束", getName()).println();
}
}

// 中断wait状态的线程
public static void testWait() {
System.out.println("InterruptTest.testWait()");
WaitThread t = new WaitThread("WaitThread");
t.start();
try {
Thread.sleep(5 * 1000);
} catch (InterruptedException e) {
// ignore
}
t.interrupt();
// System.out.println();
}

static class WaitThread extends Thread {
private Object waitObject = new Object();

public WaitThread(String name) {
super(name);
}

@Override
public void run() {
synchronized (waitObject) {
try {
System.out.printf("%s 进入wait阻塞...", getName()).println();
waitObject.wait();
System.out.printf("%s 结束wait阻塞...", getName()).println();
} catch (InterruptedException e) {
System.out.printf("%s Interrupted...", getName()).println();
}
System.out.printf("%s Business after wait...", getName()).println();
}
System.out.printf("%s 结束", getName()).println();
}
}

public static void testJoin() {
System.out.println("InterruptTest.testJoin()");
JoinThread jt = new JoinThread("JoinThread", Thread.currentThread());
jt.start();

try {
jt.join();
} catch (InterruptedException e) {
// TODO Auto-generated catch block
System.out.printf("%s 被中断，从join中返回...", Thread.currentThread().getName()).println();
}
}

static class JoinThread extends Thread {
// 调用jointhread.join的线程，需要在这里对其进行中断
private Thread callJoinThread;

public JoinThread(String name, Thread callJoinThread) {
super(name);
this.callJoinThread = callJoinThread;
}

@Override
public void run() {
int max = 1000000;
for (int i = 0; i < max; i++) {
//
}
System.out.printf("%s step1...", getName()).println();
callJoinThread.interrupt(); // 中断阻塞在join的线程
for (int i = 0; i < max; i++) {
//
}
System.out.printf("%s step2...", getName()).println();
System.out.printf("%s 结束", getName()).println();
}
}

private static void sleep(long millis) {
try {
Thread.sleep(millis);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}

运行结果如下：

InterruptTest.testBasic()
main false
main true
main true
main false
InterruptTest.testSleep()
SleepThread loop...
SleepThread loop...
SleepThread loop...
SleepThread 中断，退出loop...
SleepThread 结束
InterruptTest.testWait()
WaitThread 进入wait阻塞...
WaitThread Interrupted...
WaitThread Business after wait...
WaitThread 结束
InterruptTest.testJoin()
JoinThread step1...
main 被中断，从join中返回...
JoinThread step2...
JoinThread 结束

Nio TimeServer示例，演示中断作为服务线程退出条件

使用NIO创建一个TimeServer，启动Server线程，监听某个固定端口，接收用户连接请求，向用户输出当前时间，并关闭连接；main线程接受键盘输入，按q退出程序。

注意事项：

while循环条件可以只有running标志，可以只有!Thread.currentThread().isInterrupted()，也可以两个都加上，这个取决于自己喜好；个人一般只写running标志；

close方法通过发送中断或者调用selector.wakeUp让selector.select方法立即返回；

本质上，本例压根不需要使用中断，因为线程没有调用wait、join、sleep这几种方法中的任意一种，所以不需要使用中断来将线程从阻塞中唤醒；

使用中断照样是可以达成目标的；

package org.alive.learn.thread;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.StandardSocketOptions;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.Iterator;
import java.util.Set;

/**
* <p>
* 简单TimeServer，测试方法：telnet localhost 7890
*
* @author myumen
* @date 2017.08.30
*
*/
public class NioTimeServer extends Thread {

private Selector selector = null;

private ServerSocketChannel ssc = null;

private int port = 7890;

/** running标志 */
// private volatile boolean running = true;

/** 日期格式 */
private DateTimeFormatter fmt = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");

public NioTimeServer() {
}

public NioTimeServer(int port) {
this.port = port;
}

public int getPort() {
return port;
}

protected void init() {
try {
selector = Selector.open();

// 初始化服务端套接字通道ServerSocketChannel
ssc = ServerSocketChannel.open();
InetSocketAddress address = new InetSocketAddress(this.port);

ssc.setOption(StandardSocketOptions.SO_REUSEADDR, true);
ssc.bind(address);

// 配置非阻塞方式
ssc.configureBlocking(false);

// 向selector注册该channel的连接事件
ssc.register(selector, SelectionKey.OP_ACCEPT);

this.setName("Server-" + port);
} catch (IOException e) {
e.printStackTrace();
}
}

@Override
public void run() {
init();

try {
// 如果while循环使用running，那么close方法里可以增加selector.wakeUp方法调用，使selector.select方法立即返回
// 这样子，就完全不需要使用中断
// while (running) {

// while循环使用running和线程的中断状态判读是否应该结束
// while (running && !Thread.currentThread().isInterrupted()) {

// while循环条件只使用中断状态
while (!Thread.currentThread().isInterrupted()) {
// 以下三种情况，selector.select会立即返回：
// (1) 至少一个channel被选中；
// (2) 调用selector.wakeUp；
// (3) select所在线程收到中断请求；
int num = selector.select(2000);
if (num == 0) {
continue;
}

Set<SelectionKey> st = selector.selectedKeys();
Iterator<SelectionKey> it = st.iterator();

while (it.hasNext()) {
// SelectionKey key = (SelectionKey) it.next();
accept(it.next());
}
st.clear();
}
} catch (IOException e) {
e.printStackTrace();
}

System.out.println("Server线程退出");
}

private void accept(SelectionKey key) throws IOException {
if (key.isAcceptable()) {
SocketChannel sc = ((ServerSocketChannel) key.channel()).accept();
sc.configureBlocking(false);
// sc.setOption(StandardSocketOptions.SO_KEEPALIVE, true);
sc.setOption(StandardSocketOptions.TCP_NODELAY, true);
sc.setOption(StandardSocketOptions.SO_REUSEADDR, true);
sc.register(selector, SelectionKey.OP_WRITE); // 只注册写事件，不注册读事件
System.out.println("new connection : " + sc.socket().getRemoteSocketAddress());
}

if (key.isValid() && key.isWritable()) {
SocketChannel sc = (SocketChannel) key.channel();
String msg = "Now: " + LocalDateTime.now().format(fmt);
System.out.println("Send msg: " + msg);
sc.write(ByteBuffer.wrap(msg.getBytes()));
sc.close();
}
}

public static void main(String[] args) throws IOException {
// 启动Server线程
NioTimeServer server = new NioTimeServer();
server.setName("Server-" + server.getPort());
server.start();

// main线程继续接受键盘输入，按q退出程序
waitToQuit(server);
}

public void close() {
// while条件只使用running标志
// this.running = false;
// this.selector.wakeup();

// while条件使用running和中断
// this.running = false;
// this.interrupt();

// while条件使用中断
this.interrupt(); // 这里调用interrupt发送中断

try {
this.selector.close();
this.ssc.close();
} catch (IOException e) {
e.printStackTrace();
}
}

private static void waitToQuit(NioTimeServer server) throws IOException {
System.out.printf("Server is running on port %s, press q to quit.", server.getPort()).println();
boolean input = true;
while (input) {
int b = System.in.read();
switch (b) {
case 'q':
System.out.println("NioTimeServer关闭...");
server.close();
input = false;
break;
case '\r':
case '\n':
break;
default:
System.out.println("q -- quit.");
break;
}
}
}
}

运行程序，输入q，程序退出。

参考资料

The Java™ Tutorials - Interrupt