JAVA多线程番外篇4、AbstractQueuedSynchronizer

AbstractQueuedSynchronizer 一个用来构建锁和同步器的框架&＃xff0c;使用AQS能简单且高效地构造出应用广泛的大量的同步器。

• ReentrantLock

• Semaphore&＃xff0c;

• ReentrantReadWriteLock

• SynchronousQueue

• FutureTask

AbstractQueuedSynchronizer继承自AbstractOwnableSynchronizer抽象类&＃xff0c;并且实现了Serializable接口&＃xff0c;可以进行序列化。
AbstractOwnableSynchronizer抽象类中&＃xff0c;可以设置独占资源线程和获取独占资源线程。分别为setExclusiveOwnerThread与getExclusiveOwnerThread方法&＃xff0c;这两个方法会被子类调用。

2.1 机制

AQS内部维护了一个CLH队列来管理锁。线程会首先尝试获取锁&＃xff0c;如果失败就将当前线程及等待状态等信息包装成一个node节点加入到同步队列sync queue里。 接着会不断的循环尝试获取锁&＃xff0c;条件是当前节点为head的直接后继才会尝试。如果失败就会阻塞自己直到自己被唤醒。而当持有锁的线程释放锁的时候&＃xff0c;会唤醒队列中的后继线程。

2.2 结构

AQS底层使用了模板方法设计模式。AQS类中的其他方法都是final &＃xff0c;所以这些无法被其他类使用。AQS提供几个方法&＃xff0c;由子类覆写&＃xff0c;如果没有覆写&＃xff0c;这些方法都抛出 UnsupportedOperationException&＃xff0c;自定义的同步器继承AbstractQueuedSynchronizer并重写指定的方法即可。AQS定义两种资源共享方式&＃xff1a;Exclusive&＃xff08;独占&＃xff0c;只有一个线程能执行&＃xff0c;如ReentrantLock&＃xff09;和Share&＃xff08;共享&＃xff0c;多个线程可同时执行&＃xff0c;如Semaphore/CountDownLatch&＃xff09;。

isHeldExclusively()&＃xff1a;该线程是否正在独占资源。只有用到condition才需要去实现它。
tryAcquire(int)&＃xff1a;独占方式。尝试获取资源&＃xff0c;成功则返回true&＃xff0c;失败则返回false。
tryRelease(int)&＃xff1a;独占方式。尝试释放资源&＃xff0c;成功则返回true&＃xff0c;失败则返回false。
tryAcquireShared(int)&＃xff1a;共享方式。尝试获取资源。负数表示失败&＃xff1b;0表示成功&＃xff0c;但没有剩余可用资源&＃xff1b;正数表示成功&＃xff0c;且有剩余资源。
tryReleaseShared(int)&＃xff1a;共享方式。尝试释放资源&＃xff0c;如果释放后允许唤醒后续等待结点返回true&＃xff0c;否则返回false。

• 以ReentrantLock为例&＃xff0c;state初始化为0&＃xff0c;表示未锁定状态。A线程lock()时&＃xff0c;会调用tryAcquire()独占该锁并将state&＃43;1。此后&＃xff0c;其他线程再tryAcquire()时就会失败&＃xff0c;直到A线程unlock()到state&＃61;0&＃xff08;即释放锁&＃xff09;。释放锁之前&＃xff0c;A线程自己是可以重复获取此锁的&＃xff08;state会累加&＃xff09;&＃xff0c;获取多少次就要释放多么次。
• 以CountDownLatch以例&＃xff0c;任务分为N个子线程去执行&＃xff0c;state也初始化为N&＃xff08;注意N要与线程个数一致&＃xff09;。每个子线程执行完后countDown()一次&＃xff0c;state会CAS减1。等到所有子线程都执行完后(即state&＃61;0)。
• AQS也支持自定义同步器同时实现独占和共享两种方式&＃xff0c;如ReentrantReadWriteLock。

2.2.1 CLH

CLH&＃xff08; Craig, Landin, and Hagersten locks &＃xff09;&＃xff0c;是一个虚拟的双向队列&＃xff08;虚拟的双向队列即不存在队列实例&＃xff0c;仅存在结点之间的关联关系&＃xff09;。

AQS就是使用CLH锁。线程请求共享资源时&＃xff0c;将线程封装成一个一个结点(Node)&＃xff0c;放入CLH锁队列来实现锁的分配。。

2.2.2 Node

Node 存储线程状态CANCELLED、SIGNAL、CONDITION、PROPAGATE。

• CANCELLED(1)&＃xff1a;表示当前结点已取消调度。当timeout或被中断&＃xff08;响应中断的情况下&＃xff09;&＃xff0c;会触发变更为此状态&＃xff0c;进入该状态后的结点将不会再变化。
• SIGNAL(-1)&＃xff1a;表示后继结点在等待当前结点唤醒。后继结点入队时&＃xff0c;会将前继结点的状态更新为SIGNAL。
• CONDITION(-2)&＃xff1a;表示结点等待在Condition上&＃xff0c;当其他线程调用了Condition的signal()方法后&＃xff0c;CONDITION状态的结点将从等待队列转移到同步队列中&＃xff0c;等待获取同步锁。
• PROPAGATE(-3)&＃xff1a;共享模式下&＃xff0c;前继结点不仅会唤醒其后继结点&＃xff0c;同时也可能会唤醒后继的后继结点。线程都会被封装成一个Node结点&＃xff0c;放入队列。每个节点包含了一个Thread类型的引用&＃xff0c;并且每个节点都存在一个状态。

2.2.3 示例


import java.util.concurrent.locks.AbstractQueuedSynchronizer;public class AQSTest implements java.io.Serializable {private static class MyLock extends AbstractQueuedSynchronizer {public boolean isHeldExclusively() {return getState() &＃61;&＃61; 1;}public boolean tryAcquire(int acquires) {if (compareAndSetState(0, 1)) {setExclusiveOwnerThread(Thread.currentThread());return true;}return false;}protected boolean tryRelease(int releases) {if (getState() &＃61;&＃61; 0) throw new IllegalMonitorStateException();setExclusiveOwnerThread(null);setState(0);return true;}}private final MyLock mylock &＃61; new MyLock();public void lock() {mylock.acquire(1);}public boolean tryLock() {return mylock.tryAcquire(1);}public void unlock() {mylock.release(1);}public boolean isLocked() {return mylock.isHeldExclusively();}
}

总结

AQS是JUC中很多同步组件的构建基础&＃xff0c;简单来讲&＃xff0c;它内部实现主要是状态变量state和一个FIFO队列来完成。

多线程系列在github上有一个开源项目&＃xff0c;主要是本系列博客的实验代码。
https://github.com/forestnlp/concurrentlab

如果您对软件开发、机器学习、深度学习有兴趣请关注本博客&＃xff0c;将持续推出Java、软件架构、深度学习相关专栏。

您的支持是对我最大的鼓励。