抽象的队列同步器，一个多线程同步框架，撑起了JUC的半片天。
AQS的思想是核Synchronized基本一样的，只不过一个是c++ Jvm 层面实现了， 而AQ则是Java层面的，当然底层的api还是调用的C++函数，那你说吧，操作系统提供的函数不都是C或者C++来写的吗，所以java需要使用操作系统的线程同步函数就得借助c++，通过Unsafe类的nactive调用操作系统的线程同步函数。
AQS的关键组件与Synchronized进行对比
| Synchronized | AQS | |
|---|---|---|
| 锁的拥有者 | owner | exclusiveOwnerThread |
| 重入次数 | recursions | state |
| 同步队列 | cxq | CLH |
| 同步队列 | entrySet | CLH |
| 条件队列（等待队列） | waitSet | 条件队列 |
这只是大概的草图，真正的CLH队列远比这复杂
呃 ，入队分为三个步骤：
尝试获取锁， 获取锁失败进入队列， 进入队列之后阻塞， 在这三步中， 尝试获取锁在AQS中是不做实现的，需要子类实现这个方法， 而进入队列 和 阻塞 是AQS中抽象的方法，即不需要子类实现，AQS中已经做了实现， 那么公平锁和非公平锁的关键就是在于子类如何实现这个尝试获取锁， 而公平和非公平性主要就是体现在公平锁在尝试获取锁的时候需要判断队列中有没有元素，如果有就不去cas state 进行抢锁，需要进入队列阻塞， 非公平锁呢直接就是cas，不看队列中是否有元素，如果cas失败了也是同样需要进入队列阻塞
其中需要注意的是： 抢锁，入队 都是共享资源所以需要使用cas无锁的方式保证线程安全，cas失败的需要循环进行补偿
模拟QAS代码 MyAbstractQueueSynchronizerpublic class MyAbstractQueueSynchronizer {
private Thread ownThread;
private volatile int state;
private volatile Node head;
private volatile Node tail;
static class Node {
private static Node EXCLUSIVE = null;
private Node nextWaiter;
private volatile Thread thread;
private volatile Node prev;
private volatile Node next;
private static int CANCELLED = 1;//当前节点处于取消状态
private static int SINGAL = -1; //当前节点的前驱节点的状态属于这个状态 当前节点才能被阻塞
private volatile int waitState;
Node() {
}
Node(Thread thread, Node mode) {
this.thread = thread;
this.nextWaiter = mode;
}
public Node preNode() {
Node p = prev;
if (p == null) {
throw new NullPointerException();
}
return p;
}
}
public Thread getOwnThread() {
return ownThread;
}
public void setOwnThread(Thread ownThread) {
this.ownThread = ownThread;
}
public void acquire(int arg) {
//尝试快速获取锁，如果失败就需要入队自旋
if (!tryAcquire(arg)
&& acquireQueue(addNode(Node.EXCLUSIVE), arg)) {//独占模式入队
Thread.currentThread().interrupt(); //中断补偿
}
}
public Node addNode(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;
}
}
//走到这里说明 队列没有初始化或者cas失败
casEnqueue(node);
return node;
}
private void casEnqueue(Node node) {
while (true) {
Node tempTail = tail;
if (tempTail == null) {
if (compareAndSetHead(new Node())) {//为什么要new一个新的空节点？
//初始化完毕
tail = head;
}
//不需要初始化, cas尾入
} else {
node.prev = tempTail;
if (compareAndSetTail(tempTail, node)) {
//入队成功
tempTail.next = node;
return;
}
}
}
}
public boolean acquireQueue(Node node, int arg) {
boolean isException = true; //默认有异常，只有获取锁之后才能出while循环
try {
boolean interrupted = false;
while (true) {
//获取前驱, 是否可以cas去抢锁
Node preNode = node.preNode();
if (preNode == head
&& tryAcquire(arg)) { //只有前驱节点的后继才有资格去获取锁
//获取到锁, 把自己的节点置为头节点
setHead(node);
preNode.next = null;
isException = false;
return interrupted;
}
//2、阻塞
if (shouldParkAfterFailedAcquire(node.prev, node) && park()) {
interrupted = true;
}
}
} finally {
if (isException) {
//异常 取消就是把自己从队列中摘除
cancelAcquire(node);
}
}
}
private void cancelAcquire(Node node) {
Node pred = node.prev;
while (pred.waitState > 0) {
node.prev = pred = pred.prev;
}
pred.next = node;
Node nodeNext = pred.next;
node.waitState = Node.CANCELLED;
if (node == tail && compareAndSetTail(node, pred)) {
//修改尾节点 为啥需要cas 因为尾巴是入队多个线程可以
compareAndSetNext(pred, nodeNext, null);
// node 为中间节点
} else {
int wait = pred.waitState;
if (pred != head
&& (wait == Node.SINGAL || compareAndSetWaitState(pred, wait, Node.SINGAL))
&& node.thread != null) {
Node next = node.next;
if (next != null && next.waitState <= 0) {
compareAndSetNext(pred, nodeNext, next);
}
} else {
unparkSuccessor(node);
}
node.next = node;
}
}
private void unparkSuccessor(Node node) {
int waitState = node.waitState;
if (waitState < 0) {
compareAndSetWaitState(node, waitState, 0);
}
Node s = node.next;
if (s == null || s.waitState > 0) {
Node t = tail;
for (; t != null && t != node; t = t.prev) {
if (t.waitState <= 0) {
s = t;
}
}
}
if (s != null) {
LockSupport.unpark(s.thread);
}
}
private boolean park() {
LockSupport.park(this);//唤醒方式： 1、unpark 2、中断
return Thread.interrupted(); //返回当前线程的中断标识，并清楚中断
}
private boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
int waitState = pred.waitState;
if (waitState == Node.SINGAL) { //前驱节点的状态是SINGAL，当前节点才能被阻塞
return true;
}
if (waitState > 0) {
do {
node.prev = pred = pred.prev;
} while (pred.waitState > 0);
pred.next = node;
} else { //说明是初始化状态
compareAndSetWaitState(node, waitState, Node.SINGAL);
}
return false;
}
public boolean release(int arg) {
// state - arg
if (tryRelease(arg)) {
//唤醒头节点后继节点
Node h = head;
if (h != null && h.waitState != 0) {
unparkSuccessor(h);
}
return true;
}
return false;
}
protected boolean tryRelease(int i) {
throw new UnsupportedOperationException();
}
public boolean hasQueuePred() {
Node t = tail;
Node h = head;
Node s;
return h != t && ((s = h.next) == null || s.thread != Thread.currentThread());
}
protected boolean tryAcquire(int arg) {
throw new UnsupportedOperationException("不支持");
}
public boolean compareAndSetState(int exepect, int update) {
return unsafe.compareAndSwapInt(this, stateAddr, exepect, update);
}
public boolean compareAndSetTail(Node expect, Node update) {
return unsafe.compareAndSwapObject(this, tailAddr, expect, update);
}
public boolean compareAndSetHead(Node update) {
return unsafe.compareAndSwapObject(this, headAddr, null, update);
}
public boolean compareAndSetWaitState(Node node, int expect, int update) {
return unsafe.compareAndSwapInt(node, waitStateAddr, expect, update);
}
public boolean compareAndSetNext(Node node, Node expect, Node update) {
return unsafe.compareAndSwapObject(node, nextAddr, expect, update);
}
private static Unsafe unsafe = MyJUCUtil.getUnsafe();
private static long stateAddr;
private static long tailAddr;
private static long headAddr;
private static long waitStateAddr;
private static long nextAddr;
static {
try {
stateAddr = unsafe.objectFieldOffset(MyAbstractQueueSynchronizer.class.getDeclaredField("state"));
tailAddr = unsafe.objectFieldOffset(MyAbstractQueueSynchronizer.class.getDeclaredField("tail"));
headAddr = unsafe.objectFieldOffset(MyAbstractQueueSynchronizer.class.getDeclaredField("head"));
waitStateAddr = unsafe.objectFieldOffset(MyAbstractQueueSynchronizer.Node.class.getDeclaredField("waitState"));
nextAddr = unsafe.objectFieldOffset(Node.class.getDeclaredField("next"));
} catch (NoSuchFieldException e) {
e.printStackTrace();
}
}
public int getState() {
return state;
}
public void setState(int state) {
this.state = state;
}
public Node getHead() {
return head;
}
public void setHead(Node node) {
this.head = node;
node.thread = null;
node.prev = null;
}
public Node getTail() {
return tail;
}
public void setTail(Node tail) {
this.tail = tail;
}
}
MyReentrantLock
public class MyReentrantLock implements Lock {
private MySync mySync;
public MyReentrantLock(){
this(true);
}
public MyReentrantLock(boolean fair){
mySync = fair? new MyFairSync() : new MyUnFairSync();
}
abstract static class MySync extends MyAbstractQueueSynchronizer{
abstract void lock();
@Override
protected boolean tryRelease(int i) {
int state = getState() - i;
if (Thread.currentThread() != getOwnThread()){
throw new IllegalMonitorStateException();
}
boolean free = false;
if (state == 0){
free = true;
setOwnThread(null);
}
setState(state);
return free;
}
}
static class MyFairSync extends MySync{
@Override
protected boolean tryAcquire(int arg) {
Thread currentThread = Thread.currentThread();
int state = getState();
if (state == 0){
//检查队列, 没人排队才进行cas， 有人排队就去入队
if (!hasQueuePred()
&&compareAndSetState(state, arg)){
setOwnThread(currentThread);
return true;
}
//重入
}else if (getOwnThread() == currentThread){
int newState = state + arg;
if (newState < 0){
throw new Error("数据异常！");
}
setState(newState);
}
return false;
}
@Override
void lock() {
acquire(1);
}
}
static class MyUnFairSync extends MySync{
@Override
protected boolean tryAcquire(int arg) {
Thread currentThread = Thread.currentThread();
int state = getState();
if (state == 0){
//非公平直接cas
if (compareAndSetState(state, arg)){
setOwnThread(currentThread);
return true;
}
//重入
}else if (getOwnThread() == currentThread){
int newState = state + arg;
if (newState < 0){
throw new Error("数据异常！");
}
setState(newState);
}
return false;
}
@Override
void lock() {
if (compareAndSetState(0, 1)){
setOwnThread(Thread.currentThread());
return;
}
acquire(1);
}
}
@Override
public void lock() {
mySync.lock();
}
@Override
public void lockInterruptibly() throws InterruptedException {
}
@Override
public boolean tryLock() {
return false;
}
@Override
public boolean tryLock(long time, TimeUnit unit) throws InterruptedException {
return false;
}
@Override
public void unlock() {
mySync.release(1);
}
@Override
public Condition newCondition() {
return null;
}
}
流程总结
设计模式
获取锁的方法acquire里面用到了三个方法
- tryAcquire 子类实现获取锁的逻辑
- addNode cas入队，AQS自己实现
- acquireQueue 自旋抢锁阻塞 ， AQS实现
public void acquire(int arg) {
//尝试快速获取锁，如果失败就需要入队自旋
if (!tryAcquire(arg)
&& acquireQueue(addNode(Node.EXCLUSIVE), arg)) {//独占模式入队
Thread.currentThread().interrupt(); //中断补偿
}
}
可以看出这里使用了模板方法的设计模式，获取锁的流程三步中 尝试获取这个方法交由子类自己实现 ，这样就是实现公平锁和非公平锁。
呃， MyReentrantLock实现Lock接口，但是里面具体的方法实现是交给公平锁和非公平锁的， 有点类似于委托模式。
中断补偿呃，知识前提就是park是可以被unpark和中断唤醒的， 特别是中断， 如果线程被中断了，且这个中断标志没有被消除，那么在这个期间park()无法生效。 这样就会导致在阻塞获取锁中一直循环耗费cpu，所以需要一个中断补偿。逻辑就是如果这个线程是被中断唤醒，那么先消除中断位， 然后去抢锁，抢不到就可以继续挂起，如果抢到了，就进行中断补偿。
为什么非要进行中断补偿：如果自己的业务逻辑里面需要对中断进行判断，线程唤醒之后不补偿就是会出现问题。
cancelAcquire 取消获取锁什么时候会执行取消入队的逻辑进入是应为抛出了异常，再代码逻辑中，自旋阻塞获取锁中，获取锁调用的是tryAcquire这个方法，这个方法是由子类实现的，所以子类实现的逻辑中可能会抛出异常，当抛出异常的时候这个节点需要从队列中取消掉。呃呃，取消掉的逻辑是有些复杂的，应为node的位置情况可以分为：尾巴、中间、头节点的后继需要分别考虑。
hasQueuePred这个方法的作用是判断队列中有没有元素，主要是公平锁需要用到，因为公平锁需要判断队列中有没有元素才去抢锁。
那如何判断队列中有没有元素呢？
Node t = tail;
Node h = head;
Node s;
return h != t && ((s = h.next) == null || s.thread != Thread.currentThread());
思考：呃呃，说实在这个逻辑判断真的是不好懂。
如果队列被初始化了也就是 h!=t ，那么说明已经有线程进入， 但是自己又没有进入队列，此时另外来了一个线程，你说这个线程认为队列此时为空吗。所以公平来说就是这个线程也同样去入队。
而第二个判断条件呢？是不是同样很奇怪，我去s都不为空了，那说明队列中线程啊， 想想一种情况就是头节点后面的节点被唤醒之后去tryAcquire，同样需要进行判断队列中有没有元素，但是这时候他自己就是要被唤醒的第一个线程，所以需要这种情况下是认为队列中没有元素的。
呃，一个节点要出队，说明他后面的节点要成为头节点了（获取了锁），这个时候就会把前头节点剔除队列
什么情况下入队之后能挂起呃， B线程入队了，能park的前提就是B的前驱节点的状态是SIGNAL， 这个状态代表B的前驱节点有唤醒B的义务。所以B需要挂起的前提就是 自己的前驱节点的状态时SIGNAL 或者不是但是自己能后cas修改自己前驱节点的状态位SIGNAL。这样自己才能安安稳稳的park，要不然没人来唤醒自己，就完了。
总结呃， AQS还是挺难的。 volatile、cas 、数据结构、多线程思维、中断知识、park和unpark、CLH、设计模式。
