专栏首页实战dockerJava的wait和notify学习三部曲之一:JVM源码分析

Java的wait和notify学习三部曲之一:JVM源码分析

综述

Java的wait()、notify()学习三部曲由三篇文章组成,内容分别是:一、通过阅读openjdk8的源码,分析和理解wait,notify在JVM中的具体执行过程;二、修改JVM源码,编译构建成新的JVM,把我们感兴趣的参数打印出来,结合具体代码检查和我们的理解是否一致;三、修改JVM源码,编译构建成新的JVM,按照我们的理解去修改关键参数,看能否达到预期效果;

现在,咱们一起开始既漫长又深入的wait、notify学习之旅吧!

wait()和notify()的通常用法

Java多线程开发中,我们常用到wait()和notify()方法来实现线程间的协作,简单的说步骤如下:

  1. A线程取得锁,执行wait(),释放锁;
  2. B线程取得锁,完成业务后执行notify(),再释放锁;
  3. B线程释放锁之后,A线程取得锁,继续执行wait()之后的代码;

关于synchronize修饰的代码块

通常,对于synchronize(lock){...}这样的代码块,编译后会生成monitorenter和monitorexit指令,线程执行到monitorenter指令时会尝试取得lock对应的monitor的所有权(CAS设置对象头),取得后即获取到锁,执行monitorexit指令时会释放monitor的所有权即释放锁;

一个完整的demo

为了深入学习wait()和notify(),先用完整的demo程序来模拟场景吧,以下是源码:

public class NotifyDemo {	
    private static void sleep(long sleepVal){	
        try{	
            Thread.sleep(sleepVal);	
        }catch(Exception e){	
            e.printStackTrace();	
        }	
    }	
    private static void log(String desc){	
        System.out.println(Thread.currentThread().getName() + " : " + desc);	
    }	
    Object lock = new Object();	
    public void startThreadA(){	
        new Thread(() -> {	
            synchronized (lock){	
                log("get lock");	
                startThreadB();	
                log("start wait");	
                try {	
                    lock.wait();	
                }catch(InterruptedException e){	
                    e.printStackTrace();	
                }	
                log("get lock after wait");	
                log("release lock");	
            }	
        }, "thread-A").start();	
    }	
    public void startThreadB(){	
        new Thread(()->{	
            synchronized (lock){	
                log("get lock");	
                startThreadC();	
                sleep(100);	
                log("start notify");	
                lock.notify();	
                log("release lock");	
            }	
        },"thread-B").start();	
    }	
    public void startThreadC(){	
        new Thread(() -> {	
            synchronized (lock){	
                log("get lock");	
                log("release lock");	
            }	
        }, "thread-C").start();	
    }	
    public static void main(String[] args){	
        new NotifyDemo().startThreadA();	
    }	
}

以上就是本次实战用到的demo,代码功能简述如下:

  1. 启动线程A,取得锁之后先启动线程B再执行wait()方法,释放锁并等待;
  2. 线程B启动之后会等待锁,A线程执行wait()之后,线程B取得锁,然后启动线程C,再执行notify唤醒线程A,最后退出synchronize代码块,释放锁;
  3. 线程C启动之后就一直在等待锁,这时候线程B还没有退出synchronize代码块,锁还在线程B手里;
  4. 线程A在线程B执行notify()之后就一直在等待锁,这时候线程B还没有退出synchronize代码块,锁还在线程B手里;
  5. 线程B退出synchronize代码块,释放锁之后,线程A和线程C竞争锁;

把上面的代码在Openjdk8下面执行,反复执行多次,都得到以下结果:

thread-A : get lock	
thread-A : start wait	
thread-B : get lock	
thread-C : c thread is start	
thread-B : start notify	
thread-B : release lock	
thread-A : after wait, acquire lock again	
thread-A : release lock	
thread-C : get lock	
thread-C : release lock

针对以上结果,问题来了:第一个问题:将以上代码反复执行多次,结果都是B释放锁之后A会先得到锁,这又是为什么呢?C为何不能先拿到锁呢?

第二个问题:线程C自开始就执行了monitorenter指令,它能得到锁是容易理解的,但是线程A呢?在wait()之后并没有没有monitorenter指令,那么它又是如何取得锁的呢?

