synchronized和ReetrantLock

2020-12-08 本文已影响0人 HannahLi_9f1c

这篇文章主要是内容是剖析synchronized和ReetrantLock，并将这两种锁进行对比。这样我们在使用的时候，就能知道什么场景下选择哪种锁了。
文章结构如下：

文章结构

synchronized

原理

synchronized关键字经过Javac编译，会产生monitorenter和montinorexit两条指令。他们都需要绑定到一个对象上，如果synchronized中指定了要锁住的对象，那么就以对象引用作为绑定对象，如果没有指定，那么就根据修饰的方法的类型是静态方法还是实例方法，来选择绑定类的Class对象还是对象实例。注意，对于方法来说，不会生成monitorenter和montinorexit，而是会通过ACC_SYNCHRONIZED访问标志。
加锁过程：

《深入理解Java虚拟机》
在执行montiorenter时，首先尝试去获取对象的锁。如果对象没被锁定，或者当前线程已经拥有对象的锁，就把锁的计数器加一，而在执行montiorexit指令时会被锁计数器减一。一旦计数器减为0，锁就会被释放。如果获取对象锁失败，当前线程就会阻塞等待，直到请求锁定的对象被持有它的线程释放为止

监视器锁

上面说到在执行monitorenter时，首先会尝试去获取对象的锁，那么对象是怎么跟锁绑定在一起的呢？答案就是对象头的锁指针。

对象头

通过上图我们可以看到
Java在获取到轻量级或者重量级锁之后，会将锁标志位修改，并且利用对象存储hashcode、分代年龄的空间来存储锁的指针（我觉得这里挺奇怪的，这里占用了对象的HashCode和分代年龄，HashCode被占用还可以再算，分代年龄被占了，岂不是没法回收了）。重量级指针指向的时monitor对象，是由虚拟机用C++实现的。monitor是由ObjectMonitor实现的，结构如下：

ObjectMonitor() {
    _header       = NULL;
    _count        = 0; //记录个数
    _waiters      = 0,
    _recursions   = 0;
    _object       = NULL;
    _owner        = NULL;  //_owner指向持有ObjectMonitor对象的线程
    _WaitSet      = NULL; //处于wait状态的线程，会被加入到_WaitSet
    _WaitSetLock  = 0 ;
    _Responsible  = NULL ;
    _succ         = NULL ;
    _cxq          = NULL ;
    FreeNext      = NULL ;
    _EntryList    = NULL ; //处于等待锁block状态的线程，会被加入到该列表
    _SpinFreq     = 0 ;
    _SpinClock    = 0 ;
    OwnerIsThread = 0 ;
  }

ObjectMonitor里比较重要的数据结构是_WaitSet 和 _EntryList，多线程竞争锁，发现对象锁montior已经被其他线程占有之后，会进入_EntryList，当线程获取到对象的monitor 后进入 _Owner 区域并把monitor中的owner变量设置为当前线程同时monitor中的计数器count加1。当线程调用wait，那么会进入条件队列_WaitSet 进行等待，当获取到锁的线程调用notify时，_WaitSet的随机一个线程会进入到_EntryList，_EntryList中的线程当对象被释放之后就可以竞争对象了。而获取锁的线程调用notifyAll，会将_WaitSet的所有线程会转移到_EntryList，共同竞争资源。
从这里可以发现sync和Lock的原理都是多线程竞争锁，竞争不到就进同步队列等待。另外Lock也有条件队列，可以通过lock.newCondition生成。

锁升级

Jdk1.6对于重量级锁进行了优化，为什么要优化呢？是因为Java中的线程是映射到操作系统内核线程上的，因此阻塞和唤醒线程，都需要操作系统的帮助，阻塞和唤醒都涉及到用户态和内核态之间的转换，用户态和内核态的转换不仅仅保存当前代码数据，还需要上下文，这里的上下文包括寄存器和程序计数器，上下文的保存和恢复的时间和空间的开销都很大，所以说synchronized开销较大，因此在JDK1.6之前性能远不如ReetrantLock。

自旋锁、自适应锁：我们已经知道当一个线程无法获取锁，需要阻塞就会进入内核态，但是很多时候线程占用锁的时间比较短，那么这个线程经过短时间的阻塞又要被唤醒了，在短时间内在内核态和用户态之间切换，开销会很大。自旋锁在线程获取不到锁的时候，先进行空循环，占用CPU时间等待锁，如果能够等到锁释放，就能够避免用户态和内核态的切换了。而自适应锁是通过判断锁前一次在同一个锁的自旋时间和拥有者状态来决定是否自旋和自旋时间，如果在一个锁上很少能够自旋等待成功，那么虚拟机会选择不在这个锁上进行自旋。

自旋锁过程

自旋锁的实现原理同样也是CAS，AtomicInteger中调用unsafe进行自增操作的源码中的do-while循环就是一个自旋操作，如果修改数值失败则通过循环来执行自旋，直至修改成功。

