多线程与高并发——并发编程（3）

三、锁

1 锁的分类

1.1 可重入锁、不可重入锁

Java 中提供的 synchronized、ReentrantLock、ReentrantReadWriteLock 都是重复锁。

• 重入：当前线程获取到A锁，在获取之后尝试再次获取 A 锁，是可以直接拿到的。
• 不可重入：当前线程获取到A锁，在获取之后尝试再次获取 A 锁，无法获取到。因为 A 锁被当前线程占用着，需要等待自己释放锁，再获取锁。

1.2 乐观锁、悲观锁

Java 中提供的 synchronized、ReentrantLock、ReentrantReadWriteLock 都是悲观锁。

Java 中提供的 CAS 操作，就是乐观锁的一种实现。

• 悲观锁：获取不到锁资源时，会将当前线程挂起（进入 BLOCKED、WAITING）。线程挂起会涉及到用户态和内核态之间的切换，而这种切换是比较消耗资源的。
  • 用户态：JVM 可以自行执行的指令，不需要借助操作系统执行。
  • 内核态：JVM 不可以自行执行，需要操作系统才可以执行。
• 乐观锁：获取不到锁资源，可以再次让 CPU 调度，重新尝试获取锁资源。
  • Atomic 原子类中，就是基于 CAS 乐观锁实现的。

1.3 公平锁、非公平锁

Java 中提供的 synchronized 只能是非公平锁；Java 中提供的 ReentrantLock、ReentrantReadWriteLock 可以实现公平锁和非公平锁。

• 公平锁：线程A获取到了锁资源，线程B没有拿到锁资源，则线程B去排队；线程C来了，锁被A持有，同时线程B在排队，则C直接排到B的后面，等待B拿到资源或者B被取消后，才可以尝试去竞争锁资源。
• 非公平锁：线程A获取到了锁资源，线程B没有拿到锁资源，则线程B去排队；线程C来了，先尝试竞争一波：
  • 拿到锁资源：插队成功
  • 没有拿到锁资源：依然要排到B的后面，等待B拿到资源或者B被取消后，才可以尝试去竞争锁资源。

1.4 互斥锁、共享锁

Java 中提供的 synchronized、ReentrantLock 是互斥锁，Java 中提供的 ReentrantReadWriteLock 有互斥锁、也有共享锁。

• 互斥锁：同一时间点，只会有一个线程持有当前互斥锁。
• 共享锁：同一时间点，当前共享锁可以被多个线程同时持有。

2 深入synchronized

2.1 类锁、对象锁

synchronized 的使用一般就是同步方法或同步代码块，synchronized 的锁是基于对象实现的。如果使用同步方法：

• static：此时使用是当前 类.class 作为锁（类锁）
• 非static：此时使用是当前对象做为锁