wait()、notify()这些方法都是native方法,所以只有从JVM源码寻找答案了,本次阅读的是openjdk8的源码;

带上问题去看JVM源码

按照demo代码执行顺序,我整理了如下问题,带着这些问题去看JVM源码可以聚焦主线,不要被一些支线的次要的代码卡住(例如一些异常处理,监控和上报等):

  1. 线程A在wait()的时候做了什么?
  2. 线程C启动后,由于此时线程B持有锁,那么线程C此时在干啥?
  3. 线程B在notify()的时候做了什么?
  4. 线程B释放锁的时候做了什么?

源码中最重要的注释信息

在源码中有段注释堪称是整篇文章最重要的说明,请大家始终记住这段信息,处处都用得上:

ObjectWaiter对象存在于WaitSet、EntryList、cxq等集合中,或者正在这些集合中移动

原文如下:

请务必记住这三个集合:WaitSet、EntryList、cxq

好了,接下来看源码分析问题吧:

线程A在wait()的时候做了什么

打开hotspot/src/share/vm/runtime/objectMonitor.cpp,看ObjectMonitor::wait方法:

如上图所示,有两处代码值得我们注意:

  1. 绿框中将当前线程包装成ObjectWaiter对象,并且状态为TS_WAIT,这里对应的是jstack看到的线程状态WAITING;
  2. 红框中调用了AddWaiter方法,跟进去看下:

这个ObjectWaiter对象被放入了WaitSet中,WaitSet是个环形双向链表(circular doubly linked list)

回到ObjectMonitor::wait方法接着往下看,会发现关键代码如下图,当前线程通过park()方法开始挂起(suspend):

至此,我们把wait()方法要做的事情就理清了:

  1. 包装成ObjectWaiter对象,状态为TS_WAIT;
  2. ObjectWaiter对象被放入_WaitSet中;
  3. 当前线程挂起;

线程B持有锁的时候线程C在干啥

此时的线程C无法进入synchronized{}代码块,用jstack看应该是BLOCKED状态,如下图:

我们看看monitorenter指令对应的源码吧,位置:openjdk/hotspot/src/share/vm/interpreter/interpreterRuntime.cpp

IRT_ENTRY_NO_ASYNC(void, InterpreterRuntime::monitorenter(JavaThread* thread, BasicObjectLock* elem))	
#ifdef ASSERT	
  thread->last_frame().interpreter_frame_verify_monitor(elem);	
#endif	
  if (PrintBiasedLockingStatistics) {	
    Atomic::inc(BiasedLocking::slow_path_entry_count_addr());	
  }	
  Handle h_obj(thread, elem->obj());	
  assert(Universe::heap()->is_in_reserved_or_null(h_obj()),	
         "must be NULL or an object");	
  if (UseBiasedLocking) {	
    // Retry fast entry if bias is revoked to avoid unnecessary inflation	
    ObjectSynchronizer::fast_enter(h_obj, elem->lock(), true, CHECK);	
  } else {	
    ObjectSynchronizer::slow_enter(h_obj, elem->lock(), CHECK);	
  }	
  assert(Universe::heap()->is_in_reserved_or_null(elem->obj()),	
         "must be NULL or an object");	
#ifdef ASSERT	
  thread->last_frame().interpreter_frame_verify_monitor(elem);	
#endif	
IRT_END

上面的代码有个if (UseBiasedLocking)判断,是判断是否使用偏向锁的,本例中的锁显然已经不属于当前线程C了,所以我们还是直接看slowenter(hobj, elem->lock(), CHECK)方法吧;

打开openjdk/hotspot/src/share/vm/runtime/synchronizer.cpp:

void ObjectSynchronizer::slow_enter(Handle obj, BasicLock* lock, TRAPS) {	
  markOop mark = obj->mark();	
  assert(!mark->has_bias_pattern(), "should not see bias pattern here");	
  //是否处于无锁状态	
  if (mark->is_neutral()) {	
    // Anticipate successful CAS -- the ST of the displaced mark must	
    // be visible <= the ST performed by the CAS.	
    lock->set_displaced_header(mark);	
    //无锁状态就去竞争锁	
    if (mark == (markOop) Atomic::cmpxchg_ptr(lock, obj()->mark_addr(), mark)) {	
      TEVENT (slow_enter: release stacklock) ;	
      return ;	
    }	
    // Fall through to inflate() ...	
  } else	
  if (mark->has_locker() && THREAD->is_lock_owned((address)mark->locker())) {	
    //如果处于有锁状态,就检查是不是当前线程持有锁,如果是当前线程持有的,就return,然后就能执行同步代码块中的代码了	
    assert(lock != mark->locker(), "must not re-lock the same lock");	
    assert(lock != (BasicLock*)obj->mark(), "don't relock with same BasicLock");	
    lock->set_displaced_header(NULL);	
    return;	
  }	
#if 0	
  // The following optimization isn't particularly useful.	
  if (mark->has_monitor() && mark->monitor()->is_entered(THREAD)) {	
    lock->set_displaced_header (NULL) ;	
    return ;	
  }	
#endif	
  // The object header will never be displaced to this lock,	
  // so it does not matter what the value is, except that it	
  // must be non-zero to avoid looking like a re-entrant lock,	
  // and must not look locked either.	
  lock->set_displaced_header(markOopDesc::unused_mark());	
  //锁膨胀	
  ObjectSynchronizer::inflate(THREAD, obj())->enter(THREAD);	
}

线程C在上面代码中的执行顺序如下:

  1. 判断是否是无锁状态,如果是就通过Atomic::cmpxchg_ptr去竞争锁;
  2. 不是无锁状态,就检查当前锁是否是线程C持有;
  3. 不是线程C持有,调用inflate方法开始锁膨胀;

ObjectSynchronizer::inflate(THREAD, obj())->enter(THREAD);

来看看锁膨胀的源码:

如上图,锁膨胀的代码太长,我们这里只看关键代码吧:红框中,如果当前状态已经是重量级锁,就通过mark->monitor()方法取得ObjectMonitor指针再返回;绿框中,如果还不是重量级锁,就检查是否处于膨胀中状态(其他线程正在膨胀中),如果是膨胀中,就调用ReadStableMark方法进行等待,ReadStableMark方法执行完毕后再通过continue继续检查,ReadStableMark方法中还会调用os::NakedYield()释放CPU资源;

如果红框和绿框的条件都没有命中,目前已经是轻量级锁了(不是重量级锁并且不处于锁膨胀状态),可以开始膨胀了,如下图:

简单来说,锁膨胀就是通过CAS将监视器对象OjectMonitor的状态设置为INFLATING,如果CAS失败,就在此循环,再走前一副图中的的红框和绿框中的判断,如果CAS设置成功,会继续设置ObjectMonitor中的header、owner等字段,然后inflate方法返回监视器对象OjectMonitor;

看看之前slow_enter方法中,调用inflate方法的代码如下:

ObjectSynchronizer::inflate(THREAD, obj())->enter(THREAD);

所以inflate方法返回监视器对象OjectMonitor之后,会立刻执行OjectMonitor的enter方法,这个方法中开始竞争锁了,方法在openjdk/hotspot/src/share/vm/runtime/objectMonitor.cpp文件中:

如上图,红框中表示OjectMonitor的enter方法一进来就通过CAS将OjectMonitor的owner设置为当前线程,绿框中表示设置成功的逻辑,第一个if表示重入锁的逻辑,第二个if表示第一次设置owner成功,都意味着竞争锁成功,而我们的线程C显然是竞争失败的,会进入下图中的无线循环,反复调用EnterI方法:

进入EnterI方法看看:

如上图,首先构造一个ObjectWaiter对象node,后面的for(;;)代码块中来是一段非常巧妙的代码,同一时刻可能有多个线程都竞争锁失败走进这个EnterI方法,所以在这个for循环中,用CAS将cxq地址放入node的next,也就是把node放到cxq队列的首位,如果CAS失败,就表示其他线程把node放入到cxq的首位了,所以通过for循环再放一次,只要成功,此node就一定在最新的_cxq队列的首位。

接下来的代码又是一个无限循环,如下图:

