并发编程-2

1.锁的分类

1.1 可重入锁、不可重入锁

Java中提供的synchronized，ReentrantLock，ReentrantReadWriteLock都是可重入锁。

1.1.1重入：

当前线程获取到A锁，在获取之后尝试再次获取A锁是可以直接拿到的。

1.1.2不可重入：

当前线程获取到A锁，在获取之后尝试再次获取A锁，无法获取到的，因为A锁被当前线程占用着，需要等待自己释放锁再获取锁。

1.2 乐观锁、悲观锁

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

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

悲观锁获取不到锁资源时，会将当前线程挂起（进入BLOCKED、WAITING），线程挂起会涉及到用户态和内核的太的切换，而这种切换是比较消耗资源的。

1.2.1用户态

JVM可以自行执行的指令，不需要借助操作系统执行。

1.2.2内核态：

JVM不可以自行执行，需要操作系统才可以执行。

乐观锁：获取不到锁资源，可以再次让CPU调度，重新尝试获取锁资源。

Atomic原子性类中，就是基于CAS乐观锁实现的。

1.3 公平锁、非公平锁

Java中提供的synchronized只能是非公平锁。

Java中提供的ReentrantLock，ReentrantReadWriteLock可以实现公平锁和非公平锁

公平锁：线程A获取到了锁资源，线程B没有拿到，线程B去排队，线程C来了，锁被A持有，同时线程B在排队。直接排到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 MiTest {public static void main(String[] args) {// 锁的是，当前Test.classTest.a();Test test = new Test();// 锁的是new出来的test对象test.b();}}class Test{public static synchronized void a(){System.out.println("1111");}public synchronized void b(){System.out.println("2222");}
}

2.2synchronized的优化

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

2.2.1锁消除：

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

public synchronized void method(){// 没有操作临界资源// 此时这个方法的synchronized你可以认为木有~~
}

2.2.2锁膨胀：

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

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

2.2.3锁升级：

ReentrantLock的实现，是先基于乐观锁的CAS尝试获取锁资源，如果拿不到锁资源，才会挂起线程。synchronized在JDK1.6之前，完全就是获取不到锁，立即挂起当前线程，所以synchronized性能比较差。

synchronized就在JDK1.6做了锁升级的优化

2.2.4无锁、匿名向：

当前对象没有作为锁存在。
偏向锁：

如果当前锁资源，只有一个线程在频繁的获取和释放，那么这个线程过来，只需要判断，当前指向的线程是否是当前线程 。
  如果是，直接拿着锁资源走。
  如果当前线程不是,我基于CAS的方式，尝试将偏向锁指向当前线程。如果获取不到，触发锁升级，升级为轻量级锁。（偏向锁状态出现了锁竞争的情况）
轻量级锁：会采用自旋锁的方式去频繁的以CAS的形式获取锁资源（采用的是自适应自旋锁）
如果成功获取到，拿着锁资源走
如果自旋了一定次数，没拿到锁资源，锁升级。
重量级锁：就是最传统的synchronized方式，拿不到锁资源，就挂起当前线程。（用户态&内核态）

2.2synchronized实现原理

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

2.2.1 MarkWord

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

 2.2.2synchronized的锁升级

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

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

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

偏向锁撤销：

偏向锁撤销是指当一个线程获取到偏向锁后，发现该对象已经被其他线程访问过，就需要将偏向锁升级为轻量级锁或重量级锁，这个过程就是偏向锁撤销。在偏向锁撤销的过程中，需要将持有偏向锁的线程ID清空，同时将对象头中的状态位设置为无锁状态。这个过程会涉及到CAS操作，因此需要保证线程安全。

偏向锁撤销的主要作用是防止偏向锁持有时间过长，导致其他线程长时间等待锁的释放，从而影响系统的性能。通过偏向锁撤销，可以使锁的获取更加公平，提升系统的并发性能。

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

因为JVM在启动时，需要加载大量的.class文件到内存中，这个操作会涉及到synchronized的使用，为了避免出现偏向锁撤销操作，JVM启动初期，有一个延迟4s开启偏向锁的操作，如果正常开启偏向锁了，那么不会出现无锁状态，对象会直接变为匿名偏向

public static void main(String[] args) throws InterruptedException {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){System.out.println("main:" + ClassLayout.parseInstance(o).toPrintable());}
}

 2.2.3整个锁升级状态的转变：

