并发的编程(四)（并发编程的三个要素）

并发编程的实现原理(三)

一、Lock

二、ReentrantLock 重入锁

三、ReentrantLock的实现原理

四、ReentrantLock的源码分析

五、锁的释放流程

六、原本挂起的线程继续执行

七、公平锁和非的区别

八、 Condition 工具类

九、 Condition 源码分析

1. Java.util.concurrent 是在并发 编程中比较常用的工具类，里面包含很多来场景中使用的组件。比如线程池、阻塞队列、计时器同步并发集合队列、计时器同步并发集合等。
2. Lock

在 Lock 接口出现之前，Java中的应用程序对于 多线程并发安全处理只能基synchronized关键字来解决 但是 synchronized在有些场景中会存一短板， 也就是它并不适合于所有的发场景但是在Java5以后，Lock 的出现 可以解决synchronized 在某些场景中的短板，它比 synchronized 更加灵活。

3.Lock的实现

Lock 本质上是一个接口， 它定义了释放锁和获得的抽象方法本质上是一个接口，它定义了释放锁和获得的抽象方法就意味着它定义了锁的一个标准规范，也同时意不实现。实现 Lock 接口类有很多，以下为几个常见的锁实现

ReentrantLock：表示重入锁，它是唯一个实现了Lock 接口的类。重入锁指 的是 线程在获得锁之后，再次取该不需要阻塞而是直接关联一计数器增加重入次数。

ReentrantReadWriteLock ：重入读写锁，它实现了 ReadWriteLock 接口，在这个类中维护了两个锁，一是ReadLock ，一个是 WriteLock ，他们都分别实现了 LockLock 接口。读写锁是一种适合多少的场景下解决线程安全问题工具，基本原则是： 读和读不互斥、读和写互斥、写和写互斥。也就是说涉及到影响数据变化的都会存在互斥。

StampedLock是 JDK8引入的新锁机制，可以简单认为是读写锁的一个改进版本，读写虽然通过分离和功能使得之间可以完全并发，但是读和写有冲突的，如果大量线程存在，可能会引起饥饿。stampedLock是一种乐观的读策略，使得锁完全不会阻塞写线程。

ReentrantLock

重入锁，表示支持重新进入的锁，也就是说，如果当前线程t1通过调用lock方法获取了锁之后，再次调用lock，是不会再阻塞去获取锁的，直接增加重试次数就行了。synchronized和ReentrantLock都是可重入锁。

重入锁的设计目的

ReentrantLock的使用

ReentrantReadWriteLock

读写锁：在同一时刻可以允许多个线程，但是在写线程访问时，所有的读线程和其他写都会被阻塞。读写锁维护了一对锁，一个读锁、一个写锁、; 一般情况下，读 写锁的性能都会比排它好因为大多数场景是于。在一般情况下，读 写锁的性能都会比排它好因为大多数场景是于。在一般情况下，读 写锁的性能都会比排它好因为大多数场景是于。在一般情况下，读写锁的性能都会比排它好，因为大多数场景读是读多与写的，在读多于写的情况下，读锁能够提供比排它更好并发性和吞吐量。

在这个栗子中，通过 hashmap来模拟了一个内存缓，然后使用读写锁保证这个内存缓存的线程安全。当执行读操作时候，需要获取读锁，在并发访问的时候读锁不会被阻塞，因为读操作不会影响执行效果， 在执行写操作时线程必须获得写锁，当已经有线程持有写锁的情况时，当前线程会被阻塞，只有当写锁释放以后其他读操作才能继续执行。 使用读写锁提升读操作的并发性，也保证每次写操作对所有读写操作的可见性。

? 读锁与读锁可以共享

? 读锁与写锁不可以共享（排他）

? 写锁与写锁不可以共享（排他）

ReentrantLock 的实现原理

锁的基本原理是，于将多线程并行任务通过某一种机制实现串 行执，从而达到线程安全性的目。在 synchronized 中，看到偏向锁、 轻量级锁、乐观锁。基于以及自旋来 优化了 synchronized 的加锁开销，同时在重量级锁阶段，通过线程的 阻塞以及唤醒来达到竞争和步目。那么在 ReentrantLock 中，也一定会存在这样的需要去解决问题。 就是在多线程竞争重入锁时，失败的线程是如何实现阻塞以及被唤醒呢？ 竞争重入锁时，失败的线程是如何实现阻塞以及被唤醒呢？

AQS