从上图可以看出,进入循环后先调用TryLock方法竞争一次锁,如果成功了就退出循环,否则就调用Self->_ParkEvent->park方法使线程挂起,这里有自旋锁的逻辑,也就是park方法带了时间参数,就会在挂起一段时间后自动唤醒,如果不是自旋的条件,就一直挂起等待被其他条件唤醒,线程被唤醒后又会执行TryLock方法竞争一次锁,竞争不到继续这个for循环;

到这里我们已经把线程C在BLOCK的时候的逻辑理清楚了,小结如下:

  1. 偏向锁逻辑,未命中;
  2. 如果是无锁状态,就通过CAS去竞争锁,此处由于锁已经被线程B持有,所以不是无锁状态;
  3. 不是无锁状态,而且锁不是线程C持有,执行锁膨胀,构造OjectMonitor对象;
  4. 竞争锁,竞争失败就将线程加入_cxq队列的首位;
  5. 开始无限循环,竞争锁成功就退出循环,竞争失败线程挂起,等待被唤醒后继续竞争;

线程B在notify()的时候做了什么

接下来该线程B执行notify了,代码是objectMonitor.cpp的ObjectMonitor::notify方法:

如上图所示,首先是Policy的赋值,其次是调用DequeueWaiter()方法将WaitSet队列的第一个值取出并返回,还记得WaitSet么?所有wait的线程都被包装成ObjectWaiter对象然后放进来了;接下来对ObjectWaiter对象的处理方式,根据Policy的不同而不同:Policy == 0:放入EntryList队列的排头位置;Policy == 1:放入EntryList队列的末尾位置;Policy == 2:EntryList队列为空就放入EntryList,否则放入_cxq队列的排头位置;

如上图所示,请注意把ObjectWaiter的地址写到cxq变量的时候要用CAS操作,因为此时可能有其他线程正在竞争锁,竞争失败的时候会将自己包装成ObjectWaiter对象加入到cxq中;

这里的代码有一处疑问,期待着读着您的指教:如果_EntryList为空,就把ObjectWaiter放入ObjectWaiter中,为什么要这样做呢?

Policy == 3:放入cxq队列中,末尾位置;更新cxq变量的值的时候,同样要通过CAS注意并发问题;

这里有一段很巧妙的代码,现将cxq保存在Tail中,正常情况下将ObjectWaiter赋值给Tail->next就可以了,但是此时有可能其他线程正在cxq的尾部追加数据了,所以此时Tail对象对应的记录就不是最后一条了,那么它的next就非空了,一旦发生这种情况,就执行Tail = Tail->next,这样就获得了最新的cxq的尾部数据,如下图所示:

Policy等于其他值,立即唤醒ObjectWaiter对应的线程;

小结一下,线程B执行notify时候做的事情:

  1. 执行过wait的线程都在队列WaitSet中,此处从WaitSet中取出第一个;
  2. 根据Policy的不同,将这个线程放入EntryList或者cxq队列中的起始或末尾位置;

线程B释放锁的时候做了什么

接下来到了揭开问题的关键了,我们来看objectMonitor.cpp的ObjectMonitor::exit方法;

如上图,方法一进来先做一些合法性判断,接下来如红框所示,是偏向锁逻辑,偏向次数减一后直接返回,显然线程B在此处不会返回,而是继续往下执行;

根据QMode的不同,有不同的处理方式:

  1. QMode = 2,并且cxq非空:取cxq队列排头位置的ObjectWaiter对象,调用ExitEpilog方法,该方法会唤醒ObjectWaiter对象的线程,此处会立即返回,后面的代码不会执行了;
  2. QMode = 3,并且cxq非空:把cxq队列首元素放入_EntryList的尾部;
  3. QMode = 4,并且cxq非空:把cxq队列首元素放入_EntryList的头部;
  4. QMode = 0,不做什么,继续往下看;

只有QMode=2的时候会提前返回,等于0、3、4的时候都会继续往下执行:

如果EntryList的首元素非空,就取出来调用ExitEpilog方法,该方法会唤醒ObjectWaiter对象的线程,然后立即返回;如果EntryList的首元素为空,就取cxq的首元素,放入EntryList,然后再从_EntryList中取出来执行ExitEpilog方法,然后立即返回;

