AQS原理及用法_aqs是什么意思

全栈程序员站长

发布于 2022-11-15 11:41:56

83400

代码可运行

文章被收录于专栏：全栈程序员必看

运行总次数：0

代码可运行

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

本文参与腾讯云自媒体同步曝光计划，分享自作者个人站点/博客。

原始发表：2022年11月1日，如有侵权请联系 cloudcommunity@tencent.com 删除

unix

本文分享自作者个人站点/博客前往查看

如有侵权，请联系 cloudcommunity@tencent.com 删除。

本文参与腾讯云自媒体同步曝光计划，欢迎热爱写作的你一起参与！

unix

登录后参与评论

0 条评论

热度

AQS原理及用法_aqs是什么意思

AQS原理及用法_aqs是什么意思

AQS原理及用法

1 AQS简介

2 AQS原理

3 AQS在juc中的应用

3.1 AQS在CountDownLatch中的应用

3.1 AQS在Semaphore中的应用

3.1 AQS在ReentrantLock中的应用

社区

活动

资源

关于

腾讯云开发者

热门产品

热门推荐

更多推荐