在 LockLock 中，用到了一个同步队列AQS ，全称 AbsAbsAbstractQueuedSynchronizer，它 是 一个同步工具也是Lock 用来实现线程同步的核心组件，如果你搞懂了AQS ，那 J.U.C 中绝大部分的工具都能轻松掌握。从使用层面来说， AQS 的功能分为两种：独占和共享,独占锁，每次只能有一个线程持有 独占锁， ReentrantLock 就是 以独占方式实现的互斥锁，共享锁，允许多个线程同时获取并发访问资源比如ReentrantReadWriteLock ，AQS的内部实现维护的是一个 FIFO 的双向链表，这种结构特点是每个数据结构都有两个指针，分别指向直接的后继节点和前驱。所以双链表可从任 意一个 节点开始很方便的访问前驱和后继。每Node其实是由线 程封装，当线程争抢锁失败后会封装成 Node 加入到 ASQ 队列中去，当获取锁的线程释放以，会从队列中唤醒一个阻塞的节点(线程 )。

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;

final boolean isShared() {

return nextWaiter == SHARED;

}

final Node predecessor() throws NullPointerException {

Node p = prev;

if (p == null)

throw new NullPointerException();

else

return p;

}

Node() { // Used to establish initial head or SHARED marker

}

Node(Thread thread, Node mode) { // Used by addWaiter

this.nextWaiter = mode;

this.thread = thread;

}

Node(Thread thread, int waitStatus) { // Used by Condition

this.waitStatus = waitStatus;

this.thread = thread;

}

}

释放锁以及添加线程对于队列的变化

当出现锁竞争以及释放的时候， AQS同步队列中的节点会发生变化，首先看一 下添加节点的场景。

1. 新的线程封装成 Node节点追加到同步队列中，设置prev 节点以及修改当前节点的前置节点next 节点指向自己

2. 通过 CAS 将 tail 重新指向新的尾部节点， head 节点表示获取锁成功的， 当头结在释放同步状态时会唤醒后继节点，如果后继节点获取锁成功，会把自己设置为头结的变化过程如下

这个过程也是涉及到两变化 ：

1. 修改 head节点 指向下一个获得锁的节点

2. 新的获得锁节点，将 新的获得锁节点，prev 的指针向null

设置 head 节点不需要用CAS ，原因是设置 head节点是由获得锁的线程来完成 而同步锁只能 的，由一个线程获得，所以不需要CAS 保证，只需要把 head 节点设置为原 首节点的后继，并且断开 head 节点的next 引用即可。

ReentrantLock

调用 ReentrantLock中的 lock()方法，源码的调用过程时序图

ReentrantLock.lock()

sync 实际上是一个抽象的静态内部类，它继承了 AQS 来实现重入锁的逻辑，前面说过 AQS 是一个同步队列 ，它能够实现线程的阻塞以及唤醒 ，它并不具备 业务功能，所以在不同的 同步场景中，会继承AQS 来实现对应场景的功能

Sync有两个具体的实现类，分别是：

NofairSync ：表示可以存在抢占锁的功能，也就是说不管当前队列上是否其存其他 线程等待，新都有机会抢占锁

FailSync : 表示所有线程严格按照FIFO来获取锁

NofairSync.lock

以非公平锁为例，来看 lock中的实现

1. 非公平锁和公平锁最大的区别在于，在非公平锁和中我抢占锁逻辑是不管有 没有线程排队，我先上来cas 去抢占一下。

2. CAS 就表示获得了 成功

3. CA失败，调用acquire(1) 走锁竞争逻辑

CAS 的实现原理

通过 cas 乐观锁的方式来做比较并替换 这段代码的 意思是，如果当前内存中state 的值 和预期的值 expect 相等，则替换为 update 。更新成功返回 true ，否则返 回 false。

这个操作是原子的，不会出现线程安全问题里面涉及到 Unsafe 这个类的操作 以及 涉及到 state这个属性的意义。

state 是 AQS 中的一个属性， 它在不同实现中所表达含义不一样， 对于重入锁的实现来说，表示 一个同步状态。它有两含义锁的实现来说。

1. 当 state=0，表示无锁状态

2. 当 state>0时，表示已经有线程获得了锁也就是state=1，但是因为 ReentrantLock 允许重入， 所以同一个线程多次获得步锁的时候允许重入， 所以同一个线程多次获得同步锁的时候state会递增， 比如重入 5次，那么state=5。 而在释放 锁的时候，同样需要5次直到 state=0 其他线程才有资格获得锁。

Unsafe 类

