AbstractQueuedSynchronizer之AQS

一、可重入锁

　　可参考：可重入锁和递归锁

1，定义

• 指的是同一线程外层函数获得锁后，再进入该线程的内层方法会自动获取锁(前提：锁对象是同一个对象)。
• Java中的ReentranLock(显示锁)和Synchronized(隐式锁)都是可重入锁，可重入锁的一个优点是可在一定程度避免死锁
• 隐士锁：(即synchronized关键字使用的锁)默认是可重入锁(同步块、同步方法)

2，案例

　　synchronize隐式锁

public class Demo01_ReentrantLockSynchronizedMethod {

public static void main(String[] args) {
new Demo01_ReentrantLockSynchronizedMethod().m1();
}

private synchronized void m1() {
System.out.println("=====外层");
m2();
}

private synchronized void m2() {
System.out.println("=====中层");
m3();
}

private synchronized void m3() {
System.out.println("=====内层");

}

}

View Code

　　ReentrantLock显示锁

public class Demo01_ReentrantLockShow {

static Lock lock = new ReentrantLock();

public static void main(String[] args) {
new Thread(()->{
lock.lock();
try {
System.out.println(Thread.currentThread().getName() + "==============外部");
lock.lock();
try {
System.out.println(Thread.currentThread().getName() + "==============内部");
}finally {
lock.unlock();
}
}finally {
lock.unlock();
//                lock.unlock();
}

},"t1").start();

new Thread(()->{
lock.lock();
try {
System.out.println(Thread.currentThread().getName() + "==========进入方法");
}finally {
lock.unlock();
}
},"t2").start();
}

}

View Code

3，原理

• 每个锁对象拥有一个锁计数器和一个指向持有该锁的线程的指针
• 当执行monitorenter时,如果目标锁对象的计数器为零,那么说明它没有被其他线程持有,Java虚拟机会将该锁对象的持有线程设置为当前线程,并且将其计数器加1,否则需要等待,直至持有线程释放该锁
• 当执行monitorexit时,Java虚拟机则锁对象的计数器减1。计数器为零代表锁已经被释放

二、LockSupport

1，三种线程唤醒等待

a）synchronized，Object的wait和notify

　　代码