以上操作,均是执行过ExitEpilog方法然后立即返回,如果取出的元素为空,就执行循环继续取;

小结一下,线程B释放了锁之后,执行的操作如下:

  1. 偏向锁逻辑,此处未命中;
  2. 根据QMode的不同,将ObjectWaiter从cxq或者EntryList中取出后唤醒;
  3. 唤醒的元素会继续执行挂起前的代码,按照我们之前的分析,线程唤醒后,就会通过CAS去竞争锁,此时由于线程B已经释放了锁,那么此时应该能竞争成功;

到了现在已经将之前的几个问题搞清了,汇总起来看看:

  1. 线程A在wait() 后被加入了_WaitSet队列中;
  2. 线程C被线程B启动后竞争锁失败,被加入到_cxq队列的首位;
  3. 线程B在notify()时,从WaitSet中取出第一个,根据Policy的不同,将这个线程放入EntryList或者_cxq队列中的起始或末尾位置;
  4. 根据QMode的不同,将ObjectWaiter从cxq或者EntryList中取出后唤醒;;

所以,最初的问题已经清楚了,wait()的线程被唤醒后,会进入一个队列,然后JVM会根据Policy和QMode的不同对队列中的ObjectWaiter做不同的处理,被选中的ObjectWaiter会被唤醒,去竞争锁;

至此,源码分析已结束,但是因为我们不知道Policy和QMode参数到底是多少,所以还不能对之前的问题有个明确的结果,这些还是留在下一章来解答吧,下一章里我们去修改JVM源码,把参数都打印出来;

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

我来说两句

0 条评论
登录 后参与评论

相关文章

  • Java的wait()、notify()学习三部曲之一:JVM源码分析

    综述 Java的wait()、notify()学习三部曲由三篇文章组成,内容分别是: 一、通过阅读openjdk8的源码,分析和理解wait,notify在...

    程序员欣宸
  • Flink1.9.2源码编译和使用

    用于学习,在IDEA上开发的flink job,能直接在IDEA运行,如果运行时依赖的flink框架是我们自己编译构建的,就做到了从业务到框架都可以修改源码并验...

    程序员欣宸
  • Elasticsearch聚合学习之四:结果排序

    本文是《Elasticsearch聚合学习》系列的第四篇,在前面的实战中,聚合的结果以桶(bucket)为单位,放在JSON数组中返回,这些数据是没有排序的,今...

    程序员欣宸
  • Java的wait()、notify()学习三部曲之一:JVM源码分析

    综述 Java的wait()、notify()学习三部曲由三篇文章组成,内容分别是: 一、通过阅读openjdk8的源码,分析和理解wait,notify在...

    程序员欣宸
  • Java 线程通信之 wait/notify 机制

    Java 线程通信是将多个独立的线程个体进行关联处理,使得线程与线程之间能进行相互通信。比如线程 A 修改了对象的值,然后通知给线程 B,使线程 B 能够知道线...

    ytao
  • 深入浅出parallelStream

    今天小强带来java8的Stream,Stream是java8中新增加的一个特性,被java猿统称为流。

    程序员小强
  • Java的并发工具类汇总

    如果需要原子更新引用类型变量的话,为了保证线程安全,atomic也提供了相关的类:

    用户3467126
  • 史上最难的一道Java面试题:分析篇

    无意中了解到如下题目,觉得蛮好。 题目如下: ? 该程序的输出结果? 在java中,多线程的程序最难理解、调试,很多时候执行结果并不像我们想象的那样执行。所以在...

    CSDN技术头条
  • ReentrantLock源码解析

    在java编程中,经常需要多代码进行加锁来防止多线程可能引起的数据不一致。而锁的类型有公平锁和非公平锁。公平锁的意义就是按照顺序,而非公平锁则是相反的。也就是说...

    程序员_备忘录
  • 五种线程池的对比与使用

    通过源码可以看出底层调用的是ThreadPoolExecutor方法,传入一个同步的阻塞队列实现缓存。

    爱撸猫的杰

扫码关注云+社区

领取腾讯云代金券