线程生命周期 & 中断机制

发表于 | 分类于 | 阅读次数:
字数统计: 3,520 | 阅读时长 ≈ 15

这一篇主要涉及线程的生命周期和中断机制，对Thread类下的interrupt()源码进行了解析。

线程生命周期

线程的生命周期包括一些特有的状态，而Thread提供可以获取这些状态的方法，该方法如下：

1
public State getState()

返回的是枚举类型，枚举类结构如下：

1
2
3
4
5
6
7
8
public enum State() {
  NEW, // 还没调用start的线程状态；
  RUNNABLE, // 调用start执行run方法时的状态，包括了准备状态和运行状态；
  BLOCKED, // 阻塞状态；
  WAITING, // 等待状态；
  TIMED_WAITING, // 限时等待状态；
  TERMINATED; // 线程运行结束后的状态；
}

以上代码中的注释对这些状态进行了简要说明，它们彼此的转换关系可以用一张图来说明：

状态图

此图出自Java Thread States and Life Cycle。

在这里对这张图中的某些要点进行一次强调和说明：

Runnable状态可以分成Ready和Running，Runnable并不代表一定在执行线程，可能在等待操作系统分配时间片；

Timed Waiting在sleeptime elapsed后可以直接进入Runnable态，因为它可以拥有锁，这是它和Waiting的主要区别；

Waiting在被唤醒后，除非terminated，否则都会进入Blocked状态，等待monitor lock acquired；

接下来将解析线程的中断机制，最后将尝试在线程的不同状态下进行中断操作，配合源码分析出现的不同结果。

线程中断机制

run()方法正常执行到最后，当然是最好的结束线程生命的方式，而这里，将讨论线程的中断机制。(注：本篇不讨论一些过时的结束线程生命的方法)

Thread提供的interrupt()是线程中断机制最重要的方法，源代码如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
public void interrupt() {
    if (this != Thread.currentThread())
        checkAccess();
    synchronized (blockerLock) {
        Interruptible b = blocker;
        if (b != null) {
            interrupt0();           // Just to set the interrupt flag
            b.interrupt(this);
            return;
        }
    }
    interrupt0();
}

checkAccess()进行访问权限检查，blockerLock是普通Object对象，这里用来加锁保证线程安全。

interrupt()源码解析第一部分

本人将synchronized作用的区域作为interrupt()源码的第一部分，首先对这一部分进行解析。

Interruptible是一个普通接口：

1
2
3
public interface Interruptible {
    void interrupt(Thread var1);
}

在Thread类内部blocker赋值方法如下：

1
2
3
4
5
void blockedOn(Interruptible b) {
    synchronized (blockerLock) {
        blocker = b;
    }
}

在JDK源代码查找这个方法：

blockedOn

除了System类的内部可以指定Thread和Interruptible来调用blockedOn()以外，其它blockedOn()的调用都和AbstractInterruptibleChannel这个抽象类有关。

当然，必须说明的是，这些blockedOn()大都只是同名，并非调用的相同的方法，但即便如此，在具体实现中都封装了Thread的blockedOn()。下面简单梳理一下调用过程：

JavaLangAccess是一个接口，内部只声明了一个blockedOn()：

1
void blockedOn(Thread t, Interruptible b);

SharedSecrets拥有get和set JavaLangAccess实例的方法：

1
2
3
4
5
6
7
public static void setJavaLangAccess(JavaLangAccess jla) {
    javaLangAccess = jla;
}
public static JavaLangAccess getJavaLangAccess() {
    return javaLangAccess;
}

System内部有一个私有方法：