private static void SynchroziedWaitNotify() {
new Thread(() -> {
//如果注释掉，就会先执行进入程序等待被唤醒
try { Thread.sleep(3000); } catch (InterruptedException e) {
e.printStackTrace();
}

//如果注释掉synchronized 则会报错，因为wait和notify一定要在同步块或同步方法中
synchronized (objectLock) {
try {
System.out.println(Thread.currentThread().getName() + "=========进入");
objectLock.wait();
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println(Thread.currentThread().getName() + "=========结束");
}
}, "t1").start();

new Thread(() -> {
synchronized (objectLock) {
objectLock.notify();
System.out.println(Thread.currentThread().getName() + "=========唤醒");
}
}, "t2").start();
}

　　wait和notify的限制条件：

• wait和notify方法必须要在同步块或同步方法里且成对出现使用。 
• 先wait后notify才可以(如果先notify后wait会出现另一个线程一直处于等待状态)
• synchronized是关键字属于JVM层面。monitorenter(底层是通过monitor对象来完成,其实wait/notify等方法也依赖monitor对象只能在同步块或方法中才能调用wait/notify等方法)

b）Lock，Condition的await和signal

private static void LockAwaitSignal() {
new Thread(() -> {
//如果把下行这句代码打开,先signal后await,会出现A线程一直处于等待状态
try { Thread.sleep(3000); } catch (InterruptedException e) {
e.printStackTrace();
}
//如果不加lock锁也会出现错误同synchronize
lock.lock();
try {
System.out.println(Thread.currentThread().getName() + "=========进入");
condition.await();
} catch (InterruptedException e) {
e.printStackTrace();
}finally {
lock.unlock();
}
System.out.println(Thread.currentThread().getName() + "=========结束");
}, "t1").start();

new Thread(() -> {
lock.lock();
try {
condition.signal();
System.out.println(Thread.currentThread().getName() + "=========唤醒");
}finally {
lock.unlock();
}
}, "t2").start();
}

　　await和signal的限制条件：

• await和signal都需要许出现在lock中，否则会报错
• 必须先await再signal否则会出现线程等待

c）LockSupport的park和unpark

　　代码

private static void lockSupportParkUnpark() {
Thread t1 = new Thread(() -> {
try { Thread.sleep(3000); } catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println(Thread.currentThread().getName() + "=========进入");
//如果这里有两个LockSupport.park(),因为permit的值为1,上一行已经使用了permit，
// 所以下一行被注释的打开会导致程序处于一直等待的状态
LockSupport.park();
LockSupport.park();
System.out.println(Thread.currentThread().getName() + "=========结束");
}, "t1");
t1.start();

new Thread(() -> {
//有两个LockSupport.unpark(t1),由于permit的值最大为1,所以只能给park一个通行证
LockSupport.unpark(t1);
LockSupport.unpark(t1);
System.out.println(Thread.currentThread().getName() + "=========唤醒");
}, "t2").start();
}

2，LockSupport的描述

• LockSupport是用来创建锁和其他同步类的基本线程阻塞原语
• LockSupport是一个线程阻塞工具类，所有的方法都是静态方法，可以让线程在任意位置阻塞，阻塞之后也有对应的唤醒方法。归根结底，LockSupport调用的Unsafe中的native代码。
• LockSupport提供park()和unpark()方法实现阻塞线程和解除线程阻塞的过程LockSupport和每个使用它的线程都有一个许可(permit)关联。permit相当于1，0的开关，默认是0，每个线程都有一个相关的permit, permit最多只有一个，重复调用unpark也不会积累凭证。

3，LockSupport的面试题

a）为什么可以先唤醒线程后阻塞线程?
　　因为unpark获得了一个凭证，之后再调用park方法，就可以名正言顺的凭证消费，故不会阻塞。
b）为什么唤醒两次后阻塞两次，但最终结果还会阻塞线程?
　　因为凭证的数量最多为1，连续调用两次unpark和调用一次unpark效果一样，只会增加一个凭证;而调用两次park却需要消费两个凭证，证不够，不能放行。

三、AQS的架构

1，AQS是什么

　　是用来构建锁或者其它同步器组件的重量级基础框架及整个JUC体系的基石，通过内置的CLH(FIFO)队列的变种来完成资源获取线程的排队工作,将每条将要去抢占资源的线程封装成一个Node节点来实现锁的分配，有一个int类变量表示持有锁的状态,通过CAS完成对state值的修改(0表示没有,1表示阻塞次数用于记录可重入)

　　

2，AQS的内部结构体系

　　

 　　

四，ReentrantLock非公平锁之lock

1，NonfairSync 继承Sync

static final class NonfairSync extends Sync {
private static final long serialVersionUID = 7316153563782823691L;

//非公平锁加锁
final void lock() {
//首先尝试修改 state 如果能从0修改为1 则表示当前还没有对象加锁成功
if (compareAndSetState(0, 1))
//修改此时的线程持有者为当前线程
setExclusiveOwnerThread(Thread.currentThread());
else
//否则就表示当前已经有线程持有锁。此时开始尝试获得锁
acquire(1);
}

protected final boolean tryAcquire(int acquires) {
return nonfairTryAcquire(acquires);
}
}

• 直接尝试加锁的比较并交换，compareAndSetState(0, 1)。
• 如果线程状态state为0，会加锁成功并修改当前线程的持有者
• 如果线程状态state不为0，会加锁失败，会继续一步逻辑2
  

2,调用AbstractQueuedSynchronizer.acquire(int arg)

//尝试获得锁
public final void acquire(int arg) {
//1. tryAcquire(arg) 实际上是调用NonfairSync.tryAcquire(1)。表示当前线程是否获取锁成功
//2. addWaiter(Node.EXCLUSIVE)初始化CLH链表
//3. acquireQueued(addWaiter(Node.EXCLUSIVE), arg)
if (!tryAcquire(arg) &&
acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
selfInterrupt();
}

a）第一步tryAcquire(arg)

　　先调用NonfairSync.tryAcquire(1)实际也就是a)中的方法，最终调用Sync.nonfairTryAcquire(1)方法。

final boolean nonfairTryAcquire(int acquires) {
//获取当前线程和当前对象锁的状态
final Thread current = Thread.currentThread();
int c = getState();
//如果状态为0，表示当前锁没有被占有。 修改当前状态为1，并且修改当前持有线程。并且返回获取锁成功
if (c == 0) {
if (compareAndSetState(0, acquires)) {
setExclusiveOwnerThread(current);
return true;
}
}
//如果 当前线程为持有线程 。nextc指针为当前线程持有锁的数量，表示可重入锁。 并且返回获取锁成功
else if (current == getExclusiveOwnerThread()) {
int nextc = c + acquires;
if (nextc < 0) // overflow
throw new Error("Maximum lock count exceeded");
setState(nextc);
return true;
}
//如果当前线程没有获取到锁，就返回false。
return false;
}

• 这时当前线程会继续判断当前state是否为0，如果为0则直接抢占锁compareAndSetState(0, acquires)，如果能抢占成功就直接修改对象的线程持有者。这种情况直接返回true。
• 如果当前线程就是正在执行的线程，就会为state属性加值，也就是可重锁了。这种情况直接返回true。
• 如果当前线程并没有获取到锁则会直接返回false。这时会走b)第二步

