java高并发编程基础之AQS
2021-03-01 01:27
标签：环境 tar 转发 没有 cell int 线程等待 通知 log 曾经有一道比较比较经典的面试题“你能够说说 AQS是一种提供了原子式管理同步状态、阻塞和唤醒线程功能以及队列模型的简单框架 说到同步我们如何来保证同步？大家第一印象肯定是加锁了，说到锁的话大家肯定首先会想到的是Synchronized。 AQS里面用到了大量的CAS学习AQS之前我们还是有必要简单的先了解下 下面我们来写一个demo分析下lock 加锁和释放锁的过程 接下来我们继续分析源码，看下获取锁失败是如何被加入队列的。 接下来再看看enq方法 通过addWaiter方法已经把获取锁的线程通过封装成一个NODE加入对列。上述方法的一个执行流程图如下： 这个方法每当有一个node获取到锁了，就把当前 说到这个方法我们就要先看下NODE可能会有哪些状态在源码里面我们可以看到总共会有四种状态 挂起当前线程，并且阻塞 加锁成功了，那锁用完了就应该释放锁了，释放锁重点看下unparkSuccessor这个方法就好了 释放锁代码比较简单，基本都写在代码注释里面了，流程如下： java高并发编程基础之AQS 标签：环境 tar 转发 没有 cell int 线程等待 通知 log 原文地址：https://www.cnblogs.com/root429/p/14445442.html引言
java的并发包下面有哪些常见的类？”大多数人应该都可以说出
CountDownLatch、CyclicBarrier、Sempahore多线程并发三大利器。这三大利器都是通过AbstractQueuedSynchronizer抽象类(下面简写AQS)来实现的，所以学习三大利器之前我们有必要先来学习下AQS。
AQS结构
Synchronized大家应该基本上都会使用，加锁和释放锁都是jvm 来帮我们实现的，我们只需要简单的加个 Synchronized关键字就可以了。
用起来超级方便。但是有没有一种情况我们设置一个锁的超时时间Synchronized就有点实现不了，这时候我们就可以用ReentrantLock来实现,ReentrantLock是通过aqs来实现的，今天我们就通过ReentrantLock来学习一下aqs。CAS && 公平锁和非公平锁
CAS、公平锁和非公平锁。CAS
CAS 操作包含三个操作数 -- 内存位置、预期数值和新值。CAS 的实现逻辑是将内存位置处的数值与预期数值想比较，若相等，则将内存位置处的值替换为新值。若不相等，则不做任何操作,这个操作是个原子性操作，java里面的AtomicInteger等类都是通过cas来实现的。公平锁和非公平锁
优点：等待锁的线程不会饿死，每个线程都可以获取到锁。
缺点：整体吞吐效率相对非公平锁要低，等待队列中除第一个线程以外的所有线程都会阻塞，CPU唤醒阻塞线程的开销比非公平锁大。
优点：可以减少CPU唤醒线程的开销，整体的吞吐效率会高点，CPU也不必取唤醒所有线程，会减少唤起线程的数量。
缺点：处于等待队列中的线程可能会饿死，或者等很久才会获得锁。
文字有点拗口，我们来个实际的例子说明下。比如我们去食堂就餐的时候都要排队，大家都按照先来后到的顺序排队打饭，这就是公平锁。如果等到你准备拿盘子打饭的时候
直接蹦出了一个五大三粗的胖子插队到你前面，你看打不赢他只能忍气吞声让他插队，等胖子打完饭了又来个小个子也来插你队，这时候你没法忍了，直接大吼一声让他滚，这个
小个子只能屁颠屁颠到队尾去排队了这就是非公平锁。
我们先来看看AQS有哪些属性// 头结点
private transient volatile Node head;
// 阻塞的尾节点，每个新的节点进来，都插入到最后，也就形成了一个链表
private transient volatile Node tail;
// 这个是最重要的，代表当前锁的状态，0代表没有被占用，大于 0 代表有线程持有当前锁
// 这个值可以大于 1，是因为锁可以重入，每次重入都加上 1
private volatile int state;
// 代表当前持有独占锁的线程，举个最重要的使用例子，因为锁可以重入
// reentrantLock.lock()可以嵌套调用多次，所以每次用这个来判断当前线程是否已经拥有了锁
// if (currentThread == getExclusiveOwnerThread()) {state++}
private transient Thread exclusiveOwnerThread; //继承自AbstractOwnableSynchronizer
final void lock() {
// 上来先试试直接把状态置位1，如果此时没人获取锁就直接
if (compareAndSetState(0, 1))
// 争抢成功则修改获得锁状态的线程
setExclusiveOwnerThread(Thread.currentThread());
else
acquire(1);
}
cas尝试失败，说明已经有人再持有锁，所以进入acquire方法 public final void acquire(int arg) {
if (!tryAcquire(arg) &&
acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
selfInterrupt();
}
tryAcquire方法，看名字大概能猜出什么意思，就是试一试。
tryAcquire实际上是调用了父类Sync的nonfairTryAcquire方法 final boolean nonfairTryAcquire(int acquires) {
final Thread current = Thread.currentThread();
// 获取下当前锁的状态
int c = getState();
// 这个if 逻辑跟前面一进来就获取锁的逻辑一样都是通过cas尝试获取下锁
if (c == 0) {
if (compareAndSetState(0, acquires)) {
setExclusiveOwnerThread(current);
return true;
}
}
// 进入这个判断说明 锁重入了 状态需要进行+1
else if (current == getExclusiveOwnerThread()) {
int nextc = c + acquires;
// 如果锁的重入次数大于int的最大值，直接就抛出异常了，正常情况应该不存在这种情况，不过jdk还是严谨的
if (nextc tryAcquire方法如果获取锁失败了,那么肯定就要排队等待获取锁。排队的线程需要待在哪里等待获取锁？这个就跟我们线程池执行任务一样，线程池把任务都封装成一个work，然后当线程处理任务不过来的时候，就把任务放到队列里面。AQS同样也是类似的，把排队等待获取锁的线程封装成一个NODE。然后再把NODE放入到一个队列里面。队列如下所示，不过需要注意一点head是不存NODE的。
就要执行acquireQueued方法，执行acquireQueued方法之前需要先执行addWaiter方法 private Node addWaiter(Node mode) {
Node node = new Node(Thread.currentThread(), mode);
// Try the fast path of enq; backup to full enq on failure
Node pred = tail;
if (pred != null) {
node.prev = pred;
// cas 加入队列队尾
if (compareAndSetTail(pred, node)) {
pred.next = node;
return node;
}
}
// 尾结点不为空 || cas 加入尾结点失败
enq(node);
return node;
}
enq
// 通过自旋和CAS一定要当前node加入队尾
private Node enq(final Node node) {
for (;;) {
Node t = tail;
// 尾结点为空说明队列还是空的，还没有被初始化，所以初始化头结点，可以看到头结点的node 是没有绑定线程的也就是不存数据的
if (t == null) { // Must initialize
if (compareAndSetHead(new Node()))
tail = head;
} else {
node.prev = t;
if (compareAndSetTail(t, node)) {
t.next = node;
return t;
}
}
}
}
，接下来就是继续执行acquireQueued方法acquireQueued
final boolean acquireQueued(final Node node, int arg) {
boolean failed = true;
try {
boolean interrupted = false;
for (;;) {
// 通过自旋去获取锁 前驱节点==head的时候去尝试获取锁，这个方法在前面已经分析过了。
final Node p = node.predecessor();
if (p == head && tryAcquire(arg)) {
setHead(node);
p.next = null; // help GC
failed = false;
return interrupted;
}
// 进入这个if说明node的前驱节点不等于head 或者尝试获取锁失败了
// 判断是否需要挂起当前线程
if (shouldParkAfterFailedAcquire(p, node) &&
parkAndCheckInterrupt())
interrupted = true;
}
} finally {
// 异常情况进入cancelAcquire，在jdk11的时候这个源码直接是catch (Throwable e){ cancelAcquire(node);} 简单明了
if (failed)
cancelAcquire(node);
}
}
setHead
node节点设置为头节点，可以简单的看做当前节点获取到锁了就把当前节点”移除“（变为头结点）队列。shouldParkAfterFailedAcquire
private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
int ws = pred.waitStatus;
// 前驱节点状态 如果这个状态为-1 则返回true，把当前线程挂起
if (ws == Node.SIGNAL)
return true;
// 大于0，说明状态为CANCELLED
if (ws > 0) {
do {
// 删除被取消的node（让被取消的node成为一个没有引用的node等着下次GC被回收）
node.prev = pred = pred.prev;
} while (pred.waitStatus > 0);
pred.next = node;
} else {
// 进入这里只能是 0，-2，-3。NODE节点初始化的时候waitStatus默认值是0，所以只有这里才有修改waitStatus的地方
// 通过cas 把前驱节点的状态设置为-1，然后返回false ，外面调用这个方法的是个循环，又会调用一次这个方法
compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
}
return false;
}
parkAndCheckInterrupt
private final boolean parkAndCheckInterrupt() {
LockSupport.park(this); // 挂起当前线程，阻塞
return Thread.interrupted();
}
解锁
private void unparkSuccessor(Node node) {
// 头结点状态
int ws = node.waitStatus;
if (ws 0) {
s = null;
// 从尾节点开始往前找，找到最前面的非取消的节点 这里没有break 哦
for (Node t = tail; t != null && t != node; t = t.prev)
if (t.waitStatus 
这段代码里面有一个比较经典的面试题：
如果头结点的下一个节点为空或者头结点的下一个节点的状态为取消的时候为什么要从后往前找，找到最前面非取消的节点？
总结
结束
站在巨人的肩膀上摘苹果:
https://tech.meituan.com/2019/12/05/aqs-theory-and-apply.html
https://javadoop.com/post/AbstractQueuedSynchronizer
https://www.cnblogs.com/yanlong300/p/10953185.html
下一篇：Java多线程