2.2.4 Lock Record以及ObjectMonitor存储的内容

2.2.5重量锁底层ObjectMonitor

需要去找到openjdk，在百度中直接搜索openjdk，第一个链接就是

找到ObjectMonitor的两个文件，hpp，cpp

先查看核心属性：http://hg.openjdk.java.net/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++中实现的加锁流程

http://hg.openjdk.java.net/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.1ReentrantLock和synchronized的区别

3.1.1核心区别：

ReentrantLock是个类，synchronized是关键字，当然都是在JVM层面实现互斥锁的方式

3.1.2效率区别：

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

3.1.3底层实现区别：

实现原理是不一样，ReentrantLock基于AQS实现的，synchronized是基于ObjectMonitor

3.1.4功能向的区别：

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

 AQS内部结构和属性

3.3 加锁流程源码剖析

3.3.1 加锁流程概述

这个是非公平锁的流程

 

3.4 三种加锁源码分析

3.4.1lock方法

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

// 非公平锁
final void lock() {// 上来就先基于CAS的方式，尝试将state从0改为1if (compareAndSetState(0, 1))// 获取锁资源成功，会将当前线程设置到exclusiveOwnerThread属性，代表是当前线程持有着锁资源setExclusiveOwnerThread(Thread.currentThread());else// 执行acquire，尝试获取锁资源acquire(1);
}// 公平锁
final void lock() {//  执行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,之前持有锁的线程释放了锁资源if (c == 0) {// 再次抢一波锁资源if (compareAndSetState(0, acquires)) {setExclusiveOwnerThread(current);// 拿锁成功返回truereturn true;}}// 不是0，有线程持有着锁资源，如果是，证明是锁重入操作else if (current == getExclusiveOwnerThread()) {// 将state + 1int nextc = c + acquires;if (nextc < 0) // 说明对重入次数+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) {// 查看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的双向队列中排队
// 返回false，代表没人排队
public final boolean hasQueuedPredecessors() {// 头尾节点Node t = tail; Node h = head;// s为头结点的next节点Node s;// 如果头尾节点相等，证明没有线程排队，直接去抢占锁资源return h != t &&// s节点不为null，并且s节点的线程为当前线程（排在第一名的是不是我）(s == null || s.thread != Thread.currentThread());
}

 acquireQueued方法，判断当前线程是否还能再次尝试获取锁资源，如果不能再次获取锁资源，或者又没获取到，尝试将当前线程挂起

// 当前没有拿到锁资源后，并且到AQS排队了之后触发的方法。  中断操作这里不用考虑
final boolean acquireQueued(final Node node, int arg) {// 不考虑中断// failed：获取锁资源是否失败（这里简单掌握落地，真正触发的，还是tryLock和lockInterruptibly）boolean failed = true;try {boolean interrupted = false;// 死循环…………for (;;) {// 拿到当前节点的前继节点final Node p = node.predecessor();// 前继节点是否是head，如果是head，再次执行tryAcquire尝试获取锁资源。if (p == head && tryAcquire(arg)) {// 获取锁资源成功// 设置头结点为当前获取锁资源成功Node，并且取消thread信息setHead(node);// help GCp.next = null; // 获取锁失败标识为falsefailed = false;return interrupted;}// 没拿到锁资源……// shouldParkAfterFailedAcquire：基于上一个节点转改来判断当前节点是否能够挂起线程，如果可以返回true，// 如果不能，就返回false，继续下次循环if (shouldParkAfterFailedAcquire(p, node) &&// 这里基于Unsafe类的park方法，将当前线程挂起parkAndCheckInterrupt())interrupted = true;}} finally {if (failed)// 在lock方法中，基本不会执行。cancelAcquire(node);}
}
// 获取锁资源成功后，先执行setHead
private void setHead(Node node) {// 当前节点作为头结点  伪head = node;// 头结点不需要线程信息node.thread = null;node.prev = null;
}// 当前Node没有拿到锁资源，或者没有资格竞争锁资源，看一下能否挂起当前线程
private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {// -1，SIGNAL状态：代表当前节点的后继节点，可以挂起线程，后续我会唤醒我的后继节点// 1，CANCELLED状态：代表当前节点以及取消了int ws = pred.waitStatus;if (ws == Node.SIGNAL)// 上一个节点为-1之后，当前节点才可以安心的挂起线程return true;if (ws > 0) {// 如果当前节点的上一个节点是取消状态，我需要往前找到一个状态不为1的Node，作为他的next节点// 找到状态不为1的节点后，设置一下next和prevdo {node.prev = pred = pred.prev;} while (pred.waitStatus > 0);pred.next = node;} else {// 上一个节点的状态不是1或者-1，那就代表节点状态正常，将上一个节点的状态改为-1compareAndSetWaitStatus(pred, ws, Node.SIGNAL);}return false;
}

tryLock方法

// tryLock方法，无论公平锁还有非公平锁。都会走非公平锁抢占锁资源的操作
// 就是拿到state的值， 如果是0，直接CAS浅尝一下
// state 不是0，那就看下是不是锁重入操作
// 如果没抢到，或者不是锁重入操作，告辞，返回false
public boolean tryLock() {// 非公平锁的竞争锁操作return sync.nonfairTryAcquire(1);
}
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) // overflowthrow new Error("Maximum lock count exceeded");setState(nextc);return true;}return false;
}

 tryLock(time,unit);