public class MyTest {public static void main(String[] args) {Test.a(); // 锁的是当前Test.classTest test = new Test();test.b(); // 锁的是new出来的test对象}
}
class Test {public static synchronized void a() {System.out.println("111");}public synchronized void b() {System.out.println("222");}
}

2.2 synchronized的优化

在 JDK1.5的时候，Doug Lee 推出了 ReentrantLock，Lock 的性能远高于 synchronized，所以 JDK 团队就在 JDK 1.6 中，对 synchronized 做了大量优化。

• 锁消除：在 synchronized 修饰的代码中，如果不存在操作临界资源的情况，会触发锁消除，即便是写了 synchronized 也不会触发。

public synchronized void method() {// 没有操作临界资源// 此时这个方法的 synchronized 可以任务是没有
}

• 锁膨胀：如果在一个循环中 ，频繁的获取和释放锁资源，这样带来的消耗很大，锁膨胀就会将锁的范围扩大，避免频繁地竞争和获取锁资源带来不必要的消耗。

public synchronized void method() {for (int i = 0; i < 999999; i++) {synchronized (对象) {}}// 这时，上面的代码会触发锁膨胀synchronized (对象) {for (int i = 0; i < 999999; i++) {}}
}

• 锁升级：ReentrantLock的实现，是先基于乐观锁CAS尝试获取锁资源，如果拿不到锁资源，才会挂起线程。synchronized 在 JDK1.6 之前，完全就是获取不到锁，就立即挂起当前线程，所以 synchronized 性能比较差。synchronized 在 JDK1.6 做了锁升级的优化：
  • 无锁、匿名偏向：当前对象没有作为锁存在
  • 偏向锁：如果当前锁资源，只有一个线程在频繁的获取和释放，那么这个线程过来，只需要判断，当前指向的线程是否是当前线程。
    • 如果是，直接拿着锁资源走；
    • 如果不是，基于 CAS 的方式，尝试将偏向锁指向当前线程。如果获取不到，触发锁升级，升级为轻量级锁。（偏向锁状态出现了锁竞争的情况）
  • 轻量级锁：会采用自旋锁的方式去频繁的以 CAS 的形式获取锁资源（采用自适应自旋锁）
    • 如果成功获取到，拿着锁资源走；
    • 如果自旋了一定次数，没有拿到锁资源，触发锁升级。
  • 重量级锁：就是最传统的 synchronized 方法，拿不到锁资源，就挂起当前线程。（用户态与内核态切换）

2.3 synchronized实现原理

• synchronized 是基于对象实现的，先要对 Java 中对象在堆内存中的存储有一个了解。

展开 MarkWord

• MarkWord 中标记着四种锁的信息：无锁、偏向锁、轻量级锁、重量级锁。

2.4 synchronized的锁升级

为了可以在 Java 中看到对象头的 MarkWord 信息，需要导入依赖

<dependency><groupId>org.openjdk.jol</groupId><artifactId>jol-core</artifactId><version>0.9</version>
</dependency>

锁默认情况下，开启了偏向锁延迟。

偏向锁在升级为轻量级锁时，会涉及到偏向锁撤销，需要等到一个安全点（STW），才可以做偏向锁撤销，在明知道并发情况下，就可以选择不开启偏向锁，或者是设置偏向锁延迟开启。

因为 JVM 在启动时，需要加载大量的 .class 文件到内存中，这个操作会涉及到 synchronized 的使用，为了避免出现偏向锁撤销操作，JVM 启动初期，有一个延迟 4s 开启偏向锁的操作。

如果正常开启偏向锁了，那么不会出现无锁的状态，对象会直接变为匿名偏向。

public static void main(String[] args) throws InterruptedException {Object x = new Object();System.out.println(ClassLayout.parseInstance(x).toPrintable());Thread.sleep(5000);Object o = new Object();System.out.println(ClassLayout.parseInstance(o).toPrintable());new Thread(() -> {synchronized (o) {// t1: 偏向锁System.out.println("t1:" + ClassLayout.parseInstance(o).toPrintable());}}).start();// main: 偏向锁 -> 轻量级锁CAS -> 重量级锁synchronized (o) {// 将下面注释打开，main、t1 升级为重量级锁，
//            System.out.println("main:" + ClassLayout.parseInstance(o).toPrintable());}
}

整个锁升级状态的转变：

Lock Record 以及 ObjectMonitor 存储的内容

2.5 重量级锁底层 ObjectMonitor

需要去找到 openjdk，在百度中直接搜索 openjdk，第一个链接就是，找到 ObjectMonitor 的两个文件：hpp、cpp

• 先查看核心属性：https://hg.openjdk.org/jdk8u/jdk8u/hotspot/file/69087d08d473/src/share/vm/runtime/objectMonitor.hpp

ObjectMonitor() {_header       = NULL;   // header存储着MarkWord_count        = 0;      // 竞争锁的线程个数_waiters      = 0,      // wait的线程个数_recursions   = 0;      // 标识当前synchronized锁重入的次数_object       = NULL;_owner        = NULL;   // 持有锁的线程_WaitSet      = NULL;   // 保存wait的线程信息，双向链表_WaitSetLock  = 0 ;_Responsible  = NULL ;_succ         = NULL ;_cxq          = NULL ;  // 获取锁资源失败后，线程要放到当前的单向链表中FreeNext      = NULL ;_EntryList    = NULL ;  // _cxq以及被唤醒的WaitSet中的线程，在一定机制下，会放到EntryList中_SpinFreq     = 0 ;_SpinClock    = 0 ;OwnerIsThread = 0 ;_previous_owner_tid = 0;
}

• 适当的查看几个 C++ 中实现的加锁流程：https://hg.openjdk.org/jdk8u/jdk8u/hotspot/file/69087d08d473/src/share/vm/runtime/objectMonitor.cpp

TryLock

int ObjectMonitor::TryLock (Thread * Self) {for (;;) {// 拿到持有锁的线程void * own = _owner ;// 如果有线程持有锁，告辞if (own != NULL) return 0 ;// 说明没有线程持有锁，own是null，cmpxchg指令就是底层的CAS实现。if (Atomic::cmpxchg_ptr (Self, &_owner, NULL) == NULL) {// 成功获取锁资源return 1 ;}// 这里其实重试操作没什么意义，直接返回-1if (true) return -1 ;}
}

try_entry

bool ObjectMonitor::try_enter(Thread* THREAD) {// 判断_owner是不是当前线程if (THREAD != _owner) {// 判断当前持有锁的线程是否是当前线程，说明轻量级锁刚刚升级过来的情况if (THREAD->is_lock_owned ((address)_owner)) {_owner = THREAD ;_recursions = 1 ;OwnerIsThread = 1 ;return true;}// CAS操作，尝试获取锁资源if (Atomic::cmpxchg_ptr (THREAD, &_owner, NULL) != NULL) {// 没拿到锁资源，告辞return false;}// 拿到锁资源return true;} else {// 将_recursions + 1，代表锁重入操作。_recursions++;return true;}
}

enter：想方设法拿到锁资源，如果没拿到，则挂起扔到 _cxq 单向链表中

void ATTR ObjectMonitor::enter(TRAPS) {// 拿到当前线程Thread * const Self = THREAD ;void * cur ;// CAS走你，cur = Atomic::cmpxchg_ptr (Self, &_owner, NULL) ;if (cur == NULL) {// 拿锁成功return ;}// 锁重入操作if (cur == Self) {// TODO-FIXME: check for integer overflow!  BUGID 6557169._recursions ++ ;return ;}//轻量级锁过来的。if (Self->is_lock_owned ((address)cur)) {_recursions = 1 ;_owner = Self ;OwnerIsThread = 1 ;return ;}// 走到这了，没拿到锁资源，count++Atomic::inc_ptr(&_count);for (;;) {jt->set_suspend_equivalent();// 入队操作，进到cxq中EnterI (THREAD) ;if (!ExitSuspendEquivalent(jt)) break ;_recursions = 0 ;_succ = NULL ;exit (false, Self) ;jt->java_suspend_self();}}// count--Atomic::dec_ptr(&_count);}

EnterI

for (;;) {// 入队node._next = nxt = _cxq ;// CAS的方式入队。if (Atomic::cmpxchg_ptr (&node, &_cxq, nxt) == nxt) break ;// 重新尝试获取锁资源if (TryLock (Self) > 0) {assert (_succ != Self         , "invariant") ;assert (_owner == Self        , "invariant") ;assert (_Responsible != Self  , "invariant") ;return ;}
}

3 深入ReentrantLock

3.1 ReentrantLock和synchronized的区别

核心区别：ReentrantLock是一个类，synchronized是关键字，当然都是在JVM层面实现互斥锁的方式。

效率区别：如果竞争比较激烈，推荐使用ReentrantLock去实现，不存在锁升级概念。而synchronized是存在锁升级概念的，如果升级到重量级锁，是不存在锁降级的。

底层实现区别：实现原理不一样，ReentrantLock基于AQS实现的，synchronized是基于ObjectMonitor实现的。

功能项区别：ReentrantLock的功能比synchronized更全面

• ReentrantLock支持公平锁和非公平锁，ReentrantLock可以指定等待锁资源的时间。

选择哪个？

• 如果你对并发编程特别熟练，推荐使用 ReentrantLock，功能更加丰富；如果掌握一般般，使用 synchronized会更好。

3.2 AQS概述

AQS就是AbstractQueuedSynchronizer抽象类，AQS其实就是JUC包下的一个基类，JUC下的很多内容都是基于AQS实现了部分功能，比如ReentrantLock、ThreadPoolExecutor、阻塞队列、CountDownLatch、Semaphore、CyclicBarrier 等待都是基于AQS实现的。

首先AQS中提供了一个由volatile修饰，并且采用CAS方式修改的int类型的 state 变量；其次AQS中维护了一个双向链表，有 head、有 tail，并且每个节点都是 Node 对象。

static final class Node {static final Node SHARED = new Node();static final Node EXCLUSIVE = null;static final int CANCELLED =  1;static final int SIGNAL    = -1;static final int CONDITION = -2;static final int PROPAGATE = -3;volatile int waitStatus;volatile Node prev;volatile Node next;volatile Thread thread;Node nextWaiter;....
}

AQS内部结构和属性：

3.3 加锁流程源码剖析

3.3.1 加锁流程概述

下面是非公平锁的流程：

• 线程A先执行CAS，将state从0修改为1，线程A就获取到了锁资源，去执行业务代码即可；
• 线程B再执行CAS，发现state已经是1，无法获取到锁资源；
• 线程B需要去排队，将自己封装为 Node 对象；
• 需要将当前B线程的 Node 放到双向链表保存，排队
  • 但是双向链表中，必须先有一个伪节点作为头节点，并且放到双向队列中；
  • 将B线程的Node挂在tail的后面，并且将上一个节点的状态修改为-1，再挂起B线程。

3.3.2 三种加锁源码分析

3.3.2.1 lock方法

• 执行lock方法后，公平锁和非公平锁的执行套路不一样

// 公平锁
final void lock() {// 执行acquire，尝试获取锁资源acquire(1);
}
// 非公平锁
final void lock() {// 上来就基于CAS方法，尝试将state从0改为1if (compareAndSetState(0, 1))// 获取锁资源成功，会将当前线程设置到exclusiveOwnerThread属性，代表是当前线程持有着锁资源setExclusiveOwnerThread(Thread.currentThread());else// 执行acquire，尝试获取锁资源acquire(1);
}

• acquire方法：公平锁和非公平锁的逻辑一样

public final void acquire(int arg) {// tryAcquire: 再次查看，当前线程是否可以尝试获取锁资源if (!tryAcquire(arg) &&// 没有拿到锁资源// addWaiter(Node.EXCLUSIVE): 将当前线程封装为Node节点，插入到AQS的双向链表的结尾// acquireQueued: 查看我是否是第一个排队的节点，如果是，可以再次尝试获取锁资源，如果长时间拿不到，则挂起线程；如果不是第一个排队的节点，就尝试挂起线程即可acquireQueued(addWaiter(Node.EXCLUSIVE), arg))// 中断线程操作selfInterrupt();
}

• tryAcquire方法：竞争锁资源的逻辑，分为公平锁和非公平锁实现

// 非公平锁
final boolean nonfairTryAcquire(int acquires) {// 获取当前线程final Thread current = Thread.currentThread();// 获取state属性int c = getState();// 判断state是否为0，是0则表示之前持有锁的线程释放了锁资源if (c == 0) {// 再抢一波锁资源if (compareAndSetState(0, acquires)) {setExclusiveOwnerThread(current);// 拿锁成功返回truereturn true;}}// 不是0，即有线程持有锁。那么是不是自己？// 如果是，证明是重入锁操作else if (current == getExclusiveOwnerThread()) {// 将state+1int nextc = c + acquires;if (nextc < 0) // overflow: 说明对重入次数+1后，超过了int正数的取值范围// 01111111 11111111 11111111 11111111// 10000000 00000000 00000000 00000000// 说明重入的次数超过界限了。throw new Error("Maximum lock count exceeded");// 正常的将计算结果，复制给statesetState(nextc);// 锁重入成功return true;}// 获取锁失败，返回falsereturn false;
}// 公平锁
protected final boolean tryAcquire(int acquires) {// 获取当前线程final Thread current = Thread.currentThread();int c = getState();if (c == 0) { // 判断state是不是0// 查看AQS中是否有排队的Node// 没人排队，抢一手；有人排队，但我是第一个，也抢一手if (!hasQueuedPredecessors() &&compareAndSetState(0, acquires)) {setExclusiveOwnerThread(current);return true;}}// 锁重入else if (current == getExclusiveOwnerThread()) {int nextc = c + acquires;if (nextc < 0)throw new Error("Maximum lock count exceeded");setState(nextc);return true;}return false;
}// 查看是否有线程在AQS的双向队列中排队
// 返回fasle，代表没人排队
public final boolean hasQueuedPredecessors() {// 头尾节点Node t = tail; // Read fields in reverse initialization orderNode h = head;Node s; // s为头节点的next节点// 如果头尾节点相等，证明没有线程排队，直接去抢占锁资源return h != t &&// s节点不为null，并且s节点的线程为当前线程（排在第一名的是不是我）((s = h.next) == null || s.thread != Thread.currentThread());
}

• addWaiter方法：将没有拿到锁资源的线程扔到AQS队列中去排队。

// 没有拿到锁资源，过来排队，mode：代表互斥锁
private Node addWaiter(Node mode) {// 将当前线程封装为NodeNode node = new Node(Thread.currentThread<