1
2
3
4
5
6
7
8
9
10
private static void setJavaLangAccess() {
    // Allow privileged classes outside of java.lang
    sun.misc.SharedSecrets.setJavaLangAccess(new sun.misc.JavaLangAccess(){
        ...
        public void blockedOn(Thread t, Interruptible b) {
            t.blockedOn(b);
        }
        ...
    };
}

正如我之前所说，最终在实现匿名接口的时候会在blockedOn(Thread t, Interruptible b)内调用Thread类的blockedOn()，那么System的这个方法又是什么时候调用的呢？

在java.lang包下的System 虽然是一个不能实现化的类，但是它有一个私有的静态方法initializeSystemClass()，通过类的静态初始化，registerNatives()会注册本地方法，虚拟机就会调用initializeSystemClass()方法完成类的初始化(非clinit)，该方法调用于线程初始化后。

AbstractInterruptibleChannel 也拥有一个静态blockedOn方法，源码如下：

1
2
3
4
static void blockedOn(Interruptible intr) {         // package-private
    sun.misc.SharedSecrets.getJavaLangAccess().blockedOn(Thread.currentThread(),
                                                         intr);
}

最终它调用的正是System类设置的实现匿名JavaLangAccess接口的 blockedOn()。

接下来重点关注AbstractInterruptibleChannel类的begin()、end(boolean completed)，看看给blockedOn传参执行的时机。

首先简单介绍一下AbstractInterruptibleChannel，源码注释有一句是这么说的：

1
* Base implementation class for interruptible channels.

直译过来就是说这个类是中断通道实现类的基础。

接下来有一段关键的注释：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
* <p> This class encapsulates the low-level machinery required to implement
* the asynchronous closing and interruption of channels.  A concrete channel
* class must invoke the {@link #begin begin} and {@link #end end} methods
* before and after, respectively, invoking an I/O operation that might block
* indefinitely.  In order to ensure that the {@link #end end} method is always
* invoked, these methods should be used within a
* <tt>try</tt>&nbsp;...&nbsp;<tt>finally</tt> block: <a name="be">
*
* <blockquote><pre>
* boolean completed = false;
* try {
*     begin();
*     completed = ...;    // Perform blocking I/O operation
*     return ...;         // Return result
* } finally {
*     end(completed);
* }</pre></blockquote>

意思是这个类封装了实现异步关闭和中断通道的low-level机制，一个具体的通道类必须在一个可能出现无限期IO阻塞的操作之前和之后分别调用begin和end方法，为确保调用了end方法，应将该方法放在finally语句块中。

先来看看begin方法，源码如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
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 {
                            AbstractInterruptibleChannel.this.implCloseChannel();
                        } catch (IOException x) { }
                    }
                }};
    }
    blockedOn(interruptor);
    Thread me = Thread.currentThread();
    if (me.isInterrupted())
        interruptor.interrupt(me);
}

前文已经提到过，Interruptible 是一个普通的接口，内部只有一个带有Thread形参的interrupt方法(和Thread的interrupt方法同名)，在begin中匿名实现了这个接口，而这个接口的实现内容都有什么呢？

一开始使用通道默认标志open设为true，代表通道是打开状态，Interruptible的interrupt方法中，如果检查open为false，那么直接return，为true时继续执行，继续执行过程中将true设置为false，因为继续执行意味着将在最后关闭通道(后续try内的代码起到了关闭通道的作用)，这样下次再进行相同的检测时，就处于关闭状态进而return了。

接下来是try catch语句，主要功能是对通道进行关闭处理。implCloseChannel是一个没有实现的抽象方法，看看注释是怎么解释的：