b)第二步addWaiter(Node.EXCLUSIVE)

　　实际调用AbstractQueuedSynchronizer.addWaiter(Node mode)，传入的mode为null

private Node addWaiter(Node mode) {
//初始化node结点其中Thread为当前线程
Node node = new Node(Thread.currentThread(), mode);
// 定义pred为尾结点，第一次调用的情况下当前结点值为null
Node pred = tail;
if (pred != null) {
node.prev = pred;
if (compareAndSetTail(pred, node)) {
pred.next = node;
return node;
}
}
//如果pred为null，则执行入CLH链表操作。返回当前链表的上一次出现的尾结点，当不存在尾结点时会构建一个虚结点(但是enq这个方法没有用返回值)
//并且把当前传入的节点直接保存到当前链表的尾部
enq(node);
//返回当前加入的结点
return node;
}

• 始化node结点为当前线程的结点
• 判断tail尾结点是否为null，如果tail尾结点不为空，则将当前初始化的node结点直接加入到双向链表的最后，并令其为尾结点返回当前结点
• 如果tail尾结点为null，则需要执行enq(node)，来初始化话CLH链表。具体为调用AbstractQueuedSynchronizer.enq(final Node node)，其中node为当前线程构建的结点。

private Node enq(final Node node) {
//循环自旋
for (;;) {
//设置尾结点为t
Node t = tail;
if (t == null) {
//初始化头结点为一个空节点，也叫做哨兵结点(虚结点)
if (compareAndSetHead(new Node()))
//并设置尾结点=头结点
tail = head;
} else {
//第二次进入当前循环则得到设置传入结点的前面一个结点为头结点，构建双向链表
node.prev = t;
//第二次由于 t = tail = head，故而会比较并交换为单签node结点，也就是设置当前双向链表的尾结点为传入的node结点
if (compareAndSetTail(t, node)) {
//设置当前t结点为 虚结点，设置当前虚结点的下一个结点为当前传入的结点。并返回尾结点的上一个结点(第一个线程返回的是头结点,第二个线程返回的是第一个线程的结点)
t.next = node;
return t;
}
}
}
}

• 采用CAS自旋，如果tail尾结点为null，则初始化一个哨兵结点，并设置头结点和尾结点都为当前结点
• 如果tail尾结点不为null，就令当前传入的node结点为尾结点，保留哨兵结点。并返回尾结点的前一个结点。这样就构建了一个双向链表

c）第三步acquireQueued(addWaiter(Node.EXCLUSIVE), arg))

　　实际调用AbstractQueuedSynchronizer.acquireQueued(final Node node, int arg)，其中node为当前线程构建的结点，arg为1