Unsafe 类是在 类是在 sun.misc 包下，不属于 Java Java标准。但是很多Java 的基础类库，包 括一些被广泛使用的高性能开发库都是基于 Unsafe 类开发的比如Netty、 Hadoop、KafkaKafka 等,Unsafe可认为是Java中留下的后门，提供了一些低层次操作如直接内存访问、 、 线程的挂起和恢复、CAS 线程同步内存屏障 而 CAS就是 Unsafe 类中提供的一个原子操作 ，第一个参数为需要改变的对象， 第二个为偏移量(即之前求出来的 headOffset的值 )，第 三个参数为期待的值，，第 三个参数为期待的值四个为更新后的值 ,第四个为更新后的值整个方法的作用是如果当前时刻值 等于预期值var4 相等，则更新为的期望值var5，如果更新成功，则返回 ，true ，否则返回false。

stateOffset

一个Java对象可以看成是一段内存 ，每个字段都得按照一定的顺序放在这段内存里，通过这个方法可以准确地告诉你某字段相对于对象的起始内存址字节偏移，偏移。用于在后面的 compareAndSwapInt 中，去根据偏移量找到对象在内存中的 具体位置所以 stateOffset 表示 state 这个字段在AQS 类的内存中相对于该类首地址偏移量。

AQS .accquire

acquire是 AQS 中的方法，如果CAS操作未能成功，说明state 已经不为0，acquire(1)操作，

1. 通过 tryAcquire 尝试获取独占锁，如果成功返回 尝试获取独占锁，如果成功返回true ，失败返回 false

2. 如果 tryAcquire失败，则会通过addWaiter 方法将当前线程封装成 Node 添加 到 AQS 队列尾部

3.acquireQueued ，将 NodeNode 作为参数，通过自旋去尝试获取锁。

NonfairSync.tryAcquire

这个方法的作用是尝试获取锁，如果成功返回 true ，不成功返回 false它是重写tryAcquire方法，并且AQS 中 tryAcquire方法的定义，并没有实现而是抛出异常。按照一般思维模式既然是一个不实现的模板方法

ReentrantLock.nofairTryAcquire

1. 获取当前线程，判断的锁状态

2. 如果 state=0表示当前是无锁状态，通过cas 更新state 状态的值

3. 当前线程是属于重入，则增加次数

AQS.addWaiter

当 tryAcquire方法获取锁失败以后，则会先调用 addWaiter将当前线程封装成Node,入参 mode 表示当前节点的状态，传递参数是Node.EXCLUSIVE，表示独占状态 ，意味着 重入锁用到了 AQS 的独占锁功能

1. 将当前线程封装成Node

2. 当前链表中的tail 节点是否为空，如果不为空，则通过cas 操作把当前线程的node添加到AQS队列

3. 如果为空或者 cas失败，调用 enq将节点添加到 AQS 队列

enq 就是通过自旋操作把当前节点加入到队列中

图解分析

假设 3个线程来争抢锁，那么截止到 enq 方法运行结束之后， 或者调用 addwaiter方法结束后，AQS 中的链表结构图

AQS .acquireQueued

通过 addWaiter方法把线程添加到链表后，会接着把Node 作为参数传递给acquireQueued 方法，去竞争锁

1. 获取当前节点的prev 节点

2. 如果 prev 节点为 head 节点，那么它就有资格去争抢锁调用 tryAcquire抢占锁

3. 抢占锁成功以后，把获得的节点设置为head ，并且移除原来的初始化head节点

4. 如果获得锁失败，则根据waitStatus 决定是否需要挂起线程

5. 最后通过cancelAcquire取消获得锁的操作

NofairSync. tryAcquire

这个方法在前面分析过 ，就是通过state 的状态 来判断是否处于无锁的状态 然后在 通过 cas 进行竞争锁操作。成功表示获得锁，失败 表示没获得锁

shouldParkAfterFailedAcquire

如果 ThreadA 的锁还没有释放情况下， ThreadB 和ThreadC 来争抢锁肯定是会 失败，那么以后会调用shouldParkAfterFailedAcquire方法

Node有 5中状态，分别是： CANCELLED（1），SIGNAL（-1）、CONDITION（-2）、PROPAGATE(-3) 、默认状态 (0 )

CANCELLED:在同步队列中等待的线程超时或被断，需要从取 在同步队列中取消该 Node 的结点 , 其结点的waitStatus 为 CANCELLED，即结束状态，进入该状态后的结点将不会再变化

SIGNAL: 只要前置节点释放锁，就会通知标识为SIGNAL 状态的后续节点线程