第一波分析，类似的代码：

// tryLock(time,unit)执行的方法
public final boolean tryAcquireNanos(int arg, long nanosTimeout)throws InterruptedException {// 线程的中断标记位，是不是从false，别改为了true，如果是，直接抛异常if (Thread.interrupted())throw new InterruptedException();// tryAcquire分为公平和非公平锁两种执行方式，如果拿锁成功， 直接告辞，return tryAcquire(arg) ||// 如果拿锁失败，在这要等待指定时间doAcquireNanos(arg, nanosTimeout);
}private boolean doAcquireNanos(int arg, long nanosTimeout)throws InterruptedException {// 如果等待时间是0秒，直接告辞，拿锁失败  if (nanosTimeout <= 0L)return false;// 设置结束时间。final long deadline = System.nanoTime() + nanosTimeout;// 先扔到AQS队列final Node node = addWaiter(Node.EXCLUSIVE);// 拿锁失败，默认trueboolean failed = true;try {for (;;) {// 如果在AQS中，当前node是head的next，直接抢锁final Node p = node.predecessor();if (p == head && tryAcquire(arg)) {setHead(node);p.next = null; // help GCfailed = false;return true;}// 结算剩余的可用时间nanosTimeout = deadline - System.nanoTime();// 判断是否是否用尽的位置if (nanosTimeout <= 0L)return false;// shouldParkAfterFailedAcquire：根据上一个节点来确定现在是否可以挂起线程if (shouldParkAfterFailedAcquire(p, node) &&// 避免剩余时间太少，如果剩余时间少就不用挂起线程nanosTimeout > spinForTimeoutThreshold)// 如果剩余时间足够，将线程挂起剩余时间LockSupport.parkNanos(this, nanosTimeout);// 如果线程醒了，查看是中断唤醒的，还是时间到了唤醒的。if (Thread.interrupted())// 是中断唤醒的！throw new InterruptedException();}} finally {if (failed)cancelAcquire(node);}
}

 取消节点分析：

// 取消在AQS中排队的Node
private void cancelAcquire(Node node) {// 如果当前节点为null，直接忽略。if (node == null)return;//1. 线程设置为nullnode.thread = null;//2. 往前跳过被取消的节点，找到一个有效节点Node pred = node.prev;while (pred.waitStatus > 0)node.prev = pred = pred.prev;//3. 拿到了上一个节点之前的nextNode predNext = pred.next;//4. 当前节点状态设置为1，代表节点取消node.waitStatus = Node.CANCELLED;// 脱离AQS队列的操作// 当前Node是尾结点，将tail从当前节点替换为上一个节点if (node == tail && compareAndSetTail(node, pred)) {compareAndSetNext(pred, predNext, null);} else {// 到这，上面的操作CAS操作失败int ws = pred.waitStatus;// 不是head的后继节点if (pred != head &&// 拿到上一个节点的状态，只要上一个节点的状态不是取消状态，就改为-1(ws == Node.SIGNAL || (ws <= 0 && compareAndSetWaitStatus(pred, ws, Node.SIGNAL))) && pred.thread != null) {// 上面的判断都是为了避免后面节点无法被唤醒。// 前继节点是有效节点，可以唤醒后面的节点Node next = node.next;if (next != null && next.waitStatus <= 0)compareAndSetNext(pred, predNext, next);} else {// 当前节点是head的后继节点unparkSuccessor(node);}node.next = node; // help GC}
}