精华内容
下载资源
问答
  • Java锁--Lock实现原理(底层实现)

    万次阅读 多人点赞 2016-09-01 20:11:56
    关于java lock的底层实现原理,讲的有点深,转载学习!Lock完全用Java写成,在java这个层面是无关JVM实现的。在java.util.concurrent.locks包中有很多Lock的实现类,常用的有ReentrantLock、ReadWriteLock(实现类...

    关于java lock的底层实现原理,讲的有点深,转载学习!

    Lock完全用Java写成,在java这个层面是无关JVM实现的。

    在java.util.concurrent.locks包中有很多Lock的实现类,常用的有ReentrantLock、ReadWriteLock(实现类ReentrantReadWriteLock),其实现都依赖java.util.concurrent.AbstractQueuedSynchronizer类,实现思路都大同小异,因此我们以ReentrantLock作为讲解切入点。

    ReentrantLock的调用过程

    经过观察ReentrantLock把所有Lock接口的操作都委派到一个Sync类上,该类继承了AbstractQueuedSynchronizer

    static abstract class Sync extends AbstractQueuedSynchronizer  
    

    Sync又有两个子类:

    final static class NonfairSync extends Sync  
    
    final static class FairSync extends Sync  
    

    显然是为了支持公平锁和非公平锁而定义,默认情况下为非公平锁。
    先理一下Reentrant.lock()方法的调用过程(默认非公平锁):
    这里写图片描述

    这些讨厌的Template模式导致很难直观的看到整个调用过程,其实通过上面调用过程及AbstractQueuedSynchronizer的注释可以发现,AbstractQueuedSynchronizer中抽象了绝大多数Lock的功能,而只把tryAcquire方法延迟到子类中实现。tryAcquire方法的语义在于用具体子类判断请求线程是否可以获得锁,无论成功与否AbstractQueuedSynchronizer都将处理后面的流程。

    锁实现(加锁)

    简单说来,AbstractQueuedSynchronizer会把所有的请求线程构成一个CLH队列,当一个线程执行完毕(lock.unlock())时会激活自己的后继节点,但正在执行的线程并不在队列中,而那些等待执行的线程全部处于阻塞状态,经过调查线程的显式阻塞是通过调用LockSupport.park()完成,而LockSupport.park()则调用sun.misc.Unsafe.park()本地方法,再进一步,HotSpot在Linux中中通过调用pthread_mutex_lock函数把线程交给系统内核进行阻塞。
    该队列如图:

    这里写图片描述

    与synchronized相同的是,这也是一个虚拟队列,不存在队列实例,仅存在节点之间的前后关系。令人疑惑的是为什么采用CLH队列呢?原生的CLH队列是用于自旋锁,但Doug Lea把其改造为阻塞锁。
    当有线程竞争锁时,该线程会首先尝试获得锁,这对于那些已经在队列中排队的线程来说显得不公平,这也是非公平锁的由来,与synchronized实现类似,这样会极大提高吞吐量。
    如果已经存在Running线程,则新的竞争线程会被追加到队尾,具体是采用基于CAS的Lock-Free算法,因为线程并发对Tail调用CAS可能会导致其他线程CAS失败,解决办法是循环CAS直至成功。AbstractQueuedSynchronizer的实现非常精巧,令人叹为观止,不入细节难以完全领会其精髓,下面详细说明实现过程:

    2.1 Sync.nonfairTryAcquire

    nonfairTryAcquire方法将是lock方法间接调用的第一个方法,每次请求锁时都会首先调用该方法。

    final boolean nonfairTryAcquire(int acquires) {  
        final Thread current = Thread.currentThread();  
        int c = getState();  
        if (c == 0) {  
            if (compareAndSetState(0, acquires)) {  
                setExclusiveOwnerThread(current);  
                return true;  
            }  
        }  
        else if (current == getExclusiveOwnerThread()) {  
            int nextc = c + acquires;  
            if (nextc < 0) // overflow  
                throw new Error("Maximum lock count exceeded");  
            setState(nextc);  
            return true;  
        }  
        return false;  
    }  
    

    该方法会首先判断当前状态,如果c==0说明没有线程正在竞争该锁,如果不c !=0 说明有线程正拥有了该锁。
    如果发现c==0,则通过CAS设置该状态值为acquires,acquires的初始调用值为1,每次线程重入该锁都会+1,每次unlock都会-1,但为0时释放锁。如果CAS设置成功,则可以预计其他任何线程调用CAS都不会再成功,也就认为当前线程得到了该锁,也作为Running线程,很显然这个Running线程并未进入等待队列。
    如果c !=0 但发现自己已经拥有锁,只是简单地++acquires,并修改status值,但因为没有竞争,所以通过setStatus修改,而非CAS,也就是说这段代码实现了偏向锁的功能,并且实现的非常漂亮。

    2.2 AbstractQueuedSynchronizer.addWaiter

    addWaiter方法负责把当前无法获得锁的线程包装为一个Node添加到队尾

    private Node addWaiter(Node mode) {  
        Node node = new Node(Thread.currentThread(), mode);  
        // Try the fast path of enq; backup to full enq on failure  
        Node pred = tail;  
        if (pred != null) {  
            node.prev = pred;  
            if (compareAndSetTail(pred, node)) {  
                pred.next = node;  
                return node;  
            }  
        }  
        enq(node);  
        return node;  
    }  
    

    其中参数mode是独占锁还是共享锁,默认为null,独占锁。追加到队尾的动作分两步:
    如果当前队尾已经存在(tail!=null),则使用CAS把当前线程更新为Tail
    如果当前Tail为null或则线程调用CAS设置队尾失败,则通过enq方法继续设置Tail
    下面是enq方法:

    private Node enq(final Node node) {  
        for (;;) {  
            Node t = tail;  
            if (t == null) { // Must initialize  
                Node h = new Node(); // Dummy header  
                h.next = node;  
                node.prev = h;  
                if (compareAndSetHead(h)) {  
                    tail = node;  
                    return h;  
                }  
            }  
            else {  
                node.prev = t;  
                if (compareAndSetTail(t, node)) {  
                    t.next = node;  
                    return t;  
                }  
            }  
        }  
    }  
    

    该方法就是循环调用CAS,即使有高并发的场景,无限循环将会最终成功把当前线程追加到队尾(或设置队头)。总而言之,addWaiter的目的就是通过CAS把当前线程追加到队尾,并返回包装后的Node实例。

    把线程要包装为Node对象的主要原因,除了用Node构造供虚拟队列外,还用Node包装了各种线程状态,这些状态被精心设计为一些数字值:

    • SIGNAL(-1) :线程的后继线程正/已被阻塞,当该线程release或cancel时要重新这个后继线程(unpark)
    • CANCELLED(1):因为超时或中断,该线程已经被取消
    • CONDITION(-2):表明该线程被处于条件队列,就是因为调用了Condition.await而被阻塞
    • PROPAGATE(-3):传播共享锁
    • 0:0代表无状态

    2.3 AbstractQueuedSynchronizer.acquireQueued

    acquireQueued的主要作用是把已经追加到队列的线程节点(addWaiter方法返回值)进行阻塞,但阻塞前又通过tryAccquire重试是否能获得锁,如果重试成功能则无需阻塞,直接返回

    final boolean acquireQueued(final Node node, int arg) {  
        try {  
            boolean interrupted = false;  
            for (;;) {  
                final Node p = node.predecessor();  
                if (p == head && tryAcquire(arg)) {  
                    setHead(node);  
                    p.next = null; // help GC  
                    return interrupted;  
                }  
                if (shouldParkAfterFailedAcquire(p, node) &&  
                    parkAndCheckInterrupt())  
                    interrupted = true;  
            }  
        } catch (RuntimeException ex) {  
            cancelAcquire(node);  
            throw ex;  
        }  
    }  
    

    仔细看看这个方法是个无限循环,感觉如果p == head && tryAcquire(arg)条件不满足循环将永远无法结束,当然不会出现死循环,奥秘在于第12行的parkAndCheckInterrupt会把当前线程挂起,从而阻塞住线程的调用栈。

    private final boolean parkAndCheckInterrupt() {  
        LockSupport.park(this);  
        return Thread.interrupted();  
    }  
    

    如前面所述,LockSupport.park最终把线程交给系统(Linux)内核进行阻塞。当然也不是马上把请求不到锁的线程进行阻塞,还要检查该线程的状态,比如如果该线程处于Cancel状态则没有必要,具体的检查在shouldParkAfterFailedAcquire中:

      private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {  
          int ws = pred.waitStatus;  
          if (ws == Node.SIGNAL)  
              /* 
               * This node has already set status asking a release 
               * to signal it, so it can safely park 
               */  
              return true;  
          if (ws > 0) {  
              /* 
               * Predecessor was cancelled. Skip over predecessors and 
               * indicate retry. 
               */  
       do {  
                node.prev = pred = pred.prev;  
        } while (pred.waitStatus > 0);  
                pred.next = node;  
        } else {  
              /* 
               * waitStatus must be 0 or PROPAGATE. Indicate that we 
               * need a signal, but don't park yet. Caller will need to 
               * retry to make sure it cannot acquire before parking.  
               */  
              compareAndSetWaitStatus(pred, ws, Node.SIGNAL);  
          }   
          return false;  
      }  
    

    检查原则在于:

    • 规则1:如果前继的节点状态为SIGNAL,表明当前节点需要unpark,则返回成功,此时acquireQueued方法的第12行(parkAndCheckInterrupt)将导致线程阻塞

    • 规则2:如果前继节点状态为CANCELLED(ws>0),说明前置节点已经被放弃,则回溯到一个非取消的前继节点,返回false,acquireQueued方法的无限循环将递归调用该方法,直至规则1返回true,导致线程阻塞

    • 规则3:如果前继节点状态为非SIGNAL、非CANCELLED,则设置前继的状态为SIGNAL,返回false后进入acquireQueued的无限循环,与规则2同

    总体看来,shouldParkAfterFailedAcquire就是靠前继节点判断当前线程是否应该被阻塞,如果前继节点处于CANCELLED状态,则顺便删除这些节点重新构造队列。
    至此,锁住线程的逻辑已经完成,下面讨论解锁的过程。

    解锁

    请求锁不成功的线程会被挂起在acquireQueued方法的第12行,12行以后的代码必须等线程被解锁锁才能执行,假如被阻塞的线程得到解锁,则执行第13行,即设置interrupted = true,之后又进入无限循环。

    从无限循环的代码可以看出,并不是得到解锁的线程一定能获得锁,必须在第6行中调用tryAccquire重新竞争,因为锁是非公平的,有可能被新加入的线程获得,从而导致刚被唤醒的线程再次被阻塞,这个细节充分体现了“非公平”的精髓。通过之后将要介绍的解锁机制会看到,第一个被解锁的线程就是Head,因此p == head的判断基本都会成功。

    至此可以看到,把tryAcquire方法延迟到子类中实现的做法非常精妙并具有极强的可扩展性,令人叹为观止!当然精妙的不是这个Template设计模式,而是Doug Lea对锁结构的精心布局。

    解锁代码相对简单,主要体现在AbstractQueuedSynchronizer.release和Sync.tryRelease方法中:
    class AbstractQueuedSynchronizer

    public final boolean release(int arg) {  
        if (tryRelease(arg)) {  
            Node h = head;  
            if (h != null && h.waitStatus != 0)  
                unparkSuccessor(h);  
            return true;  
        }  
        return false;  
    }  
    

    class Sync

    protected final boolean tryRelease(int releases) {  
        int c = getState() - releases;  
        if (Thread.currentThread() != getExclusiveOwnerThread())  
            throw new IllegalMonitorStateException();  
        boolean free = false;  
        if (c == 0) {  
            free = true;  
            setExclusiveOwnerThread(null);  
        }  
        setState(c);  
        return free;  
    }  
    

    tryRelease与tryAcquire语义相同,把如何释放的逻辑延迟到子类中。

    tryRelease语义很明确:如果线程多次锁定,则进行多次释放,直至status==0则真正释放锁,所谓释放锁即设置status为0,因为无竞争所以没有使用CAS。
    release的语义在于:如果可以释放锁,则唤醒队列第一个线程(Head),具体唤醒代码如下:

    private void unparkSuccessor(Node node) {  
        /* 
         * If status is negative (i.e., possibly needing signal) try 
         * to clear in anticipation of signalling. It is OK if this 
         * fails or if status is changed by waiting thread. 
         */  
        int ws = node.waitStatus;  
        if (ws < 0)  
            compareAndSetWaitStatus(node, ws, 0);   
    
        /* 
         * Thread to unpark is held in successor, which is normally 
         * just the next node.  But if cancelled or apparently null, 
         * traverse backwards from tail to find the actual 
         * non-cancelled successor. 
         */  
        Node s = node.next;  
        if (s == null || s.waitStatus > 0) {  
            s = null;  
            for (Node t = tail; t != null && t != node; t = t.prev)  
                if (t.waitStatus <= 0)  
                    s = t;  
        }  
        if (s != null)  
            LockSupport.unpark(s.thread);  
    }  
    

    这段代码的意思在于找出第一个可以unpark的线程,一般说来head.next == head,Head就是第一个线程,但Head.next可能被取消或被置为null,因此比较稳妥的办法是从后往前找第一个可用线程。貌似回溯会导致性能降低,其实这个发生的几率很小,所以不会有性能影响。之后便是通知系统内核继续该线程,在Linux下是通过pthread_mutex_unlock完成。之后,被解锁的线程进入上面所说的重新竞争状态。

    Lock VS Synchronized

    AbstractQueuedSynchronizer通过构造一个基于阻塞的CLH队列容纳所有的阻塞线程,而对该队列的操作均通过Lock-Free(CAS)操作,但对已经获得锁的线程而言,ReentrantLock实现了偏向锁的功能。

    synchronized的底层也是一个基于CAS操作的等待队列,但JVM实现的更精细,把等待队列分为ContentionList和EntryList,目的是为了降低线程的出列速度;当然也实现了偏向锁,从数据结构来说二者设计没有本质区别。但synchronized还实现了自旋锁,并针对不同的系统和硬件体系进行了优化,而Lock则完全依靠系统阻塞挂起等待线程。

    当然Lock比synchronized更适合在应用层扩展,可以继承AbstractQueuedSynchronizer定义各种实现,比如实现读写锁(ReadWriteLock),公平或不公平锁;同时,Lock对应的Condition也比wait/notify要方便的多、灵活的多。

    展开全文
  • 深入浅出Java锁--Lock实现原理(底层实现)

    千次阅读 多人点赞 2018-12-21 16:25:53
    当多个线程需要访问某个公共资源的时候,我们知道需要通过加锁来...synchronized是java底层支持的,而concurrent包则是jdk实现。   关于synchronized的原理可以阅读再有人问你synchronized是什么,就把这篇文章...

    当多个线程需要访问某个公共资源的时候,我们知道需要通过加锁来保证资源的访问不会出问题。

    java提供了两种方式来加锁,一种是关键字:synchronized,一种是concurrent包下的lock锁

    synchronized是java底层支持的,而concurrent包则是jdk实现

     

    关于synchronized的原理可以阅读再有人问你synchronized是什么,就把这篇文章发给他。

    在这里,我会用尽可能少的代码,尽可能轻松的文字,尽可能多的图来看看lock的原理。

    我们以ReentrantLock为例做分析,其他原理类似。

     

    我把这个过程比喻成一个做菜的过程,有什么菜,做法如何?

    我先列出lock实现过程中的几个关键词:计数值、双向链表、CAS+自旋

    我们以ReentrantLock为例做分析,其他原理类似。

    实现原理

    ReentrantLock() 干了啥

      public ReentrantLock() {
    
            sync = new NonfairSync();
    
        }

    在lock的构造函数中,定义了一个NonFairSync

    static final class NonfairSync extends Sync

    NonfairSync 又是继承于Sync

    abstract static class Sync extends AbstractQueuedSynchronizer

    一步一步往上找,找到了

    这个鬼AbstractQueuedSynchronizer(简称AQS),最后这个鬼,又是继承于AbstractOwnableSynchronizer(AOS),AOS主要是保存获取当前锁的线程对象,代码不多不再展开。最后我们可以看到几个主要类的继承关系:

                               

       FairSync 与 NonfairSync的区别在于,是不是保证获取锁的公平性,因为默认是NonfairSync,我们以这个为例了解其背后的原理。

    其他几个类代码不多,最后的主要代码都是在AQS中,我们先看看这个类的主体结构。

    看看AbstractQueuedSynchronizer是个什么

    再看看Node是什么?

    看到这里的同学,是不是有种热泪盈眶的感觉,这尼玛,不就是双向链表么?我还记得第一次写这个数据结构的时候,发现居然还有这么神奇的一个东西。

    最后我们可以发现锁的存储结构就两个东西:"双向链表" + "int类型状态"。

    需要注意的是,他们的变量都被"transientvolatile修饰。

    一个int值,一个双向链表是如何烹饪处理锁这道菜的呢,Doug Lea大神就是大神,

    我们接下来看看,如何获取锁?

    lock.lock()怎么获取锁?

    public void lock() {
    
        sync.lock();
    
    }

    可以看到调用的是,NonfairSync.lock()

    看到这里,我们基本有了一个大概的了解,还记得之前AQS中的int类型的state值

    这里就是通过CAS(乐观锁)去修改state的值。lock的基本操作还是通过乐观锁来实现的。

    获取锁通过CAS,那么没有获取到锁,等待获取锁是如何实现的?我们可以看一下else分支的逻辑,acquire方法:

    public final void acquire(int arg) {
    
        if (!tryAcquire(arg) &&
    
            acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
    
            selfInterrupt();
    
    }

    这里干了三件事情:

    • tryAcquire:会尝试再次通过CAS获取一次锁。

    • addWaiter:将当前线程加入上面锁的双向链表(等待队列)中

    • acquireQueued:通过自旋,判断当前队列节点是否可以获取锁。

     

    addWaiter() 添加当前线程到等待链表中

    可以看到,通过CAS确保能够在线程安全的情况下,将当前线程加入到链表的尾部。

    enq是个自旋+上述逻辑,有兴趣的可以翻翻源码。

     

    acquireQueued()    自旋+CAS尝试获取锁

    可以看到,当当前线程到头部的时候,尝试CAS更新锁状态,如果更新成功表示该等待线程获取成功。从头部移除。

     

    每一个线程都在 自旋+CAS

    最后简要概括一下,获取锁的一个流程


    lock.unlock() 释放锁

    public void unlock() {
    
        sync.release(1);
    
    }

    可以看到调用的是,NonfairSync.release()

    最后又调用了NonfairSync.tryRelease()

    基本可以确认,释放锁就是对AQS中的状态值State进行修改。同时更新下一个链表中的线程等待节点

    总结

    lock的存储结构:一个int类型状态值(用于锁的状态变更),一个双向链表(用于存储等待中的线程)

    lock获取锁的过程:本质上是通过CAS来获取状态值修改,如果当场没获取到,会将该线程放在线程等待链表中。

    lock释放锁的过程:修改状态值,调整等待链表。

    可以看到在整个实现过程中,lock大量使用CAS+自旋。因此根据CAS特性,lock建议使用在低锁冲突的情况下。目前java1.6以后,官方对synchronized做了大量的锁优化(偏向锁、自旋、轻量级锁)。因此在非必要的情况下,建议使用synchronized做同步操作。

    最后,希望我的分析,能对你理解锁的实现有所帮助。

     

    展开全文
  • 1. 深入理解synchronized关键词,介绍其发展历程,底层实现。 2. 介绍synchronized的范围,升级、降级过程。 3. 介绍synchronized关联关键词wait,notify, notifyAll。 4. synchronized相关联的常见面试题。

    底层实现

    synchronized底层实现

    详细参考:杨晓峰极客时间上的课程《Java核心技术面试精讲》:第16讲 | synchronized底层如何实现?什么是锁的升级、降级

    • synchronized 代码块是由一对儿 monitorenter/monitorexit 指令实现的,Monitor 对象是同步的基本实现单元。

    • 发展历程:

      • JDK6之前,Monitor 的实现完全是依靠操作系统内部的互斥锁,因为需要进行用户态到内核态的切换,所以同步操作是一个无差别的重量级操作。(@所以说效率低下嘛)
      • 现代的(Oracle)JDK 中,JVM 对此进行了大刀阔斧地改进,提供了三种不同的 Monitor 实现,也就是常说的三种不同的锁:偏斜锁(Biased Locking)、轻量级锁和重量级锁大大改进了其性能。
    • 偏斜锁:JVM 会利用 CAS 操作(compare and swap),在对象头上的 Mark Word 部分设置线程 ID,以表示这个对象偏向于当前线程,所以并不涉及真正的互斥锁

    • 轻量级锁:

      • 如果有另外的线程试图锁定某个已经被偏斜过的对象,JVM 就需要撤销(revoke)偏斜锁,并切换到轻量级锁实现。轻量级锁依赖 CAS 操作 Mark Word 来试图获取锁,如果重试成功,就使用普通的轻量级锁;否则,进一步升级为重量级锁。
      • 因为重量级锁性能差,所以轻量级锁又衍生出了一种锁:自旋锁,其实现就是自循环若干次,通过CAS操作MARK WROD试图获取锁

    其他锁模型

    详细参考:王宝令极客时间上的课程《Java并发编程实战》- 08 | 管程:并发编程的万能钥匙

    • 管程模型:
      • Hasen模型:要求 notify() 放在代码的最后,这样 T2 通知完 T1 后,T2 就结束了,然后 T1 再执行,这样就能保证同一时刻只有一个线程执行。
      • Hoare模型:T2 通知完 T1 后,T2 阻塞,T1 马上执行;等 T1 执行完,再唤醒 T2,也能保证同一时刻只有一个线程执行。但是相比 Hasen 模型,T2 多了一次阻塞唤醒操作。
      • MESA模型(JAVA参考实现):MESA 管程里面,T2 通知完 T1 后,T2 还是会接着执行,T1 并不立即执行,仅仅是从条件变量的等待队列进到入口等待队列里面。这样做的好处是 notify() 不用放到代码的最后,T2 也没有多余的阻塞唤醒操作。但是也有个副作用,就是当 T1 再次执行的时候,可能曾经满足的条件,现在已经不满足了,所以需要以循环方式检验条件变量。—也就是产生假唤醒

    锁变化

    升级/膨胀

    其实就是偏斜锁=》轻量级锁=》重量级锁的过程,见第一节 #底层实现

    锁降级

    锁降级确实是会发生的,当 JVM 进入安全点(SafePoint)的时候,会检查是否有闲置的 Monitor,然后试图进行降级。

    synchronized锁的范围

    • 范围
      • 代码块
      • 方法
      • 对象
    • 对象锁和类
      • 类锁和对象锁是分开的,(现在只是个概念,用来区分对象锁的,是指静态方法的锁),程序中获得类锁的同时也可以获得对象锁。
      • 同一个类锁和同一个类锁是互斥的,同一个对象锁和同一个对象锁互斥。 非静态方法不受类锁的影响
      • 对象锁与实例对象相关, 不同的对象的对象锁不一样,可以同时获取两个不同对象的对象锁
    package com.keven;
    
    //类锁和对象锁的测试代码
    public class SyncTest {
    
        public static void main(String[] args) throws Exception {
            runObjectLockTest();
            System.out.println("finished runObjectLockTest");
            runClassLockTest();
            System.out.println("finished runClassLockTest");
            runClassObjectLockTest();
            System.out.println("finished runClassObjectLockTest");
            Thread.sleep(10000);
        }
    
        //测试对象锁和类锁是否能够同时获取, 可以看到两个线程打印数据不受影响,说明不是同一个锁
        private static void runClassObjectLockTest() {
            new Thread(SyncTest::testClassLock1, "thread1").start();
            new Thread(() -> {
                new SyncTest().testObjectLock();
            }, "thread2").start();
        }
    
        //测试类锁,显示thread1打印完成,后面thread2才开始打印,从侧面验证获取到的是同一个锁
        private static void runClassLockTest() {
            new Thread(SyncTest::testClassLock1, "thread1").start();
            new Thread(SyncTest::testClassLock2, "thread2").start();
    
        }
    
        //测试对象锁, 可以看到两个线程打印数据不受影响, 且this对象的hash值不一样
        private static void runObjectLockTest() {
            final SyncTest syncTest = new SyncTest();
            final Thread thread1 = new Thread(() -> {
                syncTest.testObjectLock();
            }, "thread1");
    
            final Thread thread2 = new Thread(() -> {
                new SyncTest().testObjectLock();
            }, "thread2");
            thread1.start();
            thread2.start();
        }
    
        private static synchronized void testClassLock1() {
            int i = 100;
            int count = 0;
            while ((i-- > 0) && (count++ < 10)) {
                System.out.println("method testClassLock1--" + Thread.currentThread().getName() + " : " + i);
                try {
                    Thread.sleep(50);
                } catch (InterruptedException ie) {
                }
            }
        }
    
        private static synchronized void testClassLock2() {
            int i = 100;
            int count = 0;
            while ((i-- > 0) && (count++ < 10)) {
                System.out.println("method testClassLock2--" + Thread.currentThread().getName() + " : " + i);
                try {
                    Thread.sleep(50);
                } catch (InterruptedException ie) {
                }
            }
        }
    
        private void testObjectLock() {
            synchronized (this) {
                System.out.println(Thread.currentThread().getName() + " : " + this);
                int i = 100;
                int count = 0;
                while ((i-- > 0) && (count++ < 5)) {
                    System.out.println("method testObjectLock--" + Thread.currentThread().getName() + " : " + i);
                    try {
                        Thread.sleep(50);
                    } catch (InterruptedException ie) {
                    }
                }
            }
        }
    
    }
    
    

    synchronized和ReentrantLock有什么区别?

    详细可参考:杨晓峰极客时间上的课程《Java核心技术面试精讲》:第15讲 | synchronized和ReentrantLock有什么区别呢?

    • synchronized 和 ReentrantLock 的性能不能一概而论:
      • 早起版本的synchronize在很多场景下性能相差较大
      • 在后续版本进行了较多的改进,在低竞争场景中表现可能优于ReentrantLock
    • 这里所谓的公平性是指在竞争场景中,当公平性为真时,会倾向于将锁赋予等待时间最久的线程。公平性是减少线程“饥饿”(个别线程长期等待锁,但始终无法获取)情况发生的一个办法。
    • ReentrantLock与Synchronized的区别:
      • ReentrantLock
        • 更加的灵活,但必须手动释放锁
          • 可通过条件控制同步
          • 可被中断,并抛出中断异常,释放锁
          • 可选择获取锁的超时时间,尝试获取锁
          • 可选择是否为公平锁
        • 只适合代码块的锁
      • synchronized
        • 无需释放锁,自动处理
        • 可修饰方法,类,代码块
        • 非公平锁,如果阻塞则必须等待cpu调度
    • ReentrantLock与Synchronized的共通点:都是独占锁或者说是排它锁

    关联关键词

    • 在上面的代码中,我用的是 notifyAll() 来实现通知机制,为什么不使用 notify() 呢?
      • 这二者是有区别的,notify() 是会随机地通知等待队列中的一个线程,而 notifyAll() 会通知等待队列中的所有线程。
      • 从感觉上来讲,应该是 notify() 更好一些,因为即便通知所有线程,也只有一个线程能够进入临界区。但那所谓的感觉往往都蕴藏着风险,实际上使用 notify() 也很有风险,它的风险在于可能导致某些线程永远不会被通知到。@随机的弊病不就是存在永远不被轮到的弊病么?这跟非公平锁的弊病是一个意思
    • wait与sleep区别在于:
      • wait会释放所有锁而sleep不会释放锁资源.
      • wait只能在同步方法和同步块中使用,而sleep任何地方都可以
      • wait无需捕捉异常,而sleep需要
      • sleep是Thread的方法,而wait是Object类的方法;
      • sleep方法调用的时候必须指定时间
        两者相同点:都会让渡CPU执行时间,等待再次调度!。补充关于二者的区别还可以看知乎的这篇帖子
    • wait()方法与sleep()方法的不同之处在于,wait()方法会释放对象的“锁标志”。当调用某一对象的wait()方法后,会使当前线程暂停执行,并将当前线程放入对象等待池中,直到调用了notify()方法后,将从对象等待池中移出任意一个线程并放入锁标志等待池中,只有锁标志等待池中的线程可以获取锁标志,它们随时准备争夺锁的拥有权。当调用了某个对象的notifyAll()方法,会将对象等待池中的所有线程都移动到该对象的锁标志等待池。
    • sleep()方法需要指定等待的时间,它可以让当前正在执行的线程在指定的时间内暂停执行,进入阻塞状态,该方法既可以让其他同优先级或者高优先级的线程得到执行的机会,也可以让低优先级的线程得到执行机会。但是sleep()方法不会释放“锁标志”,也就是说如果有synchronized同步块,其他线程仍然不能访问共享数据

    常见面试题

    • synchronized和ReentrantLock的区别 @见笔记
    • 锁什么时候升级/降级?@见笔记
    • 类锁和对象锁的区别? @见笔记
    • 为什么JDK8中ConcurrentHashMap的锁实现要用CAS+synchronized来取代Segment+ReentrantLock呢?
    • 为什么wait必须是在同步块中的呢?@重看了一遍王宝令的课程,发现这是MESA管程模型的设计范式,硬要解释的话可以是这样:
      • wait是跟notify, notifyAll配对的, 是和synchronized关键字一起使用的
      • wait的工作原理就是wait的时候,会进入同步块(synchronized)所对应的条件等待队列,在其他地方使用这个关键字是不可进入的
      • 或者说wait所对应的管程的入口在synchronied处
    • wait与sleep区别是什么? @见上面的笔记
    展开全文
  • Java多线程:底层实现

    千次阅读 2018-07-31 11:39:11
    最近准备招聘,在网上搜集了很多关于JVM获取的流程的文章,基本上所有关于偏向获取的部分都是错的,所以专门花时间自己看了看代码,整理出来供大家参考。 一、 基本概念 CAS(compare and swap) 参考...

    最近准备招聘,在网上搜集了很多关于JVM获取的锁流程的文章,基本上所有关于偏向锁获取的部分都是错的,所以专门花时间自己看了看代码,整理出来供大家参考。

    本文章是Java多线程系列的一篇文章,其他文章:
    Java多线程:锁的底层实现
    Java多线程:synchronized和volatile
    Java多线程:JUC包-锁的封装
    Java多线程:Thread的使用,以及wait(),notify(),notifyAll()
    Java多线程:线程池

    一、 基本概念

    CAS(compare and swap)

    参考资料:
    https://blog.csdn.net/ls5718/article/details/52563959
    缓存一致性协议

    1. 什么是CAS

    为了提高性能,JVM很多操作都依赖CAS实现,一种乐观锁的实现。本文锁优化中大量用到了CAS,故有必要先分析一下CAS的实现。

    CAS:Compare and Swap。如果内存位置的值与预期原值相匹配,那么处理器会自动将该位置值更新为新值 。否则,处理器不做任何操作。无论哪种情况,它都会在 CAS 指令之前返回该 位置的值。

    JNI来完成CPU指令的操作:

    openjdk\hotspot\src\os_cpu\windows_x86\vm\atomic_windows_x86.inline.hpp

    //这里只粘贴了windowsx86的一个实现
    //即:若dest和compare_value的值相同,就将dest的值替换为exchange_value
    inline intptr_t Atomic::cmpxchg_ptr(intptr_t exchange_value, volatile intptr_t* dest, intptr_t compare_value) {
      return (intptr_t)cmpxchg((jlong)exchange_value, (volatile jlong*)dest, (jlong)compare_value);
    }
    
    inline jlong    Atomic::cmpxchg    (jlong    exchange_value, volatile jlong*    dest, jlong    compare_value) {
      int mp = os::is_MP();//返回是否为多处理器
      jint ex_lo  = (jint)exchange_value;
      jint ex_hi  = *( ((jint*)&exchange_value) + 1 );
      jint cmp_lo = (jint)compare_value;
      jint cmp_hi = *( ((jint*)&compare_value) + 1 );
      __asm {
        push ebx
        push edi
        mov eax, cmp_lo
        mov edx, cmp_hi
        mov edi, dest
        mov ebx, ex_lo
        mov ecx, ex_hi
        LOCK_IF_MP(mp) //会根据当前处理器的类型来决定是否为cmpxchg指令添加lock前缀。如果程序是在多处理器上运行,就为cmpxchg指令加上lock前缀(lock cmpxchg)。反之,如果程序是在单处理器上运行,就省略lock前缀(单处理器自身会维护单处理器内的顺序一致性,不需要lock前缀提供的内存屏障效果)。
        cmpxchg8b qword ptr [edi]
        pop edi
        pop ebx
      }
    }

    CAS有3个操作数,内存值V,旧的预期值A,要修改的新值B。如果A=V,那么把B赋值给V,返回V;如果A!=V,直接返回V。

    2. CAS的目的

    利用CPU的CAS指令,同时借助JNI来完成Java的非阻塞算法。其它原子操作都是利用类似的特性完成的。而整个J.U.C都是建立在CAS之上的,因此对于synchronized阻塞算法,J.U.C在性能上有了很大的提升。

    3. CAS存在的问题

    CAS虽然很高效的解决原子操作,但是CAS仍然存在三大问题。ABA问题,循环时间长开销大和只能保证一个共享变量的原子操作

    1. ABA问题。因为CAS需要在操作值的时候检查下值有没有发生变化,如果没有发生变化则更新,但是如果一个值原来是A,变成了B,又变成了A,那么使用CAS进行检查时会发现它的值没有发生变化,但是实际上却变化了。ABA问题的解决思路就是使用版本号。在变量前面追加上版本号,每次变量更新的时候把版本号加一,那么A-B-A 就会变成1A-2B-3A。

    从Java1.5开始JDK的atomic包里提供了一个类AtomicStampedReference来解决ABA问题。这个类的compareAndSet方法作用是首先检查当前引用是否等于预期引用,并且当前标志是否等于预期标志,如果全部相等,则以原子方式将该引用和该标志的值设置为给定的更新值。

    关于ABA问题参考文档: http://blog.hesey.net/2011/09/resolve-aba-by-atomicstampedreference.html

    1. 循环时间长开销大。自旋CAS如果长时间不成功,会给CPU带来非常大的执行开销。如果JVM能支持处理器提供的pause指令那么效率会有一定的提升,pause指令有两个作用,第一它可以延迟流水线执行指令(de-pipeline),使CPU不会消耗过多的执行资源,延迟的时间取决于具体实现的版本,在一些处理器上延迟时间是零。第二它可以避免在退出循环的时候因内存顺序冲突(memory order violation)而引起CPU流水线被清空(CPU pipeline flush),从而提高CPU的执行效率。

    2. 只能保证一个共享变量的原子操作。当对一个共享变量执行操作时,我们可以使用循环CAS的方式来保证原子操作,但是对多个共享变量操作时,循环CAS就无法保证操作的原子性,这个时候就可以用锁,或者有一个取巧的办法,就是把多个共享变量合并成一个共享变量来操作。比如有两个共享变量i=2,j=a,合并一下ij=2a,然后用CAS来操作ij。从Java1.5开始JDK提供了AtomicReference类来保证引用对象之间的原子性,你可以把多个变量放在一个对象里来进行CAS操作。

    安全点(safe point)

    http://calvin1978.blogcn.com/articles/safepoint.html

    GC时的Stop the World(STW)是大家最大的敌人。但可能很多人没留意,除了GC,JVM底下还会发生这样那样的停顿。

    JVM里有一条特殊的线程--VM Threads,专门用来执行一些特殊的VM Operation,比如分派GC的STW,thread dump等,这些任务,都需要整个Heap,以及所有线程的状态是静止的,一致的才能进行。所以JVM引入了安全点(Safe Point)的概念,想办法在需要进行VM Operation时,通知所有的线程进入一个静止的安全点。 如何做到的见 聊聊JVM(九)理解进入safepoint时如何让Java线程全部阻塞

    除了GC,其他触发安全点的VM Operation包括:

    • 1. JIT相关,比如Code deoptimization, Flushing code cache
    • 2. Class redefinition (e.g. javaagent,AOP代码植入的产生的instrumentation)
    • 3. Biased lock revocation 取消偏向锁
    • 4. Various debug operation (e.g. thread dump or deadlock check)

    打开JDK7源码的vm_operations.hpp,有一个长长的列表,比如ReportJavaOutOfMemory等等不能尽录。

    Java中的Monitor

    参考:
    java线程同步原理
    Monitors – The Basic Idea of Java Synchronization => 译文:监视器–JAVA同步基本概念

    java会为每个object对象分配一个monitor,当某个对象的同步方法(synchronized methods )被多个线程调用时,该对象的monitor将负责处理这些访问的并发独占要求。
    当一个线程调用一个对象的同步方法时,JVM会检查该对象的monitor。如果monitor没有被占用,那么这个线程就得到了monitor的占有权,可以继续执行该对象的同步方法;如果monitor被其他线程所占用,那么该线程将被挂起,直到monitor被释放。
    当线程退出同步方法调用时,该线程会释放monitor,这将允许其他等待的线程获得monitor以使对同步方法的调用执行下去。
    注意:Java对象的monitor机制和传统的临界检查代码区技术不一样。java的一个同步方法并不意味着同时只有一个线程独占执行,但临界检查代码区技术确实会保证同步方法在一个时刻只被一个线程独占执行。Java的monitor机制的准确含义是:任何时刻,对一个指定object对象的某同步方法只能由一个线程来调用。
    java对象的monitor是跟随object实例来使用的,而不是跟随程序代码。两个线程可以同时执行相同的同步方法,比如:一个类的同步方法是xMethod(),有a,b两个对象实例,一个线程执行a.xMethod(),另一个线程执行b.xMethod(). 互不冲突。

    hashCode相关

    StackOverFlow:When does the jvm assign hashcode value in the object header
    StackOverFlow:How do hashCode() and identityHashCode() work at the back end?
    知乎:当Java处在偏向锁、重量级锁状态时,hashcode值存储在哪?
    hashCode和identityHashCode底层是怎么生成的

    二、 基础数据结构

    markword

    32位虚拟机markword

    oopDesc

    openjdk\hotspot\src\share\vm\oops\oop.hpp => oopDesc

    class oopDesc {
      friend class VMStructs;
     private:
      volatile markOop  _mark;//markOop:Mark Word标记字段
      union _metadata {
        Klass*      _klass;//对象类型元数据的指针
        narrowKlass _compressed_klass;
      } _metadata;
    
      // Fast access to barrier set.  Must be initialized.
      static BarrierSet* _bs;
      //以下省略
      //...
    }

    锁状态mask

    openjdk\hotspot\src\share\vm\oops\markOop.hpp => markOopDesc

      enum { 
             locked_value             = 0,//00偏向锁 
             unlocked_value           = 1,//01无锁
             monitor_value            = 2,//10监视器锁,又叫重量级锁
             marked_value             = 3,//11GC标记
             biased_lock_pattern      = 5//101偏向锁
      };

    三、 获取锁的流程

    1. monitorenter()

    synchronized关键字修饰的代码段,在JVM被编译为monitorenter、monitorexit指令来获取和释放互斥锁.。

    解释器执行monitorenter时会进入到InterpreterRuntime.cpp的InterpreterRuntime::monitorenter函数,具体实现如下:

    openjdk/hotspot/src/share/vm/interpreter/interpreterRuntime.cpp => monitor_enter()

    //------------------------------------------------------------------------------------------------------------------------
    // Synchronization
    //
    // The interpreter's synchronization code is factored out so that it can
    // be shared by method invocation and synchronized blocks.
    //%note synchronization_3
    
    //%note monitor_1
    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) {//如果开启了偏向锁,则直接使用fast_enter()
        // Retry fast entry if bias is revoked to avoid unnecessary inflation
        ObjectSynchronizer::fast_enter(h_obj, elem->lock(), true, CHECK);
      } else {//否则使用slow_enter()
        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

    总的来说,就是:如果开启了偏向锁,则直接进入fast_enter()方法,否则进入slow_enter()方法

    2. 尝试获取偏向锁:fast_enter()

    openjdk/hotspot/src/share/vm/interpreter/synchronizer.cpp => ObjectSynchronizer::fast_enter()

    // -----------------------------------------------------------------------------
    //  Fast Monitor Enter/Exit
    // This the fast monitor enter. The interpreter and compiler use
    // some assembly copies of this code. Make sure update those code
    // if the following function is changed. The implementation is
    // extremely sensitive to race condition. Be careful.
    
    void ObjectSynchronizer::fast_enter(Handle obj, BasicLock* lock, bool attempt_rebias, TRAPS) {
     if (UseBiasedLocking) {//如果开启了偏向锁
        if (!SafepointSynchronize::is_at_safepoint()) {//如果不在安全点
          BiasedLocking::Condition cond = BiasedLocking::revoke_and_rebias(obj, attempt_rebias, THREAD);
          if (cond == BiasedLocking::BIAS_REVOKED_AND_REBIASED) {//如果调用结果为BIAS_REVOKED_AND_REBIASED
            return;
          }
        } else {//当前处于安全点,调用revoke_at_safepoint
          assert(!attempt_rebias, "can not rebias toward VM thread");
          BiasedLocking::revoke_at_safepoint(obj);
        }
        assert(!obj->mark()->has_bias_pattern(), "biases should be revoked by now");
     }
    
     slow_enter (obj, lock, THREAD) ;//如果未开启偏向锁,或竞争偏向锁失败
    }
    

    这段代码很短,就不作太多解释

    2.1 revoke_and_rebias

    openjdk/hotspot/src/share/vm/runtime/biasedLocking.cpp => revoke_and_rebias()

    /**
    * @param obj: 要执行操作的对象
    * @param attempt_rebias: 尝试将该对象偏向到当前线程
    * @param TRAPS: 
    */
    BiasedLocking::Condition BiasedLocking::revoke_and_rebias(Handle obj, bool attempt_rebias, TRAPS) {
      assert(!SafepointSynchronize::is_at_safepoint(), "must not be called while at safepoint");
    
      // We can revoke the biases of anonymously-biased objects
      // efficiently enough that we should not cause these revocations to
      // update the heuristics because doing so may cause unwanted bulk
      // revocations (which are expensive) to occur.
      markOop mark = obj->mark();
      // 如果没有其他线程占用该对象(线程ID为0,后三位为101)且不尝试重新偏向(注意这里是不尝试重偏向,在`fast_enter()`中传过来的参数是true,也就是尝试重新偏向)
      if (mark->is_biased_anonymously() && !attempt_rebias) {
        // We are probably trying to revoke the bias of this object due to
        // an identity hash code computation. Try to revoke the bias
        // without a safepoint. This is possible if we can successfully
        // compare-and-exchange an unbiased header into the mark word of
        // the object, meaning that no other thread has raced to acquire
        // the bias of the object.
        //一般来讲,只有在重新计算对象hashCode的时候才会进入该分支,
        //所以直接用用CAS操作将对象设置为无锁状态
        markOop biased_value       = mark;
        markOop unbiased_prototype = markOopDesc::prototype()->set_age(mark->age());
        markOop res_mark = (markOop) Atomic::cmpxchg_ptr(unbiased_prototype, obj->mark_addr(), mark);
        if (res_mark == biased_value) {//如果CAS操作失败,说明存在竞争,交给调用者处理
          return BIAS_REVOKED;
        }
      } else if (mark->has_bias_pattern()) {//如果对象为可偏向状态101,且尝试重新偏向(不管线程ID是不是为空)
        Klass* k = obj->klass();
        markOop prototype_header = k->prototype_header();
        if (!prototype_header->has_bias_pattern()) {//如果有线程对该对象进行了全局锁定(即同步了静态方法/属性),则取消偏向操作
          // This object has a stale bias from before the bulk revocation
          // for this data type occurred. It's pointless to update the
          // heuristics at this point so simply update the header with a
          // CAS. If we fail this race, the object's bias has been revoked
          // by another thread so we simply return and let the caller deal
          // with it.
          markOop biased_value       = mark;
          markOop res_mark = (markOop) Atomic::cmpxchg_ptr(prototype_header, obj->mark_addr(), mark);
          assert(!(*(obj->mark_addr()))->has_bias_pattern(), "even if we raced, should still be revoked");
          return BIAS_REVOKED;
        } else if (prototype_header->bias_epoch() != mark->bias_epoch()) {//如果偏向时间戳没有过期,是不会重新偏向的,因为这个时候有另一个线程通过偏向锁获取到了这个对象的锁
          // The epoch of this biasing has expired indicating that the
          // object is effectively unbiased. Depending on whether we need
          // to rebias or revoke the bias of this object we can do it
          // efficiently enough with a CAS that we shouldn't update the
          // heuristics. This is normally done in the assembly code but we
          // can reach this point due to various points in the runtime
          // needing to revoke biases.
          if (attempt_rebias) {//尝试重新偏向
            assert(THREAD->is_Java_thread(), "");
            markOop biased_value       = mark;
            markOop rebiased_prototype = markOopDesc::encode((JavaThread*) THREAD, mark->age(), prototype_header->bias_epoch());
            markOop res_mark = (markOop) Atomic::cmpxchg_ptr(rebiased_prototype, obj->mark_addr(), mark);
            if (res_mark == biased_value) {
              return BIAS_REVOKED_AND_REBIASED;
            }
          } else {
            markOop biased_value       = mark;
            markOop unbiased_prototype = markOopDesc::prototype()->set_age(mark->age());
            markOop res_mark = (markOop) Atomic::cmpxchg_ptr(unbiased_prototype, obj->mark_addr(), mark);
            if (res_mark == biased_value) {
              return BIAS_REVOKED;
            }
          }
        }
      }
    
      HeuristicsResult heuristics = update_heuristics(obj(), attempt_rebias);
      if (heuristics == HR_NOT_BIASED) {
        return NOT_BIASED;
      } else if (heuristics == HR_SINGLE_REVOKE) {
        Klass *k = obj->klass();
        markOop prototype_header = k->prototype_header();
        if (mark->biased_locker() == THREAD &&
            prototype_header->bias_epoch() == mark->bias_epoch()) {
          // A thread is trying to revoke the bias of an object biased
          // toward it, again likely due to an identity hash code
          // computation. We can again avoid a safepoint in this case
          // since we are only going to walk our own stack. There are no
          // races with revocations occurring in other threads because we
          // reach no safepoints in the revocation path.
          // Also check the epoch because even if threads match, another thread
          // can come in with a CAS to steal the bias of an object that has a
          // stale epoch.
          ResourceMark rm;
          if (TraceBiasedLocking) {
            tty->print_cr("Revoking bias by walking my own stack:");
          }
          BiasedLocking::Condition cond = revoke_bias(obj(), false, false, (JavaThread*) THREAD);
          ((JavaThread*) THREAD)->set_cached_monitor_info(NULL);
          assert(cond == BIAS_REVOKED, "why not?");
          return cond;
        } else {
          VM_RevokeBias revoke(&obj, (JavaThread*) THREAD);
          VMThread::execute(&revoke);
          return revoke.status_code();
        }
      }
    
      assert((heuristics == HR_BULK_REVOKE) ||
             (heuristics == HR_BULK_REBIAS), "?");
      VM_BulkRevokeBias bulk_revoke(&obj, (JavaThread*) THREAD,
                                    (heuristics == HR_BULK_REBIAS),
                                    attempt_rebias);
      VMThread::execute(&bulk_revoke);
      return bulk_revoke.status_code();
    }

    这个函数定义了获取偏向锁的流程,具体逻辑总结:
    1. 判断当前对象是否为可偏向(101),且偏向时间戳已过期(没有其他线程在占用该对象),如果是,则进入步骤2,否则进入步骤3
    2. 执行CAS操作将markword中的线程ID替换为本线程ID。如果成功则进入步骤4,否则进入步骤3
    3. 存在竞争,当达到全局安全点(safepoint),获得偏向锁的线程被挂起,撤销偏向锁,并升级为轻量级,升级完成后被阻塞在安全点的线程继续执行同步代码块;
    4. 执行同步代码

    2.2 偏向锁的撤销: revoke_at_safepoint

    只有当其它线程尝试竞争偏向锁时,持有偏向锁的线程才会释放锁,偏向锁的撤销由BiasedLocking::revoke_at_safepoint方法实现:
    openjdk\hotspot\src\share\vm\runtime\synchronizer.cpp

    void BiasedLocking::revoke_at_safepoint(Handle h_obj) {
      assert(SafepointSynchronize::is_at_safepoint(), "must only be called while at safepoint");
      oop obj = h_obj();
      HeuristicsResult heuristics = update_heuristics(obj, false);
      if (heuristics == HR_SINGLE_REVOKE) {
        revoke_bias(obj, false, false, NULL);
      } else if ((heuristics == HR_BULK_REBIAS) ||
                 (heuristics == HR_BULK_REVOKE)) {
        bulk_revoke_or_rebias_at_safepoint(obj, (heuristics == HR_BULK_REBIAS), false, NULL);
      }
      clean_up_cached_monitor_info();
    }

    偏向锁的释放,需要等待全局安全点(在这个时间点上没有正在执行的字节码),它会首先暂停拥有偏向锁的线程,然后检查持有偏向锁的线程是否还活着,如果线程不处于活动状态,则将对象头设置成无锁状态。如果线程仍然活着,拥有偏向锁的栈会被执行,遍历偏向对象的所记录。栈帧中的锁记录和对象头的Mark Word要么重新偏向其他线程,要么恢复到无锁,或者标记对象不适合作为偏向锁。最后唤醒暂停的线程。

    偏向锁在Java 1.6之后是默认启用的,但在应用程序启动几秒钟之后才激活,可以使用-XX:BiasedLockingStartupDelay=0参数关闭延迟,如果确定应用程序中所有锁通常情况下处于竞争状态,可以通过XX:-UseBiasedLocking=false参数关闭偏向锁。

    3.1 轻量级锁的获取: slow_enter

    如果关闭了偏向锁功能,则会直接再monitorenter方法中进入slow_enter。否则,在faster_enter中竞争偏向锁导致偏向锁升级为轻量级锁后进入。

    openjdk\hotspot\src\share\vm\runtime\synchronizer.cpp => slow_enter()

    // -----------------------------------------------------------------------------
    // Interpreter/Compiler Slow Case
    // This routine is used to handle interpreter/compiler slow case
    // We don't need to use fast path here, because it must have been
    // failed in the interpreter/compiler code.
    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())) {
        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);//锁膨胀升级
    }

    1、markOop mark = obj->mark()方法获取对象的markOop数据mark;
    2、mark->is_neutral()方法判断mark是否为无锁状态:mark的偏向锁标志位为 0,锁标志位为 01;
    3、如果mark处于无锁状态,则进入步骤(4),否则执行步骤(6);
    4、把mark保存到BasicLock对象的_displaced_header字段;
    5、通过CAS尝试将Mark Word更新为指向BasicLock对象的指针,如果更新成功,表示竞争到锁,则执行同步代码,否则执行步骤(6);
    6、如果当前mark处于加锁状态,且mark中的ptr指针指向当前线程的栈帧,则执行同步代码,否则说明有多个线程竞争轻量级锁,轻量级锁需要膨胀升级为重量级锁;

    假设线程A和B同时执行到临界区if (mark->is_neutral())
    1、线程AB都把Mark Word复制到各自的_displaced_header字段,该数据保存在线程的栈帧上,是线程私有的;
    2、Atomic::cmpxchg_ptr原子操作保证只有一个线程可以把指向栈帧的指针复制到Mark Word,假设此时线程A执行成功,并返回继续执行同步代码块;
    3、线程B执行失败,退出临界区,通过ObjectSynchronizer::inflate方法开始膨胀锁;

    3.2轻量级锁的释放:fast_exit

    void ObjectSynchronizer::fast_exit(oop object, BasicLock* lock, TRAPS) {
      assert(!object->mark()->has_bias_pattern(), "should not see bias pattern here");
      // if displaced header is null, the previous enter is recursive enter, no-op
      markOop dhw = lock->displaced_header();
      markOop mark ;
      if (dhw == NULL) {
         // Recursive stack-lock.
         // Diagnostics -- Could be: stack-locked, inflating, inflated.
         mark = object->mark() ;
         assert (!mark->is_neutral(), "invariant") ;
         if (mark->has_locker() && mark != markOopDesc::INFLATING()) {
            assert(THREAD->is_lock_owned((address)mark->locker()), "invariant") ;
         }
         if (mark->has_monitor()) {
            ObjectMonitor * m = mark->monitor() ;
            assert(((oop)(m->object()))->mark() == mark, "invariant") ;
            assert(m->is_entered(THREAD), "invariant") ;
         }
         return ;
      }
    
      mark = object->mark() ;
    
      // If the object is stack-locked by the current thread, try to
      // swing the displaced header from the box back to the mark.
      if (mark == (markOop) lock) {
         assert (dhw->is_neutral(), "invariant") ;
         if ((markOop) Atomic::cmpxchg_ptr (dhw, object->mark_addr(), mark) == mark) {//成功的释放了锁
            TEVENT (fast_exit: release stacklock) ;
            return;
         }
      }
    
      //如果没有成功释放锁,则说明有其他线程竞争,进行锁膨胀
      ObjectSynchronizer::inflate(THREAD, object)->exit (true, THREAD) ;
    }

    1、确保处于偏向锁状态时不会执行这段逻辑;
    2、取出在获取轻量级锁时保存在BasicLock对象的mark数据dhw;
    3、通过CAS尝试把dhw替换到当前的Mark Word,如果CAS成功,说明成功的释放了锁,否则执行步骤(4);
    4、如果CAS失败,说明有其它线程在尝试获取该锁,这时需要将该锁升级为重量级锁,并释放;

    4. 重量级锁:

    重量级锁通过对象内部的监视器(monitor)实现,其中monitor的本质是依赖于底层操作系统的Mutex Lock实现,操作系统实现线程之间的切换需要从用户态到内核态的切换,切换成本非常高。

    4.1 重量级锁的膨胀过程:ObjectSynchronizer::inflate

    
    // Note that we could encounter some performance loss through false-sharing as
    // multiple locks occupy the same $ line.  Padding might be appropriate.
    
    
    ObjectMonitor * ATTR ObjectSynchronizer::inflate (Thread * Self, oop object) {
      // Inflate mutates the heap ...
      // Relaxing assertion for bug 6320749.
      assert (Universe::verify_in_progress() ||
              !SafepointSynchronize::is_at_safepoint(), "invariant") ;
    
      for (;;) {//自旋
          const markOop mark = object->mark() ;
          assert (!mark->has_bias_pattern(), "invariant") ;
    
          // The mark can be in one of the following states:
          // *  Inflated     - just return
          // *  Stack-locked - coerce it to inflated
          // *  INFLATING    - busy wait for conversion to complete
          // *  Neutral      - aggressively inflate the object.
          // *  BIASED       - Illegal.  We should never see this
    
          // CASE: inflated  --已膨胀,即重量级锁
          if (mark->has_monitor()) {//判断当前是否为重量级锁
              ObjectMonitor * inf = mark->monitor() ;//获取指向ObjectMonitor的指针
              assert (inf->header()->is_neutral(), "invariant");
              assert (inf->object() == object, "invariant") ;
              assert (ObjectSynchronizer::verify_objmon_isinpool(inf), "monitor is invalid");
              return inf ;
          }
    
          // CASE: inflation in progress - inflating over a stack-lock. .膨胀等待(其他线程正在从轻量级锁转为膨胀锁)
          // Some other thread is converting from stack-locked to inflated.
          // Only that thread can complete inflation -- other threads must wait.
          // The INFLATING value is transient.
          // Currently, we spin/yield/park and poll the markword, waiting for inflation to finish.
          // We could always eliminate polling by parking the thread on some auxiliary list.
          if (mark == markOopDesc::INFLATING()) {
             TEVENT (Inflate: spin while INFLATING) ;
             ReadStableMark(object) ;
             continue ;
          }
    
          // CASE: stack-locked 栈锁(轻量级锁) 
          // Could be stack-locked either by this thread or by some other thread.
          //
          // Note that we allocate the objectmonitor speculatively, _before_ attempting
          // to install INFLATING into the mark word.  We originally installed INFLATING,
          // allocated the objectmonitor, and then finally STed the address of the
          // objectmonitor into the mark.  This was correct, but artificially lengthened
          // the interval in which INFLATED appeared in the mark, thus increasing
          // the odds of inflation contention.
          //
          // We now use per-thread private objectmonitor free lists.
          // These list are reprovisioned from the global free list outside the
          // critical INFLATING...ST interval.  A thread can transfer
          // multiple objectmonitors en-mass from the global free list to its local free list.
          // This reduces coherency traffic and lock contention on the global free list.
          // Using such local free lists, it doesn't matter if the omAlloc() call appears
          // before or after the CAS(INFLATING) operation.
          // See the comments in omAlloc().
    
          if (mark->has_locker()) {
              ObjectMonitor * m = omAlloc (Self) ;//获取一个可用的ObjectMonitor 
              // Optimistically prepare the objectmonitor - anticipate successful CAS
              // We do this before the CAS in order to minimize the length of time
              // in which INFLATING appears in the mark.
              m->Recycle();
              m->_Responsible  = NULL ;
              m->OwnerIsThread = 0 ;
              m->_recursions   = 0 ;
              m->_SpinDuration = ObjectMonitor::Knob_SpinLimit ;   // Consider: maintain by type/class
    
              markOop cmp = (markOop) Atomic::cmpxchg_ptr (markOopDesc::INFLATING(), object->mark_addr(), mark) ;
              if (cmp != mark) {//CAS失败,说明冲突了,自旋等待
                 omRelease (Self, m, true) ;//释放监视器锁
                 continue ;       // Interference -- just retry
              }
    
              // We've successfully installed INFLATING (0) into the mark-word.
              // This is the only case where 0 will appear in a mark-work.
              // Only the singular thread that successfully swings the mark-word
              // to 0 can perform (or more precisely, complete) inflation.
              //
              // Why do we CAS a 0 into the mark-word instead of just CASing the
              // mark-word from the stack-locked value directly to the new inflated state?
              // Consider what happens when a thread unlocks a stack-locked object.
              // It attempts to use CAS to swing the displaced header value from the
              // on-stack basiclock back into the object header.  Recall also that the
              // header value (hashcode, etc) can reside in (a) the object header, or
              // (b) a displaced header associated with the stack-lock, or (c) a displaced
              // header in an objectMonitor.  The inflate() routine must copy the header
              // value from the basiclock on the owner's stack to the objectMonitor, all
              // the while preserving the hashCode stability invariants.  If the owner
              // decides to release the lock while the value is 0, the unlock will fail
              // and control will eventually pass from slow_exit() to inflate.  The owner
              // will then spin, waiting for the 0 value to disappear.   Put another way,
              // the 0 causes the owner to stall if the owner happens to try to
              // drop the lock (restoring the header from the basiclock to the object)
              // while inflation is in-progress.  This protocol avoids races that might
              // would otherwise permit hashCode values to change or "flicker" for an object.
              // Critically, while object->mark is 0 mark->displaced_mark_helper() is stable.
              // 0 serves as a "BUSY" inflate-in-progress indicator.
    
    
              // fetch the displaced mark from the owner's stack.
              // The owner can't die or unwind past the lock while our INFLATING
              // object is in the mark.  Furthermore the owner can't complete
              // an unlock on the object, either.
              markOop dmw = mark->displaced_mark_helper() ;
              assert (dmw->is_neutral(), "invariant") ;
    
              // Setup monitor fields to proper values -- prepare the monitor
              //CAS成功,设置ObjectMonitor的_header、_owner和_object等
              m->set_header(dmw) ;
    
              // Optimization: if the mark->locker stack address is associated
              // with this thread we could simply set m->_owner = Self and
              // m->OwnerIsThread = 1. Note that a thread can inflate an object
              // that it has stack-locked -- as might happen in wait() -- directly
              // with CAS.  That is, we can avoid the xchg-NULL .... ST idiom.
              m->set_owner(mark->locker());
              m->set_object(object);
              // TODO-FIXME: assert BasicLock->dhw != 0.
    
              // Must preserve store ordering. The monitor state must
              // be stable at the time of publishing the monitor address.
              guarantee (object->mark() == markOopDesc::INFLATING(), "invariant") ;
              object->release_set_mark(markOopDesc::encode(m));
    
              // Hopefully the performance counters are allocated on distinct cache lines
              // to avoid false sharing on MP systems ...
              if (ObjectMonitor::_sync_Inflations != NULL) ObjectMonitor::_sync_Inflations->inc() ;
              TEVENT(Inflate: overwrite stacklock) ;
              if (TraceMonitorInflation) {
                if (object->is_instance()) {
                  ResourceMark rm;
                  tty->print_cr("Inflating object " INTPTR_FORMAT " , mark " INTPTR_FORMAT " , type %s",
                    (void *) object, (intptr_t) object->mark(),
                    object->klass()->external_name());
                }
              }
              return m ;
          }
    
          // CASE: neutral 无锁
          // TODO-FIXME: for entry we currently inflate and then try to CAS _owner.
          // If we know we're inflating for entry it's better to inflate by swinging a
          // pre-locked objectMonitor pointer into the object header.   A successful
          // CAS inflates the object *and* confers ownership to the inflating thread.
          // In the current implementation we use a 2-step mechanism where we CAS()
          // to inflate and then CAS() again to try to swing _owner from NULL to Self.
          // An inflateTry() method that we could call from fast_enter() and slow_enter()
          // would be useful.
    
          assert (mark->is_neutral(), "invariant");
          ObjectMonitor * m = omAlloc (Self) ;
          // prepare m for installation - set monitor to initial state
          m->Recycle();
          m->set_header(mark);
          m->set_owner(NULL);
          m->set_object(object);
          m->OwnerIsThread = 1 ;
          m->_recursions   = 0 ;
          m->_Responsible  = NULL ;
          m->_SpinDuration = ObjectMonitor::Knob_SpinLimit ;       // consider: keep metastats by type/class
    
          if (Atomic::cmpxchg_ptr (markOopDesc::encode(m), object->mark_addr(), mark) != mark) {
              m->set_object (NULL) ;
              m->set_owner  (NULL) ;
              m->OwnerIsThread = 0 ;
              m->Recycle() ;
              omRelease (Self, m, true) ;
              m = NULL ;
              continue ;
              // interference - the markword changed - just retry.
              // The state-transitions are one-way, so there's no chance of
              // live-lock -- "Inflated" is an absorbing state.
          }
    
          // Hopefully the performance counters are allocated on distinct
          // cache lines to avoid false sharing on MP systems ...
          if (ObjectMonitor::_sync_Inflations != NULL) ObjectMonitor::_sync_Inflations->inc() ;
          TEVENT(Inflate: overwrite neutral) ;
          if (TraceMonitorInflation) {
            if (object->is_instance()) {
              ResourceMark rm;
              tty->print_cr("Inflating object " INTPTR_FORMAT " , mark " INTPTR_FORMAT " , type %s",
                (void *) object, (intptr_t) object->mark(),
                object->klass()->external_name());
            }
          }
          return m ;
      }
    }

    膨胀过程的实现比较复杂,大概实现过程如下:
    1、整个膨胀过程在自旋下完成;
    2、mark->has_monitor()方法判断当前是否为重量级锁(上图18-25行),即Mark Word的锁标识位为 10,如果当前状态为重量级锁,执行步骤(3),否则执行步骤(4);
    3、膨胀过程已经完成:mark->monitor()方法获取指向ObjectMonitor的指针,并返回。
    4、当前锁处于膨胀中(上图33-37行),说明该锁正在被其它线程执行膨胀操作,则当前线程就进行自旋等待锁膨胀完成,这里需要注意一点,虽然是自旋操作,但不会一直占用cpu资源,每隔一段时间会通过os::NakedYield方法放弃cpu资源,或通过park方法挂起;如果其他线程完成锁的膨胀操作,则退出自旋并返回;
    5、当前是轻量级锁状态(上图58-138行),即锁标识位为 00,膨胀过程如下:
    通过omAlloc方法,获取一个可用的ObjectMonitor monitor,并重置monitor数据;
    通过CAS尝试将Mark Word设置为markOopDesc:INFLATING,标识当前锁正在膨胀中,如果CAS失败,说明同一时刻其它线程已经将Mark Word设置为markOopDesc:INFLATING,当前线程进行自旋等待膨胀完成;
    如果CAS成功,设置monitor的各个字段:_header、_owner和_object等,并返回;
    6、无锁(中立,上图150-186行),重置监视器值;

    4.2 重量级锁的竞争:ObjectMonitor::enter

    当锁膨胀完成并返回对应的monitor时,并不表示该线程竞争到了锁,真正的锁竞争发生在ObjectMonitor::enter方法中。

    openjdk\hotspot\src\share\vm\runtime\objectMonitor.cpp => ObjectMonitor::enter()

    void ATTR ObjectMonitor::enter(TRAPS) {
      // The following code is ordered to check the most common cases first
      // and to reduce RTS->RTO cache line upgrades on SPARC and IA32 processors.
      Thread * const Self = THREAD ;
      void * cur ;
    
      cur = Atomic::cmpxchg_ptr (Self, &_owner, NULL) ;
      if (cur == NULL) {//CAS成功
         // Either ASSERT _recursions == 0 or explicitly set _recursions = 0.
         assert (_recursions == 0   , "invariant") ;
         assert (_owner      == Self, "invariant") ;
         // CONSIDER: set or assert OwnerIsThread == 1
         return ;
      }
    
      if (cur == Self) {//重入锁
         // TODO-FIXME: check for integer overflow!  BUGID 6557169.
         _recursions ++ ;
         return ;
      }
    
      if (Self->is_lock_owned ((address)cur)) {
        assert (_recursions == 0, "internal state error");
        _recursions = 1 ;
        // Commute owner from a thread-specific on-stack BasicLockObject address to
        // a full-fledged "Thread *".
        _owner = Self ;
        OwnerIsThread = 1 ;
        return ;
      }
    
      // We've encountered genuine contention.
      assert (Self->_Stalled == 0, "invariant") ;
      Self->_Stalled = intptr_t(this) ;
    
      // Try one round of spinning *before* enqueueing Self
      // and before going through the awkward and expensive state
      // transitions.  The following spin is strictly optional ...
      // Note that if we acquire the monitor from an initial spin
      // we forgo posting JVMTI events and firing DTRACE probes.
      if (Knob_SpinEarly && TrySpin (Self) > 0) {
         assert (_owner == Self      , "invariant") ;
         assert (_recursions == 0    , "invariant") ;
         assert (((oop)(object()))->mark() == markOopDesc::encode(this), "invariant") ;
         Self->_Stalled = 0 ;
         return ;
      }
    
      assert (_owner != Self          , "invariant") ;
      assert (_succ  != Self          , "invariant") ;
      assert (Self->is_Java_thread()  , "invariant") ;
      JavaThread * jt = (JavaThread *) Self ;
      assert (!SafepointSynchronize::is_at_safepoint(), "invariant") ;
      assert (jt->thread_state() != _thread_blocked   , "invariant") ;
      assert (this->object() != NULL  , "invariant") ;
      assert (_count >= 0, "invariant") ;
    
      // Prevent deflation at STW-time.  See deflate_idle_monitors() and is_busy().
      // Ensure the object-monitor relationship remains stable while there's contention.
      Atomic::inc_ptr(&_count);
    
      EventJavaMonitorEnter event;
    
      { // Change java thread status to indicate blocked on monitor enter.
        JavaThreadBlockedOnMonitorEnterState jtbmes(jt, this);
    
        DTRACE_MONITOR_PROBE(contended__enter, this, object(), jt);
        if (JvmtiExport::should_post_monitor_contended_enter()) {
          JvmtiExport::post_monitor_contended_enter(jt, this);
        }
    
        OSThreadContendState osts(Self->osthread());
        ThreadBlockInVM tbivm(jt);
    
        Self->set_current_pending_monitor(this);
    
        // TODO-FIXME: change the following for(;;) loop to straight-line code.
        for (;;) {
          jt->set_suspend_equivalent();
          // cleared by handle_special_suspend_equivalent_condition()
          // or java_suspend_self()
    
          EnterI (THREAD) ;
    
          if (!ExitSuspendEquivalent(jt)) break ;
    
          //
          // We have acquired the contended monitor, but while we were
          // waiting another thread suspended us. We don't want to enter
          // the monitor while suspended because that would surprise the
          // thread that suspended us.
          //
              _recursions = 0 ;
          _succ = NULL ;
          exit (false, Self) ;
    
          jt->java_suspend_self();
        }
        Self->set_current_pending_monitor(NULL);
      }
    
      Atomic::dec_ptr(&_count);
      assert (_count >= 0, "invariant") ;
      Self->_Stalled = 0 ;
    
      // Must either set _recursions = 0 or ASSERT _recursions == 0.
      assert (_recursions == 0     , "invariant") ;
      assert (_owner == Self       , "invariant") ;
      assert (_succ  != Self       , "invariant") ;
      assert (((oop)(object()))->mark() == markOopDesc::encode(this), "invariant") ;
    
      // The thread -- now the owner -- is back in vm mode.
      // Report the glorious news via TI,DTrace and jvmstat.
      // The probe effect is non-trivial.  All the reportage occurs
      // while we hold the monitor, increasing the length of the critical
      // section.  Amdahl's parallel speedup law comes vividly into play.
      //
      // Another option might be to aggregate the events (thread local or
      // per-monitor aggregation) and defer reporting until a more opportune
      // time -- such as next time some thread encounters contention but has
      // yet to acquire the lock.  While spinning that thread could
      // spinning we could increment JVMStat counters, etc.
    
      DTRACE_MONITOR_PROBE(contended__entered, this, object(), jt);
      if (JvmtiExport::should_post_monitor_contended_entered()) {
        JvmtiExport::post_monitor_contended_entered(jt, this);
      }
    
      if (event.should_commit()) {
        event.set_klass(((oop)this->object())->klass());
        event.set_previousOwner((TYPE_JAVALANGTHREAD)_previous_owner_tid);
        event.set_address((TYPE_ADDRESS)(uintptr_t)(this->object_addr()));
        event.commit();
      }
    
      if (ObjectMonitor::_sync_ContendedLockAttempts != NULL) {
         ObjectMonitor::_sync_ContendedLockAttempts->inc() ;
      }
    }

    1、通过CAS尝试把monitor的_owner字段设置为当前线程;
    2、如果设置之前的_owner指向当前线程,说明当前线程再次进入monitor,即重入锁,执行_recursions ++ ,记录重入的次数;
    3、如果之前的_owner指向的地址在当前线程中,这种描述有点拗口,换一种说法:之前_owner指向的BasicLock在当前线程栈上,说明当前线程是第一次进入该monitor,设置_recursions为1,_owner为当前线程,该线程成功获得锁并返回;
    4、如果获取锁失败,则等待锁的释放;

    4.3. monitor等待

    monitor竞争失败的线程,通过自旋执行ObjectMonitor::EnterI方法等待锁的释放,EnterI方法的部分逻辑实现如下:

    ObjectWaiter node(Self) ;
        Self->_ParkEvent->reset() ;
        node._prev   = (ObjectWaiter *) 0xBAD ;
        node.TState  = ObjectWaiter::TS_CXQ ;
    
        // Push "Self" onto the front of the _cxq.
        // Once on cxq/EntryList, Self stays on-queue until it acquires the lock.
        // Note that spinning tends to reduce the rate at which threads
        // enqueue and dequeue on EntryList|cxq.
        ObjectWaiter * nxt ;
        for (;;) {
            node._next = nxt = _cxq ;
            if (Atomic::cmpxchg_ptr (&node, &_cxq, nxt) == nxt) break ;
    
            // Interference - the CAS failed because _cxq changed.  Just retry.
            // As an optional optimization we retry the lock.
            if (TryLock (Self) > 0) {
                assert (_succ != Self         , "invariant") ;
                assert (_owner == Self        , "invariant") ;
                assert (_Responsible != Self  , "invariant") ;
                return ;
            }
        }

    1、当前线程被封装成ObjectWaiter对象node,状态设置成ObjectWaiter::TS_CXQ;
    2、在for循环中,通过CAS把node节点push到_cxq列表中,同一时刻可能有多个线程把自己的node节点push到_cxq列表中;
    3、node节点push到_cxq列表之后,通过自旋尝试获取锁,如果还是没有获取到锁,则通过park将当前线程挂起,等待被唤醒,实现如下:

    for (;;) {
    
            if (TryLock (Self) > 0) break ;
            assert (_owner != Self, "invariant") ;
    
            if ((SyncFlags & 2) && _Responsible == NULL) {
               Atomic::cmpxchg_ptr (Self, &_Responsible, NULL) ;
            }
    
            // park self
            if (_Responsible == Self || (SyncFlags & 1)) {
                TEVENT (Inflated enter - park TIMED) ;
                Self->_ParkEvent->park ((jlong) RecheckInterval) ;
                // Increase the RecheckInterval, but clamp the value.
                RecheckInterval *= 8 ;
                if (RecheckInterval > 1000) RecheckInterval = 1000 ;
            } else {
                TEVENT (Inflated enter - park UNTIMED) ;
                Self->_ParkEvent->park() ;//当前线程挂起
            }
    
            if (TryLock(Self) > 0) break ;
    
            // The lock is still contested.
            // Keep a tally of the # of futile wakeups.
            // Note that the counter is not protected by a lock or updated by atomics.
            // That is by design - we trade "lossy" counters which are exposed to
            // races during updates for a lower probe effect.
            TEVENT (Inflated enter - Futile wakeup) ;
            if (ObjectMonitor::_sync_FutileWakeups != NULL) {
               ObjectMonitor::_sync_FutileWakeups->inc() ;
            }
            ++ nWakeups ;
    
            // Assuming this is not a spurious wakeup we'll normally find _succ == Self.
            // We can defer clearing _succ until after the spin completes
            // TrySpin() must tolerate being called with _succ == Self.
            // Try yet another round of adaptive spinning.
            if ((Knob_SpinAfterFutile & 1) && TrySpin (Self) > 0) break ;
    
            // We can find that we were unpark()ed and redesignated _succ while
            // we were spinning.  That's harmless.  If we iterate and call park(),
            // park() will consume the event and return immediately and we'll
            // just spin again.  This pattern can repeat, leaving _succ to simply
            // spin on a CPU.  Enable Knob_ResetEvent to clear pending unparks().
            // Alternately, we can sample fired() here, and if set, forgo spinning
            // in the next iteration.
    
            if ((Knob_ResetEvent & 1) && Self->_ParkEvent->fired()) {
               Self->_ParkEvent->reset() ;
               OrderAccess::fence() ;
            }
            if (_succ == Self) _succ = NULL ;
    
            // Invariant: after clearing _succ a thread *must* retry _owner before parking.
            OrderAccess::fence() ;
        }

    4、当该线程被唤醒时,会从挂起的点继续执行,通过ObjectMonitor::TryLock尝试获取锁,TryLock方法实现如下:

    int ObjectMonitor::TryLock (Thread * Self) {
       for (;;) {
          void * own = _owner ;
          if (own != NULL) return 0 ;
          if (Atomic::cmpxchg_ptr (Self, &_owner, NULL) == NULL) {//CAS成功,获取锁
             // Either guarantee _recursions == 0 or set _recursions = 0.
             assert (_recursions == 0, "invariant") ;
             assert (_owner == Self, "invariant") ;
             // CONSIDER: set or assert that OwnerIsThread == 1
             return 1 ;
          }
          // The lock had been free momentarily, but we lost the race to the lock.
          // Interference -- the CAS failed.
          // We can either return -1 or retry.
          // Retry doesn't make as much sense because the lock was just acquired.
          if (true) return -1 ;
       }
    }

    其本质就是通过CAS设置monitor的_owner字段为当前线程,如果CAS成功,则表示该线程获取了锁,跳出自旋操作,执行同步代码,否则继续被挂起;

    4.4. monitor释放

    当某个持有锁的线程执行完同步代码块时,会进行锁的释放,给其它线程机会执行同步代码,在HotSpot中,通过退出monitor的方式实现锁的释放,并通知被阻塞的线程,具体实现位于ObjectMonitor::exit方法中。

    void ATTR ObjectMonitor::exit(bool not_suspended, TRAPS) {
       Thread * Self = THREAD ;
       if (THREAD != _owner) {
         if (THREAD->is_lock_owned((address) _owner)) {
           // Transmute _owner from a BasicLock pointer to a Thread address.
           // We don't need to hold _mutex for this transition.
           // Non-null to Non-null is safe as long as all readers can
           // tolerate either flavor.
           assert (_recursions == 0, "invariant") ;
           _owner = THREAD ;
           _recursions = 0 ;
           OwnerIsThread = 1 ;
         } else {
           // NOTE: we need to handle unbalanced monitor enter/exit
           // in native code by throwing an exception.
           // TODO: Throw an IllegalMonitorStateException ?
           TEVENT (Exit - Throw IMSX) ;
           assert(false, "Non-balanced monitor enter/exit!");
           if (false) {
              THROW(vmSymbols::java_lang_IllegalMonitorStateException());
           }
           return;
         }
       }
    
       if (_recursions != 0) {
         _recursions--;        // this is simple recursive enter
         TEVENT (Inflated exit - recursive) ;
         return ;
       }
    ...省略...

    1、如果是重量级锁的释放,monitor中的_owner指向当前线程,即THREAD == _owner;
    2、根据不同的策略(由QMode指定),从cxq或EntryList中获取头节点,通过ObjectMonitor::ExitEpilog方法唤醒该节点封装的线程,唤醒操作最终由unpark完成,实现如下:

    void ObjectMonitor::ExitEpilog (Thread * Self, ObjectWaiter * Wakee) {
       assert (_owner == Self, "invariant") ;
    
       // Exit protocol:
       // 1. ST _succ = wakee
       // 2. membar #loadstore|#storestore;
       // 2. ST _owner = NULL
       // 3. unpark(wakee)
    
       _succ = Knob_SuccEnabled ? Wakee->_thread : NULL ;
       ParkEvent * Trigger = Wakee->_event ;
    
       // Hygiene -- once we've set _owner = NULL we can't safely dereference Wakee again.
       // The thread associated with Wakee may have grabbed the lock and "Wakee" may be
       // out-of-scope (non-extant).
       Wakee  = NULL ;
    
       // Drop the lock
       OrderAccess::release_store_ptr (&_owner, NULL) ;
       OrderAccess::fence() ;                               // ST _owner vs LD in unpark()
    
       if (SafepointSynchronize::do_call_back()) {
          TEVENT (unpark before SAFEPOINT) ;
       }
    
       DTRACE_MONITOR_PROBE(contended__exit, this, object(), Self);
       Trigger->unpark() ;
    
       // Maintain stats and report events to JVMTI
       if (ObjectMonitor::_sync_Parks != NULL) {
          ObjectMonitor::_sync_Parks->inc() ;
       }
    }

    3、被唤醒的线程,继续执行monitor的竞争;

    四、总结

    直接看代码虽然可以窥探加锁流程的细节,但是源代码由于复合了对象的多个状态(无锁、偏向、轻量级、重量级)并对其进行了很多优化,所以我们并不能很直观的了解到各个锁状态之间的转换。所以整理出了下面的状态机图帮助理解:

    获取锁的流程

    本文代码解析部分参考了 jdk源码剖析二: 对象内存布局、synchronized原理,并修正了其中的部分错误。
    其他参考资料:
    深入JVM锁机制1-synchronized
    Java Language Specification
    转载本文请注明出处https://blog.csdn.net/qq_29753285/article/details/81299509

    展开全文
  • CAS(Compare And Swap 比较并且替换)是乐观的一种实现方式,是一种轻量级,JUC 中很多工具类的实现就是基于 CAS 的,也可以理解为自旋 JUC是指import java.util.concurrent下面的包, 比如:import java....
  • Java CAS底层实现原理实例详解

    千次阅读 2020-07-18 15:39:24
    CAS,全称Compare And Swap(比较与交换),解决多线程并行情况下使用造成性能损耗的一种机制。 CAS(V, A, B),V为内存地址、A为预期原值,B为新值。如果内存地址的值与预期原值相匹配,那么将该位置值更新为新...
  • 接下里将会解释同步代码块、同步方法、全局底层实现。 同步代码块底层实现: 首先看一个简单的同步代码块: /////同步代码块底层实现 public class DiCeng { public static void main(String[] args) ...
  • java锁底层原理

    千次阅读 2019-05-09 20:46:46
    底层 升级过程、CAS操作的缺点【替换线程和copy mw】 优化 代码优化:同步代码块、减少粒度、读并发 JDK自带 偏置、轻量级(CAS操作)、自适应自旋、粗化、消除 Volatile 概念:非...
  • lock底层实现

    千次阅读 2018-04-15 18:02:42
    这当然取决于它们的底层实现,所以今天我们就先来看一下lock的底层原理吧,因为lock底层实现这块内容比较多,所以sychronized我会再写一篇博文和大家分享~ Lock通过Java编写,与JVM实现无关。 在J.U.C....
  • java lock的底层实现原理

    千次阅读 2016-10-06 19:08:39
    关于java lock的底层实现原理,讲的有点深,转载学习! Lock完全用Java写成,在java这个层面是无关JVM实现的。 在java.util.concurrent.locks包中有很多Lock的实现类,常用的有ReentrantLock、ReadWriteLock...
  • 大家当做简介看好了。下拉正文。...3、Java内存模型(JMM) 4、Java内存模型带来的问题 4.1 可见性问题 4.2 竞争问题 4.3 重排序 5、volatile详解 5.1 volatile特性 5.2 volatile的内存语义 5.3 ...
  • Redisson的分布式可重入读写RReadWriteLock Java对象实现java.util.concurrent.locks.ReadWriteLock接口。同时还支持自动过期解锁。该对象允许同时有多个读取,但是最多只能有一个写。写是排它,获取写...
  • 介绍了synchronized关键字实现锁底层原理以及JDK对于synchronized做出的升级优化!
  • ​ volatile关键字在Java中多线程编程中作为必不可少的关键字,它的作用和原理你知道多少?在我们线程之间通信有很多种方式,它主要是作用在什么方式中呢?在这种通信方式中它是通过什么方式来实现线程之间的数据...
  • Redisson单进程Redis分布式悲观的使用与实现 本文基于Redisson 3.7.5 2. 公平 这种的使用方式和Java本身框架中的FairLock一模一样: RLock fairLock = redisson.getFairLock(&...

空空如也

空空如也

1 2 3 4 5 ... 20
收藏数 123,251
精华内容 49,300
关键字:

java锁的底层实现

java 订阅