AQS原理及用法
1 AQS简介
- AQS全称为AbstractQueuedSynchronizer,是Java中的一个抽象类。
- AQS是一个用于构建锁、同步器、协作工具类的工具类(框架)。有了AQS之后,更多的协作工具类都可以方便得被写出来。
- 有了AQS,构建线程协作类就容易多了。
- 控制并发流程的类,都需要线程等待和唤醒的功能,这是这些类的共同特点,因此可以抽象出一个基类,这就是
AQS
。 - AQS广泛用于控制并发流程的类,如下图:
- 其中
Sync
是这些类中都有的内部类,其结构如下:
- 可以看到:
Sync
是AQS
的实现。 -
AQS
主要完成的任务:- (1)同步状态(比如说计数器)的原子性管理;
- (2)线程的阻塞和解除阻塞;
- (3)队列的管理。
2 AQS原理
- AQS最核心的就是三大部分:
- 状态:state;
- 控制线程抢锁和配合的FIFO队列(双向链表);
- 期望协作工具类去实现的获取/释放等重要方法(重写)。
状态state
- 这里state的具体含义,会根据具体实现类的不同而不同:比如在Semapore里,他表示剩余许可证的数量;在CountDownLatch里,它表示还需要倒数的数量;在ReentrantLock中,state用来表示“锁”的占有情况,包括可重入计数,当state的值为0的时候,标识该Lock不被任何线程所占有。
- state是volatile修饰的,并被并发修改,所以修改state的方法都需要保证线程安全,比如getState、setState以及compareAndSetState操作来读取和更新这个状态。这些方法都依赖于unsafe类。
FIFO队列
- 这个队列用来存放“等待的线程,AQS就是“排队管理器”,当多个线程争用同一把锁时,必须有排队机制将那些没能拿到锁的线程串在一起。当锁释放时,锁管理器就会挑选一个合适的线程来占有这个刚刚释放的锁。
- AQS会维护一个等待的线程队列,把线程都放到这个队列里,这个队列是双向链表形式。
实现获取/释放等方法
- 这里的获取和释放方法,是利用AQS的协作工具类里最重要的方法,是由协作类自己去实现的,并且含义各不相同;
- 获取方法:获取操作会以来state变量,经常会阻塞(比如获取不到锁的时候)
- 在Semaphore中,获取就是acquire方法,作用是获取一个许可证;
- 而在CountDownLatch里面,获取就是await方法,作用是等待,直到倒数结束;
- 释放方法
- 在Semaphore中,释放就是release方法,作用是释放一个许可证;
- 在CountDownLatch里面,获取就是countDown方法,作用是将倒数的数减一;
- 需要每个实现类重写tryAcquire和tryRelease等方法。
3 AQS在juc中的应用
AQS在juc中用法套路:
- 第一步:写一个类:想好协作的逻辑,实现获取/释放方法;
- 第二步:类的内部写一个Sync类,继承AbstractQueuedSychronizer;
- 第三步:根据是否独占来重写 tryAcquire/tryRelease 或 tryAcquireShared(int acquires) /tryReleaseShared(int release) 等方法,在之前写的获取/释放方法中调用、AQS的 acquire/release 或者 acquireShared/releaseShared 方法。
- 为了方便,下面分析中的
AbstractQueuedSynchronizer
类都用AQS
代替。
3.1 AQS在CountDownLatch中的应用
- 当我们新建一个
CountDownLatch
对象时:
CountDownLatch cdl = new CountDownLatch(3);
- 会在
cdl
内部创建一个Sync
对象,并将this.sync
中的state
设置为3,代表要倒数三次。
- 当我们调用
cdl.await()
,此时调用该语句的线程会陷入等待,原因分析如下: -
await()
方法中的内容如下:
public void await() throws InterruptedException {
// 位于CountDownLatch类中
sync.acquireSharedInterruptibly(1);
}
acquireSharedInterruptibly
位于AQS
类中,内容如下:
public final void acquireSharedInterruptibly(int arg)
throws InterruptedException {
// 位于AQS类中
if (Thread.interrupted()) throw new InterruptedException();
if (tryAcquireShared(arg) < 0)
doAcquireSharedInterruptibly(arg); // 把当前线程放入队列等待,并且让当前线程阻塞
}
tryAcquireShared(arg)
被AQS
的子类Sync
重写,内容如下:
protected int tryAcquireShared(int acquires) {
// 位于CountDownLatch类的内部类Sync中
return (getState() == 0) ? 1 : -1;
}
getState()
位于AQS
类中,内容如下:
protected final int getState() {
// 位于AQS类中
return state;
}
- 因为
getState()
返回的是 3,因此tryAcquireShared
会返回 -1,因此会执行doAcquireSharedInterruptibly
,然后当前线程进入阻塞。
- 当调用
countDown()
函数,会让state--
,如果值变为0,则会唤醒之前等待的线程。原因分析如下: -
countDown()
方法中的内容如下:
public void countDown() {
// 位于CountDownLatch类中
sync.releaseShared(1);
}
releaseShared
位于AQS
类中,内容如下:
public final boolean releaseShared(int arg) {
// 位于AQS类中
if (tryReleaseShared(arg)) {
doReleaseShared(); // 把之前阻塞的线程全部唤醒
return true;
}
return false;
}
tryReleaseShared(arg)
被AQS
的子类Sync
重写,内容如下:
protected boolean tryReleaseShared(int releases) {
// 位于CountDownLatch类的内部类Sync中
// Decrement count; signal when transition to zero
for (;;) {
int c = getState();
if (c == 0)
return false;
int nextc = c-1;
if (compareAndSetState(c, nextc))
return nextc == 0;
}
}
- 可以看到当
state
的值减少到 0 之后会唤醒之前阻塞进程。
3.1 AQS在Semaphore中的应用
- 当我们新建一个
Semaphore
对象时:
Semaphore s = new Semaphore(3);
- 会在
s
内部创建一个NonfairSync
对象(继承自Sync
),并将this.sync
中的state
设置为3,代表有三个通行证。
- 当我们调用
s.acquire(2)
,会获取到两个许可证,分析如下: -
acquire(int permits)
方法中的内容如下:
public void acquire(int permits) throws InterruptedException {
// 位于Semaphore类中
if (permits < 0) throw new IllegalArgumentException();
sync.acquireSharedInterruptibly(permits);
}
acquireSharedInterruptibly
位于AQS
类中,内容如下:
public final void acquireSharedInterruptibly(int arg)
throws InterruptedException {
// 位于AQS类中
if (Thread.interrupted()) throw new InterruptedException();
if (tryAcquireShared(arg) < 0)
doAcquireSharedInterruptibly(arg); // 把当前线程放入队列等待,并且让当前线程阻塞
}
tryAcquireShared(arg)
被AQS
的子类Sync
重写,内容如下:
protected int tryAcquireShared(int acquires) {
// 位于Semaphore类的内部类NonfairSync中
return nonfairTryAcquireShared(acquires);
}
nonfairTryAcquireShared
内容如下:
final int nonfairTryAcquireShared(int acquires) {
// 位于Semaphore类的内部类Sync中
for (;;) {
int available = getState();
int remaining = available - acquires;
if (remaining < 0 ||
compareAndSetState(available, remaining))
return remaining;
}
}
- 上面的
available
就是state
,当我们想要获取的许可证数量acquires
小于等于state
时,返回正数,在函数acquireSharedInterruptibly
判断不成立,不会陷入阻塞,并将state
更新为最新的值;否则如果acquires
大于state
时,返回负数,当前线程会陷入阻塞。
- 当我们调用
s.release(2)
,会获取到两个许可证,分析如下: -
release(int permits)
方法中的内容如下:
public void release(int permits) {
// 位于Semaphore类中
if (permits < 0) throw new IllegalArgumentException();
sync.releaseShared(permits);
}
releaseShared
位于AQS
类中,内容如下:
public final boolean releaseShared(int arg) {
// 位于AQS类中
if (tryReleaseShared(arg)) {
doReleaseShared(); // 把之前阻塞的线程全部唤醒
return true;
}
return false;
}
tryAcquireShared(arg)
被AQS
的子类Sync
重写,内容如下:
protected final boolean tryReleaseShared(int releases) {
// 位于Semaphore类的内部类Sync中
for (;;) {
int current = getState();
int next = current releases;
if (next < current) // overflow
throw new Error("Maximum permit count exceeded");
if (compareAndSetState(current, next))
return true;
}
}
- 该函数会将
state
的值增加releases
,也就是增加许可证的数量。
3.1 AQS在ReentrantLock中的应用
- 当我们新建一个
ReentrantLock
对象时:
ReentrantLock r = new ReentrantLock();
- 会在
r
内部创建一个NonfairSync
对象sync
(继承自Sync
)。
- 当我们调用
r.lock()
,分析如下: -
r.lock()
方法中的内容如下:
public void lock() {
// 位于ReentrantLock类中
sync.lock();
}
sync.lock()
内容如下:
final void lock() {
// 位于ReentrantLock类的内部类NonfairSync中
if (compareAndSetState(0, 1)) // CAS只有在没有人持有这把锁的时候才能成功
setExclusiveOwnerThread(Thread.currentThread()); // 把当前线程设置为持有锁的线程
else
acquire(1);
}
acquire(1)
位于AQS
类中,内容如下:
public final void acquire(int arg) {
// 位于AQS类中
if (!tryAcquire(arg) &&
acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
selfInterrupt();
}
tryAcquire
内容如下:
protected final boolean tryAcquire(int acquires) {
// 位于ReentrantLock类的内部类NonfairSync中
return nonfairTryAcquire(acquires);
}
nonfairTryAcquire
内容如下:
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) // overflow
throw new Error("Maximum lock count exceeded");
setState(nextc);
return true;
}
return false;
}
- 由该函数可以看出,当
state
为0时,代表没有任何人持有该锁,获取该锁;否则如果当前线程正好是该锁的持有者,重入。
- 当我们调用
r.unlock()
,分析如下: -
r.unlock()
方法中的内容如下:
public void unlock() {
// // 位于ReentrantLock类中
sync.release(1);
}
sync.release(1)
内容如下:
public final boolean release(int arg) {
// 位于AQS类中
if (tryRelease(arg)) {
Node h = head;
if (h != null && h.waitStatus != 0)
unparkSuccessor(h);
return true;
}
return false;
}
tryRelease
被AQS
的子类Sync
重写,内容如下:
protected final boolean tryRelease(int releases) {
// 位于ReentrantLock类的内部类Sync中
int c = getState() - releases; // 将重入次数减少releases
if (Thread.currentThread() != getExclusiveOwnerThread()) // 判断当前线程是不是持有锁的线程
throw new IllegalMonitorStateException();
boolean free = false;
if (c == 0) {
// 只有当重入次数为0时,才去释放锁
free = true;
setExclusiveOwnerThread(null); // 设置当前持有这把锁的线程为null
}
setState(c);
return free;
}
- 可以看到
state
的值被减去releases
,如果state
变为0,则设置当前持有这把锁的线程为null。
版权声明:本文内容由互联网用户自发贡献,该文观点仅代表作者本人。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容, 请发送邮件至 举报,一经查实,本站将立刻删除。
发布者:全栈程序员栈长,转载请注明出处:https://javaforall.cn/234952.html原文链接:https://javaforall.cn