1
2
3
4
5
6
7
8
* <p> This method is invoked by the {@link #close close} method in order
* to perform the actual work of closing the channel.  This method is only
* invoked if the channel has not yet been closed, and it is never invoked
* more than once.
*
* <p> An implementation of this method must arrange for any other thread
* that is blocked in an I/O operation upon this channel to return
* immediately, either by throwing an exception or by returning normally.

大致意思是说，该方法由close方法调用，以关闭通道，仅当通道未关闭时调用，且永远只会调用一次；此方法的实现还必须运作此通道内阻塞在IO操作上的其它线程立即返回，或抛异常或正常返回。

后续的代码解释起来就不复杂了：blockedOn(interruptor)相当于将interruptor注册到当前的线程中，在调用begin方法的时候首先判断一次线程是否被中断(调用当前线程isInterrupted方法来判断)，如果为true，就会调用interruptor的interrupt方法。

我们再次回到Thread的interrupt方法，看下面这段代码：

1
2
3
4
5
6
7
8
synchronized (blockerLock) {
    Interruptible b = blocker;
    if (b != null) {
        interrupt0();           // Just to set the interrupt flag
        b.interrupt(this);
        return;
    }
}

interrupt0()在这段代码中Just to set the interrupt flag！而后面b.interrupt(this)相当于回调了Interruptible的方法，等同于begin()中最后代码的作用。

再来看end方法：

1
2
3
4
5
6
7
8
9
10
11
12
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();
}

completed用来表示IO操作是否完成，整个end主要用来置空和抛出相应异常。在方法begin()中设置了interrupted，而end中第一个if用来判断中断线程是否为当前线程，如果为true，将interrupted置空并抛出ClosedByInterruptException异常，如果通道关闭，IO操作未完成，抛出AsynchronousCloseException异常。

第一部分小结

Thread interrupt()的第一部分就解析完了，这一部分主要“服务于”实现了AbstractInterruptibleChannel的通道类。在解析源码之前的先验知识告诉我们，interrupt()通常是set flag，但在这一段代码中，我们可以发现，interrupt会主动回调interruptor的interrupt()来关闭通道，同时回调的方法也会更改字段open的状态，这个状态可以作为条件之一用于end()中判断是否抛出AsynchronousCloseException。

interrupt()源码解析第二部分

第二部分是对剩下的interrupt()进行简要解析。

目标代码如下：

1
private native void interrupt0();

这是一个native方法，去往链接：

http://hg.openjdk.java.net/jdk7u/jdk7u/jdk/file/tip/src/share/native/java/lang/Thread.c

查找interrupt0方法，得到：

1
{"interrupt0",       "()V",        (void *)&JVM_Interrupt}

这是JNINativeMethod结构体的一个实例，该结构体描述了注册信息，我在前文介绍System类时提到过本地方法注册(关于JNI注册本地方法这里就不展开了)。这里”interrupt0”为Java类中Native方法名，”()V”表示无参无返回值，”JVM_Interrupt”为对应的本地方法名，所以这里的注册信息表明了一个一一对应的映射关系。

那么本地方法实现到底在什么地方？

RednaxelaFX在这个链接进行了回答：

http://hllvm.group.iteye.com/group/topic/35385#post-236056

具体的实现代码在这里：

http://hg.openjdk.java.net/jdk7u/jdk7u/hotspot/file/tip/src/share/vm/prims/jvm.cpp

源码如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
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

定位关键方法：Thread::interrupt(thr)，此方法来自这里：

http://hg.openjdk.java.net/jdk7u/jdk7u/hotspot/file/tip/src/share/vm/runtime/thread.cpp

继续查找相关实现，跳转到798行：

1
2
3
4
5
void Thread::interrupt(Thread* thread) {
  trace("interrupt", thread);
  debug_only(check_for_dangling_thread_pointer(thread);)
  os::interrupt(thread);
}

定位关键方法：os::interrupt(thread)，这里os对应多个系统，所以有多个不同场景，具体对应哪些系统见下图：

os

这里不妨选取Linux进行分析，先摆出代码：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
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() ;
}

接下来将解读上述代码：

程序通过thread->osthread()来获取系统线程，Java线程基于操作系统原生线程来实现。

osthread->set_interrupted(true)用来设置中断标志。

通过OrderAccess::fence()这段代码创建内存屏障，其目的是让其它线程对我们执行unpark()之前保存的interrupted()值具有可见性。

slp->unpark()是中断Thread.sleep()的方法。

((JavaThread*)thread)->parker()->unpark()是中断LockSupport.park()的方法。

ev->unpark()是中断Object.wait()的方法。

关于ParkEvent和Parker的区别，HotSpot注释里有一个比较简要的概括：