CONDITION：和Condition 有关系，后续说

PROPAGATE 共享模式下， PROPAGATE状态的线程处于可运行状态

0:初始状态 这个作用是，通过 Node 的状态来判断,ThreadA竞争锁失败以后是否应该被挂起

1. 如果 ThreadA的pred节点状态为SIGNAL，那就表示可以放心挂起当前线程

2. 通过循环扫描链表把CANCELLED状态的节点移除

3. 修改 pred节点的状态为SIGNAL返回false.

返回false时，也就是不需要挂起返回true ，则需要调用 parkAndCheckInterrupt挂起当前线程

private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {

int ws = pred.waitStatus;//前置节点的waitStatus

if (ws == Node.SIGNAL)

//如果前置节点为SIGNAL，意味着只需要等待其他前置节点的线程被释放，

return true;//返回true，意味着可以直接放心的挂起了

if (ws > 0) {

//ws大于0，意味着prev节点取消了排队，直接移除这个节点就行

do { node.prev = pred = pred.prev;

//相当于: pred=pred.prev;

node.prev=pred;

} while (pred.waitStatus > 0);

//这里采用循环，从双向列表中移除CANCELLED的节点

pred.next = node;

} else {

//利用cas设置prev节点的状态为SIGNAL(-1)

compareAndSetWaitStatus(pred, ws, Node.SIGNAL);

}

return false;

}

parkAndCheckInterrupt

使用 LockSupport挂起当前线程线程WATING状态

Thread.interrupted ，返回 当前线程是否 被其它线程触发过中断请求，也就是thread.interrupt(); 如果有触发过中断请求，那么这个方法会返回当前的标识 true ，并且对中断标识进行复位已经响应过了请求如果返回true ，意味着 acquire方法中会执行selfInterrupt()。

selfInterrupt： 标识 如果当前线程在acquireQueued 中被断过，则需要产生一 个中断请求 ，原因是线程在调用acquireQueued 方法的时候是不会响应中断请求的

通过 acquireQueued方法来竞争锁，如果 ThreadA还在执行中没有释放锁的话，意味着 ThreadB和 ThreadC只能挂起了。

LockSupport

LockSupport类是Java6引入的一个类，提供了基本线程同步原语。 引入的一个类，提供了基本线程同步原语。LockSupport实际上是调用了 Unsafe 类里的函数，归结到Unsafe 里只有两个函数

unpark函数为线程提供“许可 (permit)”，线程调用park 函数则等待“许可 ”。这个有点像信号量，但是这个 点像信号量，但是这个“许可 ”是一次性的。 是一次性的。 是一次性的。

permit相当于 0/1 的开关，默认是 0，调用一次unpark就加 1变成了 变成了 1. 调用一次park会消费permit ，又会变成 0。 如果再调用一次parkpark 会阻塞，因为 会阻塞permit已经是 0了。直到permit 变成 1. 这时调用unpark 会把permit 设置为1. 每个线程都有一个相关的permit最多只有一个，重复调用unpark 不会累积。

锁的释放流程

如果这个时候ThreadA 释放锁了，那么我们来看锁被释放后会产生什么效果

ReentrantLock.unlock

unlock中，会调用release方法来释放锁

ReentrantLock.tryRelease 这个 方法 可以认为是一个设置锁状态的操作， 通过 将 state 状态减掉传入的参数值（参数是 1），如果结状态为 0，就将排它锁的Owner 设置为null ，以使得其它的线程有机会进行执行。

在排它锁中，加 的时候状态会增1（当然可以自己修改这个值），在解锁的时 候减掉 1，同一个锁，在可以重入后能会被叠加为 2、3、4这些值，只有unlock() 的次数与与lock() 的次数对应才会将 Owner 线程设置为空，而且也只有这种情况下才会返回true 。

unparkSuccessor

为什么在释放锁的时候是从tail进行扫描

1. 将新的节点prev 指向tail

2. 通过 cas 将tail设置为 新的节点，cas 是原子操作所以能够保证线程安全性

3. t.next=node设置原tail的next 节点指向新的节点

在 cas 操作之后，t.next=node操作之前。存在其他线程调用 unlock 方法从head 开始往后遍历，由于 t.next .next =node还没执行意味着链表的关系还没建立完整。就会导致遍历到 t节点的时候被中断。 所以从后往前遍历，一定不会存在这个问题。

通过锁的释放，原本结构就发生了一些变化head 节点的waitStatus 变成了 0， ThreadB ThreadB 被唤醒

