其他
我画了35张图,就是为了让你深入 AQS!
The following article is from 程序员cxuan Author 一枝花算不算浪漫
AQS(AbstractQueuedSynchronizer)
,所谓的AQS
即是抽象的队列式的同步器,内部定义了很多锁相关的方法,我们熟知的ReentrantLock
、ReentrantReadWriteLock
、CountDownLatch
、Semaphore
等都是基于AQS
来实现的。AQS
相关的UML
图:AQS
中 维护了一个volatile int state
(代表共享资源)和一个FIFO
线程等待队列(多线程争用资源被阻塞时会进入此队列)。volatile
能够保证多线程下的可见性,当state=1
则代表当前对象锁已经被占有,其他线程来加锁时则会失败,加锁失败的线程会被放入一个FIFO
的等待队列中,比列会被UNSAFE.park()
操作挂起,等待其他获取锁的线程释放锁才能够被唤醒。state
的操作都是通过CAS
来保证其并发修改的安全性。AQS
中提供了很多关于锁的实现方法,getState():获取锁的标志state值 setState():设置锁的标志state值 tryAcquire(int):独占方式获取锁。尝试获取资源,成功则返回true,失败则返回false。 tryRelease(int):独占方式释放锁。尝试释放资源,成功则返回true,失败则返回false。
ReentrantLock
作为突破点通过源码和画图的形式一步步了解AQS
内部实现原理。AQS
源码:线程一加锁成功时 AQS
内部实现线程二/三加锁失败时 AQS
中等待队列的数据模型线程一释放锁及线程二获取锁实现原理 通过线程场景来讲解公平锁具体实现原理 通过线程场景来讲解Condition中a wait()
和signal()
实现原理
AQS
内部的数据结构和实现原理场景分析
线程一加锁成功
AQS
内部数据为:Node
是一个双向链表,这里SIGNAL
是Node
中waitStatus
属性,Node
中还有一个nextWaiter
属性,这个并未在图中画出来,这个到后面Condition
会具体讲解的。java.util.concurrent.locks.ReentrantLock .NonfairSync:
final void lock() {
if (compareAndSetState(0, 1))
setExclusiveOwnerThread(Thread.currentThread());
else
acquire(1);
}
protected final boolean tryAcquire(int acquires) {
return nonfairTryAcquire(acquires);
}
}
CAS
尝试抢占锁,如果抢占成功state
值回被改为1,且设置对象独占锁线程为当前线程。如下所示:return unsafe.compareAndSwapInt(this, stateOffset, expect, update);
}
protected final void setExclusiveOwnerThread(Thread thread) {
exclusiveOwnerThread = thread;
}
线程二抢占锁失败
state
变为1,线程二通过CAS
修改state
变量必然会失败。此时AQS
中FIFO
(First In First Out 先进先出)队列中数据如图所示:java.util.concurrent.locks.AbstractQueuedSynchronizer.acquire()
:if (!tryAcquire(arg) &&
acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
selfInterrupt();
}
tryAcquire()
的具体实现:java.util.concurrent.locks.ReentrantLock .nonfairTryAcquire()
: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)
throw new Error("Maximum lock count exceeded");
setState(nextc);
return true;
}
return false;
}
nonfairTryAcquire()
方法中首先会获取state
的值,如果不为0则说明当前对象的锁已经被其他线程所占有,接着判断占有锁的线程是否为当前线程,如果是则累加state
值,这就是可重入锁的具体实现,累加state
值,释放锁的时候也要依次递减state
值。state
为0,则执行CAS
操作,尝试更新state
值为1,如果更新成功则代表当前线程加锁成功。state
修改为1,所以线程二通过CAS
修改state
的值不会成功。加锁失败。tryAcquire()
后会返回false,接着执行addWaiter(Node.EXCLUSIVE)
逻辑,将自己加入到一个FIFO
等待队列中,代码实现如下:java.util.concurrent.locks.AbstractQueuedSynchronizer.addWaiter()
:Node node = new Node(Thread.currentThread(), mode);
Node pred = tail;
if (pred != null) {
node.prev = pred;
if (compareAndSetTail(pred, node)) {
pred.next = node;
return node;
}
}
enq(node);
return node;
}
Node
节点,Node
为双向链表。此时等待对内中的tail
指针为空,直接调用enq(node)
方法将当前线程加入等待队列尾部:for (;;) {
Node t = tail;
if (t == null) {
if (compareAndSetHead(new Node()))
tail = head;
} else {
node.prev = t;
if (compareAndSetTail(t, node)) {
t.next = node;
return t;
}
}
}
}
tail
指针为空,进入if逻辑,使用CAS
操作设置head
指针,将head
指向一个新创建的Node
节点。此时AQS
中数据:head
、tail
、t
都指向第一个Node
元素。else
逻辑,此时已经有了head
节点,这里要操作的就是将线程二对应的Node
节点挂到head
节点后面。此时队列中就有了两个Node
节点:addWaiter()
方法执行完后,会返回当前线程创建的节点信息。继续往后执行acquireQueued(addWaiter(Node.EXCLUSIVE), arg)
逻辑,此时传入的参数为线程二对应的Node
节点信息:java.util.concurrent.locks.AbstractQueuedSynchronizer.acquireQueued()
:boolean failed = true;
try {
boolean interrupted = false;
for (;;) {
final Node p = node.predecessor();
if (p == head && tryAcquire(arg)) {
setHead(node);
p.next = null; // help GC
failed = false;
return interrupted;
}
if (shouldParkAfterFailedAcquire(p, node) &&
parkAndChecknIterrupt())
interrupted = true;
}
} finally {
if (failed)
cancelAcquire(node);
}
}
private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
int ws = pred.waitStatus;
if (ws == Node.SIGNAL)
return true;
if (ws > 0) {
do {
node.prev = pred = pred.prev;
} while (pred.waitStatus > 0);
pred.next = node;
} else {
compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
}
return false;
}
private final boolean parkAndCheckInterrupt() {
LockSupport.park(this);
return Thread.interrupted();
}
acquireQueued()
这个方法会先判断当前传入的Node
对应的前置节点是否为head
,如果是则尝试加锁。加锁成功过则将当前节点设置为head
节点,然后空置之前的head
节点,方便后续被垃圾回收掉。Node
的前置节点不是head
节点,就会通过shouldParkAfterFailedAcquire
方法 将head
节点的waitStatus
变为了SIGNAL=-1
,最后执行parkAndChecknIterrupt
方法,调用LockSupport.park()
挂起当前线程。AQS
中的数据如下图:AQS
的等待队列里面了,等着其他线程释放锁来唤醒它。线程三抢占锁失败
addWaiter(Node mode)
方法:Node node = new Node(Thread.currentThread(), mode);
Node pred = tail;
if (pred != null) {
node.prev = pred;
if (compareAndSetTail(pred, node)) {
pred.next = node;
return node;
}
}
enq(node);
return node;
}
tail
节点指向线程二,进入if
逻辑后,通过CAS
指令将tail
节点重新指向线程三。接着线程三调用enq()
方法执行入队操作,和上面线程二执行方式是一致的,入队后会修改线程二对应的Node
中的waitStatus=SIGNAL
。最后线程三也会被挂起。此时等待队列的数据如图:线程一释放锁
head
节点的后置节点,也就是我们现在的线程二,具体操作流程如下:AQS
中数据如图:java.util.concurrent.locks.AbstractQueuedSynchronizer.release()
if (tryRelease(arg)) {
Node h = head;
if (h != null && h.waitStatus != 0)
unparkSuccessor(h);
return true;
}
return false;
}
tryRelease()
方法,这个方法具体实现在ReentrantLock
中,如果tryRelease
执行成功,则继续判断head
节点的waitStatus
是否为0,前面我们已经看到过,head
的waitStatue
为SIGNAL(-1)
,这里就会执行unparkSuccessor()
方法来唤醒head
的后置节点,也就是我们上面图中线程二对应的Node
节点。ReentrantLock.tryRelease()
中的具体实现: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;
}
ReentrantLock.tryRelease()
后,state
被设置成0,Lock对象的独占锁被设置为null。此时看下AQS
中的数据:java.util.concurrent.locks.AbstractQueuedSynchronizer.unparkSuccessor()
方法,唤醒head
的后置节点:int ws = node.waitStatus;
if (ws < 0)
compareAndSetWaitStatus(node, ws, 0);
Node s = node.next;
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;
}
if (s != null)
LockSupport.unpark(s.thread);
}
head
节点的waitStatus
设置为0,然后解除head
节点next
的指向,使head
节点空置,等待着被垃圾回收。head
指针指向线程二对应的Node
节点,且使用LockSupport.unpark
方法来唤醒线程二。CAS
指令修改state
数据。执行完成后可以查看AQS
中数据:park
的地方继续执行,继续执行acquireQueued()
方法。线程二唤醒继续加锁
boolean failed = true;
try {
boolean interrupted = false;
for (;;) {
final Node p = node.predecessor();
if (p == head && tryAcquire(arg)) {
setHead(node);
p.next = null; // help GC
failed = false;
return interrupted;
}
if (shouldParkAfterFailedAcquire(p, node) &&
parkAndCheckInterrupt())
interrupted = true;
}
} finally {
if (failed)
cancelAcquire(node);
}
}
for
循环,判断线程二的前置节点是否为head
,如果是则继续使用tryAcquire()
方法来尝试获取锁,其实就是使用CAS
操作来修改state
值,如果修改成功则代表获取锁成功。接着将线程二设置为head
节点,然后空置之前的head
节点数据,被空置的节点数据等着被垃圾回收。AQS
中队列数据如下:线程二释放锁/线程三加锁
AQS
中队列数据如图:ReentrantLock
的默认实现,那我们接着来看一下公平锁的实现原理,这里先用一张图来解释公平锁和非公平锁的区别:tryAcquire()
方法使用CAS
操作来尝试修改state
值,如果此时又来了一个线程四也来执行加锁操作,同样会执行tryAcquire()
方法。AQS
等待队列中是存在节点,如果存在节点则会直接入队等待,具体代码如下.acquire()
方法,只不过公平锁单独实现了tryAcquire()
方法:#java.util.concurrent.locks.AbstractQueuedSynchronizer.acquire()
:if (!tryAcquire(arg) &&
acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
selfInterrupt();
}
ReentrantLock
中公平锁的tryAcquire()
方法#java.util.concurrent.locks.ReentrantLock.FairSync.tryAcquire()
:protected final boolean tryAcquire(int acquires) {
final Thread current = Thread.currentThread();
int c = getState();
if (c == 0) {
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;
}
}
state
值,如果不为0且获取锁的线程不是当前线程,直接返回false代表获取锁失败,被加入等待队列。如果是当前线程则可重入获取锁。state=0
则代表此时没有线程持有锁,执行hasQueuedPredecessors()
判断AQS
等待队列中是否有元素存在,如果存在其他等待线程,那么自己也会加入到等待队列尾部,做到真正的先来后到,有序加锁。具体代码如下:#java.util.concurrent.locks.AbstractQueuedSynchronizer.hasQueuedPredecessors()
:Node t = tail;
Node h = head;
Node s;
return h != t &&
((s = h.next) == null || s.thread != Thread.currentThread());
}
false
代表队列中没有节点或者仅有一个节点是当前线程创建的节点。返回true
则代表队列中存在等待节点,当前线程需要入队等待。head
是否等于tail
,如果队列中只有一个Node
节点,那么head
会等于tail
,接着判断head
的后置节点,这里肯定会是null
,如果此Node
节点对应的线程和当前的线程是同一个线程,那么则会返回false
,代表没有等待节点或者等待节点就是当前线程创建的Node
节点。此时当前线程会尝试获取锁。head
和tail
不相等,说明队列中有等待线程创建的节点,此时直接返回true
,如果只有一个节点,而此节点的线程和当前线程不一致,也会返回true
CPU
唤醒线程的开销,整体的吞吐效率会高点,CPU
也不必取唤醒所有线程,会减少唤起线程的数量ReentrantLock
默认创建非公平锁的原因之一了。Condition实现原理
AQS
所提供的核心功能,当然它还有很多其他的特性,这里我们来继续说下Condition
这个组件。Condition
是在java 1.5
中才出现的,它用来替代传统的Object
的wait()
、notify()
实现线程间的协作,相比使用Object
的wait()
、notify()
,使用Condition
中的await()
、signal()
这种方式实现线程间协作更加安全和高效。因此通常来说比较推荐使用Condition
AbstractQueueSynchronizer
中实现了Condition
中的方法,主要对外提供awaite(Object.wait())
和signal(Object.notify())
调用。Condition Demo示例
* ReentrantLock 实现源码学习
* @author 一枝花算不算浪漫
* @date 2020/4/28 7:20
*/
public class ReentrantLockDemo {
static ReentrantLock lock = new ReentrantLock();
public static void main(String[] args) {
Condition condition = lock.newCondition();
new Thread(() -> {
lock.lock();
try {
System.out.println("线程一加锁成功");
System.out.println("线程一执行await被挂起");
condition.await();
System.out.println("线程一被唤醒成功");
} catch (Exception e) {
e.printStackTrace();
} finally {
lock.unlock();
System.out.println("线程一释放锁成功");
}
}).start();
new Thread(() -> {
lock.lock();
try {
System.out.println("线程二加锁成功");
condition.signal();
System.out.println("线程二唤醒线程一");
} finally {
lock.unlock();
System.out.println("线程二释放锁成功");
}
}).start();
}
}
await()
方法挂起当前线程并释放锁,线程二获取锁后使用signal
唤醒线程一。Condition实现原理图解
demo
作为实例,执行的流程如下:await()
方法:#java.util.concurrent.locks.AbstractQueuedSynchronizer.ConditionObject.await()
:if (Thread.interrupted())
throw new InterruptedException();
Node node = addConditionWaiter();
int savedState = fullyRelease(node);
int interruptMode = 0;
while (!isOnSyncQueue(node)) {
LockSupport.park(this);
if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
break;
}
if (acquireQueued(node, savedState) && interruptMode != THROW_IE)
interruptMode = REINTERRUPT;
if (node.nextWaiter != null) // clean up if cancelled
unlinkCancelledWaiters();
if (interruptMode != 0)
reportInterruptAfterWait(interruptMode);
}
await()
方法中首先调用addConditionWaiter()
将当前线程加入到Condition
队列中。Condition
队列中的数据:Node t = lastWaiter;
if (t != null && t.waitStatus != Node.CONDITION) {
unlinkCancelledWaiters();
t = lastWaiter;
}
Node node = new Node(Thread.currentThread(), Node.CONDITION);
if (t == null)
firstWaiter = node;
else
t.nextWaiter = node;
lastWaiter = node;
return node;
}
Node
节点,waitStatus
为CONDITION
。接着会释放该节点的锁,调用之前解析过的release()
方法,释放锁后此时会唤醒被挂起的线程二,线程二会继续尝试获取锁。isOnSyncQueue()
方法判断当前节点是否为Condition
队列中的头部节点,如果是则调用LockSupport.park(this)
挂起Condition
中当前线程。此时线程一被挂起,线程二获取锁成功。signal()
方法:AQS
等待队列中已经没有了数据。if (!isHeldExclusively())
throw new IllegalMonitorStateException();
Node first = firstWaiter;
if (first != null)
doSignal(first);
}
doSignal()
方法来唤醒线程。do {
if ( (firstWaiter = first.nextWaiter) == null)
lastWaiter = null;
first.nextWaiter = null;
} while (!transferForSignal(first) &&
(first = firstWaiter) != null);
}
final boolean transferForSignal(Node node) {
if (!compareAndSetWaitStatus(node, Node.CONDITION, 0))
return false;
Node p = enq(node);
int ws = p.waitStatus;
if (ws > 0 || !compareAndSetWaitStatus(p, ws, Node.SIGNAL))
LockSupport.unpark(node.thread);
return true;
}
/**
* Inserts node into queue, initializing if necessary. See picture above.
* @param node the node to insert
* @return node's predecessor
*/
private Node enq(final Node node) {
for (;;) {
Node t = tail;
if (t == null) { // Must initialize
if (compareAndSetHead(new Node()))
tail = head;
} else {
node.prev = t;
if (compareAndSetTail(t, node)) {
t.next = node;
return t;
}
}
}
}
transferForSignal()
方法来看,通过上面的分析我们知道Condition
队列中只有线程一创建的一个Node
节点,且waitStatue
为CONDITION
,先通过CAS
修改当前节点waitStatus
为0,然后执行enq()
方法将当前线程加入到等待队列中,并返回当前线程的前置节点。CAS
修改当前节点的前置节点waitStatus
为SIGNAL
,并且唤醒当前线程。此时AQS
中等待队列数据为:await()
方法中的 while 循环。if (Thread.interrupted())
throw new InterruptedException();
Node node = addConditionWaiter();
int savedState = fullyRelease(node);
int interruptMode = 0;
while (!isOnSyncQueue(node)) {
LockSupport.park(this);
if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
break;
}
if (acquireQueued(node, savedState) && interruptMode != THROW_IE)
interruptMode = REINTERRUPT;
if (node.nextWaiter != null) // clean up if cancelled
unlinkCancelledWaiters();
if (interruptMode != 0)
reportInterruptAfterWait(interruptMode);
}
waitStatus
已经被修改为0,所以执行isOnSyncQueue()
方法会返回false
。跳出while
循环。acquireQueued()
方法,这里之前也有讲过,尝试重新获取锁,如果获取锁失败继续会被挂起。直到另外线程释放锁才被唤醒。boolean failed = true;
try {
boolean interrupted = false;
for (;;) {
final Node p = node.predecessor();
if (p == head && tryAcquire(arg)) {
setHead(node);
p.next = null; // help GC
failed = false;
return interrupted;
}
if (shouldParkAfterFailedAcquire(p, node) &&
parkAndCheckInterrupt())
interrupted = true;
}
} finally {
if (failed)
cancelAcquire(node);
}
}
Condition总结
Condition 可以精准的对多个不同条件进行控制,wait/notify 只能和 synchronized 关键字一起使用,并且只能唤醒一个或者全部的等待队列; Condition 需要使用 Lock 进行控制,使用的时候要注意 lock() 后及时的 unlock(),Condition 有类似于 await 的机制,因此不会产生加锁方式而产生的死锁出现,同时底层实现的是 park/unpark 的机制,因此也不会产生先唤醒再挂起的死锁,一句话就是不会产生死锁,但是 wait/notify 会产生先唤醒再挂起的死锁。
ReentrantLock
的实现方式和实现原理,而ReentrantLock
底层就是基于AQS
实现的,所以我们也对AQS
有了深刻的理解。Condition
的实现原理,基本上都是使用源码+绘图的讲解方式,尽量让大家更容易去理解。
推荐阅读