Thread.interrupt()的理解

Thread.interrupt()到底做了啥？ 作者：tracy_668  链接：https://www.jianshu.com/p/e2b22c6bcd22

Thread.interrupt()的理解   作者：zoterap   链接：https://www.jianshu.com/p/84df313ff1d2

目标

一个线程不应该由其他线程来强制中断或停止，而是应该由线程自己自行停止。
Thread.interrupt 的作用其实也不是中断线程，而是「通知线程应该中断」

代码

package com.zoterap.javabasic.current;

import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;

import static java.lang.String.format;
import static java.util.concurrent.TimeUnit.SECONDS;

public class ThreadInterruptDemo {

    public static String getCurrentTime() {
        return LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss SSS"));
    }
    
    public static void main(String[] args) {
        Thread t = new Thread(() -> {
            for (int i = 0; i < 10; i++) {
                System.out.println(format("current i[%d], dateTime[%s]", i, getCurrentTime()));

                try {
                    SECONDS.sleep(1);
                } catch (InterruptedException e) {
                    System.out.println("EXCEPTION:" + e.getMessage());
                    Thread.currentThread().interrupt();
                }

            }
        });

        /**
         * 开启线程
         */
        t.start();

        /**
         * 主线程休眠5秒
         */
        try {
            SECONDS.sleep(5);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        /**
         * 打印线程，尝试中断
         */
        t.interrupt();

    }

}

 

结果

current i[0], dateTime[2019-02-13 21:24:33 988]
current i[1], dateTime[2019-02-13 21:24:35 011]
current i[2], dateTime[2019-02-13 21:24:36 015]
current i[3], dateTime[2019-02-13 21:24:37 020]
current i[4], dateTime[2019-02-13 21:24:38 022]
EXCEPTION:sleep interrupted
current i[5], dateTime[2019-02-13 21:24:38 963]
EXCEPTION:sleep interrupted
current i[6], dateTime[2019-02-13 21:24:38 964]
EXCEPTION:sleep interrupted
current i[7], dateTime[2019-02-13 21:24:38 964]
EXCEPTION:sleep interrupted
current i[8], dateTime[2019-02-13 21:24:38 964]
EXCEPTION:sleep interrupted
current i[9], dateTime[2019-02-13 21:24:38 965]
EXCEPTION:sleep interrupted

Thread.interrupt()到底做了啥？