原本挂起的线程继续执行

通过 ReentrantLock.unlock ，原本挂起的线程， 被唤醒以后继续执行，在 acquireQueued 方法中被唤 醒以后继续开始执行

AQS .acquireQueued

只关注一下ThreadB被唤醒以后的执行流由于ThreadB 的prev 节点指向的是 head ，并且 ThreadA 已经释放了锁。所以这 个时候调用tryAcquire方法时，可以顺利获取到锁。

1. 把 ThreadB节点当成 head

2. 把原 head节点的 节点的 next 节点指向为 null

1. 设置新 head 节点的prev=null

2. 设置原 head 节点的next节点为null

公平锁和非的区别

锁的公平性是相对于获取顺序而言， 如果是一个锁那么获的锁取顺序 就应该符合请求的绝对时间顺序，也是 FIFO。 在上面分析的例子来说，只要 CAS设置同步状态成功，则表示当前线程获取了锁,而公平锁不一样差异点有两个

FairSync.tryAcquire

非公平锁在获取锁的时候，会先通过CAS 进行抢占，而公平锁则不会

FairSync .tryAcquire

这个 方法与nonfairTryAcquire(int acquires)比较，不同的地方在于判断条件多了 判断条件多了 hasQueuedPredecessors()方法，也就是 加入了[同步队列中当前节点是否有前驱节点 ]的判断，如果该方法返回true ， 则表示有线程比当前更早地请求获取锁， 因此需要等待前驱线程获取并释放后才能继续获取锁

Condition

在前面说 synchronized 的时候，讲到 wait/notify的基本使用，结合 synchronized 可以实现对线程的通信。那么这个时候我就在思考了，既然 可以实现对线程的通信。那么这个时候我就在思考了，既然 J.U.C 里 面提供了锁的实现机制，那 J.U.C.里面有没提供类似的线程通信工具呢？ 发现了一个 Condition工具类 。

Condition是一个多线程协调通信的工具类，可以让某些起等待条件(condition)，只有满足条件时，线程才会被唤醒。

Condition的基本使用

ConditionSignal

通过这个案例简单实现了wait和 notify的功能，当调用await 方法后，当前线程会释放锁并等待，而其他线程调用condition 对象的 signal 或者 signalall 方法通知并被阻塞的线程，然后自己执行unlock 释放锁，被唤醒的线程获得之前的锁，续执行，最后释放锁。

所以condition中两个最重要的方法，一个是 await，一个是 signal方法

await:把当前线程阻塞挂起

signal:唤醒阻塞的线程

Condition源码

调用 Condition，需要获得 ，需要获得 Lock 锁， 所以意味着会存在一个AQS同步队列Condition .await方法

condition.await

调用 Condition的 await()方法（或者以 await开头的方法），会使当前线程进入等待队列并释放锁，同时线程状态变为等待状态。当从await()方法返回时当前线程一定获取了Condition 相关联的锁 相关联的锁

Condition.signal

调用 Condition的signal()方法，将会唤醒在等待队列中时间最长的节点（首节点）节点），在 唤醒节点之前，会将移到同步队列中

Condition.doSignal

该方法先是CAS 修改了节点状态，如果成功就将这个放到AQS 队列中，然后唤醒这个节点上的线程。此时，那就会在await 方法中苏醒

AQS.transferForSignal

该方法先是 CAS 修改了节点状态，如果成功就将这个放到 AQS 队列中， 然后唤醒这个节点上的线程。。此时，那就会在 await 方法中苏醒

final boolean transferForSignal(Node node) {

if (!compareAndSetWaitStatus(node, Node.CONDITION, 0))

return false;

Node p = enq(node);

int ws = p.waitStatus;

// 如果上一个节点的状态被取消了, 或者尝试设置上一个节点的状态为 SIGNAL 失败了(SIGNAL 表示: 他的 next 节点需要停止阻塞),

if (ws > 0 || !compareAndSetWaitStatus(p, ws, Node.SIGNAL)) LockSupport.unpark(node.thread);

// 唤醒输入节点上的线程.

return true;

}

Condition

总结 阻塞： await()方法中，在线程释放锁资源 如果节点不在 AQS 等待队列，则阻塞当前线程如果在等待队，自旋尝试获取锁 释放：signal() 后，节点会从 condition队列移动到 AQS 等待队列，则进入 正常锁的获取流程。

并发编程的实现原理(三)

我是阳光随馨馨，如果你看完了，点个赞，加个关注，转发一下哈