The JVM uses 2 very similar constructs:

  • ParkEvent are used for Java-level “monitor” synchronization.
  • Parkers are used by JSR166-JUC park-unpark.

We`ll want to eventually merge these redundant facilities and use ParkEvent.

所以它们结构基本上是相似的，未来HotSpot可能将它们进行整合。不同的地方在于ParkEvent“服务”于monitor，monitor依赖底层实现，对象头Mark Word记录了一些相关信息，这里为了方便理解可以认为monitor被对象持有，所以ParkEvent需要同步对象。

unpark()：这里留个钩子，暂时没有继续进行解析，未来有需要再展开。

第二部分小结

通过第二部分的解析，了解了调用Thread类的interrupt()会对线程的哪些状态进行直接操作，对照本文起初的线程状态转换关系图，这些状态属于Timed Waiting状态以及Waiting状态，本文后续部分，将讨论interrupt()对线程不同状态的作用。

线程不同状态的中断响应

RUNNABLE状态

在这个状态下，interrupt()只会设置线程的中断标志，现在假设在目标线程运行的时候，对它进行interrupt()，然后将该线程睡眠，此时线程会响应中断吗？运行代码：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
public class ThreadInterruptTestDemo03{
    public static volatile boolean flag01 = true;
    public static void main(String[] args) throws InterruptedException {
        Thread t = new ThreadDemo();
        t.start();
        t.interrupt();
        Thread.sleep(2000);
        System.out.println("主线程睡眠结束！");
        flag01 = false;
    }
}
class ThreadDemo extends Thread {
    @Override
    public void run() {
        while (flag01) {
        }
        try {
            Thread.sleep(10000);
            System.out.println("正常睡眠结束！");
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

结果如下：

1
2
3
4
5
6
主线程睡眠结束！
java.lang.InterruptedException: sleep interrupted
	at java.lang.Thread.sleep(Native Method)
	at com.cg.ThreadDemo.run(ThreadInterruptTestDemo03.java:26)
Process finished with exit code 0

Timed Waiting / Waiting状态

此状态，本人在之前的篇幅中已经解析完毕。现在考虑这样一种情形，在一个线程的run()里，首先interrupt()一个sleep()，如果run()后续存在LockSupport.park()，那么线程会转为waiting状态吗？

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
public class ThreadInterruptTestDemo03 {
    public static volatile boolean flag01 = true;
    public static void main(String[] args) throws InterruptedException {
        Thread t = new ThreadDemo();
        t.start();
        t.interrupt();
        Thread.sleep(2000);
        System.out.println("主线程睡眠结束！");
        flag01 = false;
    }
}
class ThreadDemo extends Thread {
    @Override
    public void run() {
        while (flag01) {
        }
        try {
            Thread.sleep(10000);
            System.out.println("正常睡眠结束！");
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println(Thread.currentThread().isInterrupted());
        LockSupport.park();
    }
       // try {
       //     Thread.sleep(10000);
       //     System.out.println("又睡醒了！");
       // } catch (InterruptedException e) {
       //     e.printStackTrace();
       // }
}

该程序会运行结束。如果取消注释掉的代码，程序会timed waiting十秒然后结束。

BLOCKED

如果线程处在BLOCKED状态，interrupt()只会设置线程的中断标志。

线程状态中断响应小结

对照本文起初的线程状态转换关系图，可以知道，通常，线程调用sleep方法、wait方法、join方法以及park方法后会进入waiting / timed waiting状态，如果是await等JUC提供的方法呢？由于这些方法实现上都借助了park等方法，所以它们仍将进入此状态。

本篇小结

见各部分小结。

参考

相关源码的查找见此篇： http://hllvm.group.iteye.com/group/topic/35385#post-236056

HotSpot源码目录注释： http://hllvm.group.iteye.com/group/topic/26998#193368

这是一篇关于LockSupport的park和unpark的源码剖析：http://www.dayexie.com/detail1684592.html