   在以前可以通过thread.stop()可以让一个线程去停止另一个线程，但这种方法太暴力，突然间停止其他线程会导致被停止的线程无法完成一些清理工作，所以stop()已经被抛弃了。

Java线程的终止操作最初是直接暴露给用户的，java.lang.Thread类提供了stop()方法，允许用户暴力的终止一个线程并退出临界区（释放所有锁，并在当前调用栈抛出ThreadDeath Exception）。 同样的，Thread.suspend()和Thread.resume()方法允许用户灵活的暂停和恢复线程。然而这些看似简便的API在JDK1.2就被deprecate掉了，原因是stop()方法本质上是不安全的，它会强制释放掉线程持有的锁，这样临界区的数据中间状态就会遗留出来，从而造成不可预知的后果。

当然Java线程不可能没有办法终止，在Java程序中，唯一的也是最好的办法就是让线程从run()方法返回。更具体来说，有以下几种情况：

• 对于runnable的线程，利用一个变量做标记位，定期检查

private volatile boolean flag = true;
public void stop() {
    flag = false;
}
public void run() {
    while (flag) {
        //do something...
    }
}

 

• 对于非runnable的线程，应该采取中断的方式退出阻塞，并处理捕获的中断异常

1. 对于大部分阻塞线程的方法，使用Thread.interrupt()，可以立刻退出等待，抛出InterruptedException. 这些方法包括Object.wait(), Thread.join()，Thread.sleep()，以及各种AQS衍生类：Lock.lockInterruptibly()等任何显示声明throws InterruptedException的方法。

private volatile Thread thread;
public void stop() {
    thread.interrupt();
}
public void run() {
    thread = Thread.currentThread();
    while (flag) {
        try {
            Thread.sleep(interval);
        } catch (InterruptedException e){
            //current thread was interrupted
            return;
        }
    }
}

 

2. 被阻塞的nio Channel也会响应interrupt()，抛出ClosedByInterruptException，相应nio通道需要实现java.nio.channels.InterruptibleChannel接口

private volatile Thread thread; public void stop() { thread.interrupt(); } public void run() { thread = Thread.currentThread(); BufferedReader in = new BufferedReader(new InputStreamReader(Channels.newInputStream( new FileInputStream(FileDescriptor.in).getChannel()))); while (flag) { try { String line = null; while ((line = in.readLine()) != null) { System.out.println("Read line:'"+line+"'"); } } catch (ClosedByInterruptException e) { //current channel was interrupted return; } catch (IOException e) { e.printStackTrace(); } } } 

如果使用的是传统IO（非Channel，如ServerSocket.accept），所在线程被interrupt时不会抛出ClosedByInterruptException。但可以使用流的close方法实现退出阻塞。

3. 还有一些阻塞方法不会响应interrupt，如等待进入synchronized段、Lock.lock()。他们不能被动的退出阻塞状态。

而Thread.interrupt的作用其实不是中断线程，而是通知线程应该中断了，给这个线程发一个信号，告诉它，它应该结束了，设置一个停止标志， 具体到底中断还是运行，应该由被通知的线程自己处理。具体来说,对一个线程，调用interrupt()时：

• 如果一个线程处于了阻塞状态（如线程调用了thread.sleep、thread.join、thread.wait、1.5中的condition.await、以及可中断的通道上的 I/O 操作方法后可进入阻塞状态），则在线程在检查中断标示时如果发现中断标示为true，则会在这些阻塞方法（sleep、join、wait、1.5中的condition.await及可中断的通道上的 I/O 操作方法）调用处抛出InterruptedException异常，
• 如果线程处于正常活动状态，那么会将该线程中的中断标志设置为true，仅此而已， 被设置中断标志的线程将继续正常执行，不受影响。
  interrupt()并不能真正中断线程，需要被调用的线程自己进行配合才行。一个线程如果有被中断的需求，那么就可以这样做：
• 在正常运行任务时，经常检查本线程的中断标志位，如果被设置了中断标志，就自行停止线程
• 在调用阻塞方法时正常处理InterruptException异常，

hread thread = new Thread(() -> {
    while (!Thread.interrupted()) {
        // do more work.
    }
});
thread.start();

// 一段时间以后
thread.interrupt();

 

上面代码中thread.interrupted()清除标志位是为了下次继续监测标志位，不然会对被中断线程的下次运行有影响。通过interrupt()和.interrupted()方法两者的配合可以实现正常去停止一个线程，线程A通过调用线程B的interrupt方法通知线程B让它结束线程，在线程B的run方法内部，通过循环检查.interrupted()方法是否为真来接收线程A的信号，如果为真就可以抛出一个异常，在catch中完成一些清理工作，然后结束线程。Thread.interrupted()会清除标志位，并不是代表线程又恢复了，可以理解为仅仅是代表它已经响应完了这个中断信号然后又重新置为可以再次接收信号的状态。从始至终，理解一个关键点，interrupt()方法仅仅是改变一个标志位的值而已，和线程的状态并没有必然的联系。

源码分析

Thread.interrupt()方法设计的目的，是提示一个线程应该终止，但不强制该线程终止。程序员可以来决定如何响应这个终止提示。直接上源码：

//Class java.lang.Thread
    public void interrupt() {
        if (this != Thread.currentThread())
            checkAccess();
        synchronized (blockerLock) {
            Interruptible b = blocker;   //中断触发器
            if (b != null) {
                interrupt0();           
                b.interrupt(this);   //触发回调接口
                return;
            }
        }
        interrupt0();
    }

 

校验权限
如果不是当前线程自我中断，会先做一次权限检查。如果被中断的线程属于系统线程组（即JVM线程），checkAccess()方法会使用系统的System.getSecurityManager()来判断权限。由于Java默认没有开启安全策略，此方法其实会跳过安全检查。

触发中断回调接口
如果线程的中断触发器blocker不为null，则触发中断触发回调接口Interruptible。那么这个触发器blocker是什么时候被设置的呢？

上面提到，如果一个nio通道实现了InterruptibleChannel接口，就可以响应interrupt()中断，其原理就在InterruptibleChannel接口的抽象实现类AbstractInterruptibleChannel的方法begin()中：

//Class java.nio.channels.spi.AbstractInterruptibleChannel protected final void begin() { if (interruptor == null) { interruptor = new Interruptible() { public void interrupt(Thread target) { synchronized (closeLock) { if (!open) return; open = false; interrupted = target; try {//关闭io通道 AbstractInterruptibleChannel.this.implCloseChannel(); } catch (IOException x) { } } }}; } blockedOn(interruptor);//将interruptor设置为当前线程的blocker Thread me = Thread.currentThread(); if (me.isInterrupted()) interruptor.interrupt(me); } protected final void end(boolean completed) throws AsynchronousCloseException { blockedOn(null); Thread interrupted = this.interrupted; if (interrupted != null && interrupted == Thread.currentThread()) { interrupted = null; throw new ClosedByInterruptException(); } if (!completed && !open) throw new AsynchronousCloseException(); } //Class java.nio.channels.Channels.ReadableByteChannelImpl public int read(ByteBuffer dst) throws IOException { ...... try { begin(); bytesRead = in.read(buf, 0, bytesToRead); finally { end(bytesRead > 0); } ...... } 

以上述代码为例，nio通道ReadableByteChannel每次执行阻塞方法read()前，都会执行begin()，把Interruptible回调接口注册到当前线程上，以实现能够响应其他线程的中断。当线程收到中断时，Thread.interrupt()触发回调接口，在回调接口Interruptible中关闭io通道并返回，最后在finally块中执行end()，end()方法会检查中断标记，抛出ClosedByInterruptException。

interrupt0()

 无论是否设置了中断触发回调blocker，都会执行这个关键的native方法interrupt0()： 

private native void interrupt0();

 

以openJDK7的Hotspot虚拟机为例，先找到native方法映射

//jdk\src\share\native\java\lang\Thread.c static JNINativeMethod methods[] = { {"start0", "()V", (void *)&JVM_StartThread}, {"stop0", "(" OBJ ")V", (void *)&JVM_StopThread}, {"isAlive", "()Z", (void *)&JVM_IsThreadAlive}, {"suspend0", "()V", (void *)&JVM_SuspendThread}, {"resume0", "()V", (void *)&JVM_ResumeThread}, {"setPriority0", "(I)V", (void *)&JVM_SetThreadPriority}, {"yield", "()V", (void *)&JVM_Yield}, {"sleep", "(J)V", (void *)&JVM_Sleep}, {"currentThread", "()" THD, (void *)&JVM_CurrentThread}, {"countStackFrames", "()I", (void *)&JVM_CountStackFrames}, {"interrupt0", "()V", (void *)&JVM_Interrupt}, {"isInterrupted", "(Z)Z", (void *)&JVM_IsInterrupted}, {"holdsLock", "(" OBJ ")Z", (void *)&JVM_HoldsLock}, {"getThreads", "()[" THD, (void *)&JVM_GetAllThreads}, {"dumpThreads", "([" THD ")[[" STE, (void *)&JVM_DumpThreads}, }; 

可以找到interrupt0对应JVM_Interrupt这个函数，继续找到实现代码

//hotspot\src\share\vm\prims\jvm.cpp JVM_ENTRY(void, JVM_Interrupt(JNIEnv* env, jobject jthread)) JVMWrapper("JVM_Interrupt"); // Ensure that the C++ Thread and OSThread structures aren't freed before we operate oop java_thread = JNIHandles::resolve_non_null(jthread); MutexLockerEx ml(thread->threadObj() == java_thread ? NULL : Threads_lock); // We need to re-resolve the java_thread, since a GC might have happened during the // acquire of the lock JavaThread* thr = java_lang_Thread::thread(JNIHandles::resolve_non_null(jthread)); if (thr != NULL) { Thread::interrupt(thr); } JVM_END 

可以看到这是一个JNI方法，JVM_ENTRY是JNI调用的宏。关键函数Thread::interrupt(thr)，继续跟踪：

//hotspot\src\share\vm\runtime\thread.cpp void Thread::interrupt(Thread* thread) { trace("interrupt", thread); debug_only(check_for_dangling_thread_pointer(thread);) os::interrupt(thread); } 

关键函数os::interrupt，os此时分为linux、solaris和windows，以linux为例继续跟踪：

//hotspot\src\os\linux\vm\os_linux.cpp void os::interrupt(Thread* thread) { assert(Thread::current() == thread || Threads_lock->owned_by_self(), "possibility of dangling Thread pointer"); OSThread* osthread = thread->osthread(); if (!osthread->interrupted()) { osthread->set_interrupted(true); // More than one thread can get here with the same value of osthread, // resulting in multiple notifications. We do, however, want the store // to interrupted() to be visible to other threads before we execute unpark(). OrderAccess::fence(); ParkEvent * const slp = thread->_SleepEvent ; if (slp != NULL) slp->unpark() ; } // For JSR166. Unpark even if interrupt status already was set if (thread->is_Java_thread()) ((JavaThread*)thread)->parker()->unpark(); ParkEvent * ev = thread->_ParkEvent ; if (ev != NULL) ev->unpark() ; } 

每一个Java线程都与一个osthread一一对应，如果相应的os线程没有被中断，则会设置osthread的interrupt标志位为true（对应一个volatile int），并唤醒线程的SleepEvent。随后唤醒线程的parker和ParkEvent。

简而言之，interrupt操作会对三种事件进行unpark唤醒，分别是thread->_SleepEvent、thread->parker()和thread->_ParkEvent，这些变量的具体声明如下：

//hotspot\src\share\vm\runtime\thread.cpp public: ParkEvent * _ParkEvent ; // for synchronized() ParkEvent * _SleepEvent ; // for Thread.sleep // JSR166 per-thread parker private: Parker* _parker; public: Parker* parker() { return _parker; } 

唤醒ParkEvent
Thread类中包含了两种作用不同的ParkEvent，_ParkEvent变量用于synchronized同步块和Object.wait()，_SleepEvent变量用于Thread.sleep()，ParkEvent类的声明如下：

class ParkEvent : public os::PlatformEvent { ... //略，直接看父类 } //hotspot\src\os\linux\vm\os_linux.cpp class PlatformEvent : public CHeapObj { private: pthread_mutex_t _mutex [1] ; pthread_cond_t _cond [1] ; ... public: PlatformEvent() { int status; status = pthread_cond_init (_cond, NULL); assert_status(status == 0, status, "cond_init"); status = pthread_mutex_init (_mutex, NULL); assert_status(status == 0, status, "mutex_init"); _Event = 0 ; _nParked = 0 ; _Assoc = NULL ; } void park () ; void unpark () ; int park (jlong millis) ; ... } ; 

ParkEvent包含了一把mutex互斥锁和一个cond条件变量，并在构造函数中进行了初始化，线程的阻塞和唤醒（park和unpark）就是通过他们实现的：

• PlatformEvent::park() 方法会调用库函数pthread_cond_wait(_cond, _mutex)实现线程等待。 synchronized块的进入和Object.wait()的线程等待都是通过PlatformEvent::park()方法实现, (Thread.join()是使用的Object.wait()实现的)

• PlatformEvent::park(jlong millis) 方法会调用库函数pthread_cond_timedwait(_cond, _mutex, _abstime)实现计时条件等待,Thread.sleep(millis)就是通过PlatformEvent::park(jlong millis)实现

• PlatformEvent::unpark() 方法会调用库函数pthread_cond_signal (_cond)唤醒上述等待的条件变量。Thread.interrupt()就会触发其子类SleepEvent和ParkEvent的unpark()方法。synchronized块的退出也会触发unpark()。其所在对象ObjectMonitor维护了ParkEvent数组作为唤醒队列，synchronized同步块退出时，会触发ParkEvent::unpark()方法来唤醒等待进入同步块的线程，或等待在Object.wait()的线程。

上述Thread类的两个ParkEvent成员变量：_ParkEvent和_SleepEvent，都会在Thread.interrupt()时触发unpark()动作。

对于_ParkEvent来说，它可以代表一个synchronized等待进入同步块的时事件，也可以代表一个Object.wait()等待条件变量的事件。不同的是，如果是synchronized等待事件，被唤醒后会尝试获取锁，如果失败则会通过循环继续park()等待，因此synchronized等待实际上是不会被interrupt()中断的；如果是Object.wait()事件，则会通过标记为判断出是否是被notify()唤醒的，如果不是则抛出InterruptedException实现中断。

对于_SleepEvent相对简单一些，它只代表线程sleep动作，可能是java.lang.Thread.sleep()，也可能是jvm内部调用的线程os::sleep()。如果是java.lang.Thread.sleep()，则会通过线程的is_interrupted标记位来判断抛出InterruptedException。

唤醒Parker
除了唤醒SleepEvent和ParkEvent，Thread.interrupt()还会调用thread->parker()->unpark()来唤醒Thread的parker变量。Parker类与上面的ParkEvent类很相似，都持有一把mutex互斥锁和一个cond条件变量。具体代码见：

class Parker : public os::PlatformParker { public: // For simplicity of interface with Java, all forms of park (indefinite, // relative, and absolute) are multiplexed into one call. void park(bool isAbsolute, jlong time); void unpark(); ... //略，直接看父类 } //hotspot\src\os\linux\vm\os_linux.hpp class PlatformParker : public CHeapObj { protected: pthread_mutex_t _mutex [1] ; pthread_cond_t _cond [1] ; public: PlatformParker() { int status; status = pthread_cond_init (_cond, NULL); assert_status(status == 0, status, "cond_init"); status = pthread_mutex_init (_mutex, NULL); assert_status(status == 0, status, "mutex_init"); } } 

与ParkEvent一样，Parker使用着自己的锁和park()/unpark()方法。

• Parker::park(bool isAbsolute, jlong time) 方法会调用库函数pthread_cond_timedwait(_cond, _mutex, _abstime)实现计时条件等待，如果time=0则会直接使用pthread_cond_wait(_cond, _mutex)实现线程等待
• Parker::unpark() 方法会调用库函数pthread_cond_signal (_cond)唤醒上述等待的条件变量

如源码注释，Thread的_parker变量更具有通用性。凡是在Java代码里通过unsafe.park()/unpark()的调用都会对应到Thread的_parker变量去执行。而unsafe.park()/unpark()由java.util.concurrent.locks.LockSupport类调用，它支持了java.util.concurrent的各种锁、条件变量等线程同步操作。例如：ReentrantLock, CountDownLatch, ReentrantReadWriteLock, Semaphore, ThreadPoolExecutor, ConditionObject, ArrayBlockingQueue等。

线程被Thread.interrupt()中断时，并不意味着上述类的等待方法都会返回并抛出InterruptedException。尽管上述类最终等待在的unsafe.unpark()方法都会被唤醒，其是否继续执行park()等待仍取决于具体实现。例如，Lock.lock()方法不会响应中断，Lock.lockInterruptibly()方法则会响应中断并抛出异常，二者实现的区别就在于park()等待被唤醒时是否继续执行park()来等待锁

如何处理InterruptedException

• 方式1 如果自己很清楚当前线程被中断后的处理方式，则按自己的方式处理。 通常是做好善后工作，主动退出线程
• 方式2 直接在方法声明中throws InterruptedException，丢给上层处理。这种方式也很常见，将中断的处置权交给具体的业务来处理
• 方式3 重新设置中断标记位，Thread.currentThread().interrupt()，交给后续方法处理
  原因是底层抛出InterruptedException时会清除中断标记位，捕获到异常后如果不想处理，可以重新设置中断标记位

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

注意 请不要吞掉InterruptedException，可能会导致上层的调用方出现不可预料的结果

小结

终止一个Java线程最好的方式，就是让run()方法主动退出。因为强制的让一个线程被动的退出是很不安全的，内部的数据不一致会对程序造成不可预知的后果。
为了能够通知一个线程需要被终止，Java提供了Thread.interrupt()方法，该方法会设置线程中断的标记位，并唤醒可中断的阻塞方法，包括Thread.sleep()，Object.wait()，nio通道的IO等待，以及LockSupport.park()。识别一个方法是否会被中断，只需要看其声明中是否会throws InterruptedException或ClosedByInterruptException。

每个Java线程都会对应一个osthread，它持有了三种条件变量，分别用于Thread.sleep()，Object.wait()和unsafe.park()。Thread.interrupt()会依次唤醒三个条件变量，以达到中断的目的。线程的同步与唤醒最终都使用了pthread_cond_wait和pthread_cond_signal这些pthread库函数。