AtomicInteger增加
锁消除：锁消除就是通过逃逸分析，如果判断一个一段代码在堆上的数据不会被其他线程访问到，那么可以把这段代码当成栈数据来使用，不需要锁
锁粗化：如果在同一个对象上连续进行加锁解锁，消耗也很大，锁粗化就是把同步范围变大，避免反复加锁解锁。
轻量级锁：
image
在进入同步块之前，对象是无锁的状态，锁标志位是01，虚拟机在当前线程的栈帧建立一个Lock Record存储Mark Word的拷贝。然后虚拟机利用CAS尝试将对象的Mark Word更新为指向Lock Record的指针，如果成功，表示加锁成功，Mark Word标志位变为00。如果CAS失败，虚拟机会检查对象的Mark Word是否已变为当前线程的栈帧，如果不是，说明其他线程已经占有了该对象。此时就会膨胀为重量级锁，标志位变为10。解锁的过程也是使用CAS。轻量级锁提升性能的依据就是绝大部分锁，在整个同步周期，是不存在竞争的
偏向锁：如果当前虚拟机开启了偏向锁，那么当锁对象第一次被线程获取的时候，会把对象头的标志位设置为“01”，偏向标志记为1，并用CAS把当前线程的ID记录在对象头。下次这个线程再次获取这个对象锁时，就不需要同步了。一旦其他线程尝试获取这个锁，偏向模式就结束了

锁升级过程

如果开启了偏向锁，第一个尝试获取锁的线程会通过CAS将对象头的线程id设置为自己的线程ID，以后这个线程再次进入临界区只需要简单地判断是否偏向了自己就可以进入同步代码。第二个线程进入同步代码块的时候，如果第一个线程没有存活，那就可以撤销偏向，将对象头的线程id更新为第二个线程，但是如果第一个线程仍然存活，就需要进行锁升级，锁升级之前需要锁撤销，锁撤销过程如下：

在安全点停止拥有锁的线程（什么是安全点？）
遍历线程栈，将锁记录修复为无锁状态。
唤醒第一个线程，使其升级为轻量级锁
轻量级锁的升级过程如下：

轻量级锁设置
线程在栈帧中创建锁记录LockRecord
将锁对象的Mark Word复制到LockRecord
将锁记录的Owner设置为锁对象
CAS操作将对象头的锁指针指向锁记录（我觉得后两步才是CAS操作吧？）
轻量锁分为两种，自旋锁和自适应自旋锁。就是获取不到锁的其他线程会一直自旋忙等待，当自旋超过一定次数，轻量级锁会膨胀为重量级锁，重量级锁把除了拥有锁的线程都阻塞，然后再改变对象头，设置为重量级锁。

锁升级
竞争

ReetrantLock

原理

ReetrantLock是基于AQS实现的独占方式获取锁，线程先尝试获取锁，如果失败就进入队列，进入队列之后设置前向未被取消的节点，确保前向节点获取资源的时候会唤醒当前线程，然后当前线程调用park进入休眠，进入线程的WAITING状态。具体源码的执行过程可以参考ReetrantLock源码分析

公平锁vs非公平锁

ReetrantLock的公平锁和非公平锁实现是通过hasQueuedPredecessors来实现的，这个方法判断目的是判断同步队列中，是否已经有正在等待资源的其他线程。

公平锁，非公平锁对比

判断是否有其他节点等待

非阻塞获取锁

ReetrantLock的非阻塞方式获取锁是通过tryLock实现的，具体是调用了nonfairTryAcquire方法，先通过CAS获取资源，获取成功或者可重入线程独占的锁返回true，否则返回false，不会调用acquire进入同步队列等待。

  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;
        }

等待可中断、超时等待

等待可中断和超时等待是通过tryLock(long timeout,TimeUnit unit)实现

   public boolean tryLock(long timeout, TimeUnit unit)
            throws InterruptedException {
        return sync.tryAcquireNanos(1, unit.toNanos(timeout));
    }
    public final boolean tryAcquireNanos(int arg, long nanosTimeout)
            throws InterruptedException {
        if (Thread.interrupted())
            throw new InterruptedException();
        return tryAcquire(arg) ||
            doAcquireNanos(arg, nanosTimeout);
    }

tryAcquireNanos方法先是看能不能成功获取到锁，获取不到就调用doAcquireNanos，超时获取锁。doAcquireNanos源码可以看出是通过 LockSupport.parkNanos(this, nanosTimeout);进行超时等待的，每次for循环等待一段时间，并且响应中断。

  private boolean doAcquireNanos(int arg, long nanosTimeout)
            throws InterruptedException {
        if (nanosTimeout <= 0L)
            return false;
        final long deadline = System.nanoTime() + nanosTimeout;
        final Node node = addWaiter(Node.EXCLUSIVE);
        boolean failed = true;
        try {
            for (;;) {
                final Node p = node.predecessor();
                if (p == head && tryAcquire(arg)) {
                    setHead(node);
                    p.next = null; // help GC
                    failed = false;
                    return true;
                }
                nanosTimeout = deadline - System.nanoTime();
                if (nanosTimeout <= 0L)
                    return false;
                if (shouldParkAfterFailedAcquire(p, node) &&
                    nanosTimeout > spinForTimeoutThreshold)
                    LockSupport.parkNanos(this, nanosTimeout);
                if (Thread.interrupted())
                    throw new InterruptedException();
            }
        } finally {
            if (failed)
                cancelAcquire(node);
        }
    }