final boolean acquireQueued(final Node node, int arg) {
boolean failed = true;
try {
boolean interrupted = false;
for (;;) {
//设置p结点为 当前传入结点的前一个结点,第一次为头结点.第二次为第一次的结点
final Node p = node.predecessor();
//如果p结点为头结点 ,就会再次去尝试获取锁
if (p == head && tryAcquire(arg)) {
//如果当前结点获取到锁,会设置头结点为当前结点.并设置p结点的下一个结点为null
//这样就是为了释放头结点
setHead(node);
p.next = null; // help GC
failed = false;
return interrupted;
}
//二次调用这个方法会返回true , 然后执行parkAndCheckInterrupt会将当前线程挂起
if (shouldParkAfterFailedAcquire(p, node) &&
parkAndCheckInterrupt())
interrupted = true;
}
} finally {
if (failed)
cancelAcquire(node);
}
}

• 采用CAS自旋，先获取当前结点的前一个结点p
• 如果p结点为头结点，证明当前结点为队列中下一个需要被调用的线程，首先再次尝试去获取锁
• 如果获取成功，则将当前结点设置为头结点，并将p结点引用置空方便回收。返回false中断当前线程获取锁的操作
• 如果获取失败，此时调用方法shouldParkAfterFailedAcquire，将当前线程park()挂起

private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
//第一次使用的时候初始化为0
int ws = pred.waitStatus;
//SIGNAL为-1,第二次会为true
if (ws == Node.SIGNAL)
//返回true
return true;
if (ws > 0) {

do {
node.prev = pred = pred.prev;
} while (pred.waitStatus > 0);
pred.next = node;
} else {
//第一次走这步,会设置pred结点中的ws为-1
compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
}
return false;
}

• 第一次初始化ws为0，然后修改pre的节点waitstatus为-1，并返回false，自旋继续调用
• 第二次返回true，挂起线程结束循环。

3，最终实现的效果图为

　　

五、ReentrantLock非公平锁之unlock

1，AbstractQueuedSynchronizer.release(int arg)

 　　传入参数arg为-1

public final boolean release(int arg) {
//尝试去释放锁
if (tryRelease(arg)) {
Node h = head;
if (h != null && h.waitStatus != 0)
unparkSuccessor(h);
return true;
}
return false;
}

• 先尝试释放锁，如果释放成功，则判断如果头结点不为空并且头结点的waitstatus不为0，就是释放队列中的下一个线程

2，尝试释放锁Sync.tryRelease(arg)

protected final boolean tryRelease(int releases) {
//将当前状态值减去release
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;
}

3，调用AbstractQueuedSynchronizer.unparkSuccessor

private void unparkSuccessor(Node node) {
//获取头结点的ws
int ws = node.waitStatus;
if (ws < 0)
//如果为-1则修改为0
compareAndSetWaitStatus(node, ws, 0);
//定义s为下一个结点
Node s = node.next;
//如果下个结点为null 或者是 下个结点的waitStatus>0
//则从为结点往前遍历,知道碰到waitStatus<=0的结点,赋值给s
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;
}
//给s线程一个许可证，其实就是按照队列的顺序去释放锁
if (s != null)
LockSupport.unpark(s.thread);
}

六、对比公平锁与非公平锁的区别

　　 从源码层面来看主要有两个地方不一样：

1，非公平锁当调用lock()方法时，当前线程会首先尝试去获取锁，而非公平锁会直接调用acquire(1)方法

　　

2，调用acquire()方法时也不同

　　

　　非公平锁会再次尝试修改当前锁状态，获得锁；而公平锁会调用hasQueuedPredecessors()方法首先判断是否需要入队列，再决定是否获取当前锁。

public final boolean hasQueuedPredecessors() {
Node t = tail;
Node h = head;
Node s;
//如果队列没有初始化，也就是不存在等待队列，那么t=null,h=null，会直接返回false。那么非公平锁取反会尝试获取锁
//如果队列已经初始化，那么t肯定不等于h（因为队列初始化之后存在哨兵结点）则 h!=t -> true，那么获取到的s为头结点的下一个结点
//如果s结点为null，则直接返回true，说明当前结点中只存在头结点，这种情况不会出现。因为队列中起码有一个元素
//如果s结点不为null -> false，并且s.thread!=Thread.currentThread() -> false ，说明下一个将要执行的线程为当前线程则不需要排队了，尝试获取锁
//如果s结点不为null -> false，并且s.thread!=Thread.currentThread() -> true说明下一个结点的线程不是当前线程，返回true，需要去排队
return h != t &&
((s = h.next) == null || s.thread != Thread.currentThread());
}