AQS是什么？
AbstractQueuedSynchronizer简称AQS是一个抽象同步框架，可以用来实现一个依赖状态的同步器。
Provides a framework for implementing blocking locks and related synchronizers (semaphores, events, etc) that rely on first-in-first-out (FIFO) wait queues. This class is designed to be a useful basis for most kinds of synchronizers that rely on a single atomic int value to represent state.
提供一个框架，用于实现依赖先进先出（FIFO）等待队列的阻塞锁和相关同步器（信号量，事件等）。该类被设计为大多数类型的同步器的有用依据，这些同步器依赖于单个原子int值来表示状态。
This class supports either or both a default exclusive mode and a shared mode.
此类支持默认独占模式和共享模式。
Even though this class is based on an internal FIFO queue, it does not automatically enforce FIFO acquisition policies. The core of exclusive synchronization takes the form:
即使这个类基于内部FIFO队列，它也不会自动执行FIFO采集策略。 排他同步的核心形式如下：
1Acquire: 2 while (!tryAcquire(arg)) { 3 enqueue thread if it is not already queued; 4 possibly block current thread; 5 } 6 7 Release: 8 if (tryRelease(arg)) 9 unblock the first queued thread; 10 11（共享模式类似，但可能包含级联信号。）
AQS 结构
1 /** 2 * 当前持有独占锁的线程 3 */ 4 private transient Thread exclusiveOwnerThread; 5 6 /** 7 * 等待队列的头结点，一般为当前持有锁的线程 （volatile） 8 */ 9 private transient volatile Node head; 10 11 /** 12 * 等待队列的尾结点，每次新结点进来，都是加到尾部，形成了链表 （volatile） 13 */ 14 private transient volatile Node tail; 15 16 /** 17 * 锁的状态，0 表示没有被占用，具体由子类实现 （volatile） 18 */ 19 private volatile int state; 20 21 // 等待队列的结点 22 static final class Node { 23 // 标识节点当前在共享模式下 24 static final Node SHARED = new Node(); 25 // 标识节点当前在独占模式下 26 static final Node EXCLUSIVE = null; 27 28 // 大于等于0 表明这个结点的线程取消了争抢这个锁，不需要去唤醒 29 /** 30 * The values are arranged numerically to simplify use. 31 * Non-negative values mean that a node doesn't need to 32 * signal. So, most code doesn't need to check for particular 33 * values, just for sign. 34 * 35 * The field is initialized to 0 for normal sync nodes, and 36 * CONDITION for condition nodes. It is modified using CAS 37 * (or when possible, unconditional volatile writes). 38 */ 39 volatile int waitStatus; 40 41 // 前一个结点 42 volatile Node prev; 43 44 // 下一个结点 45 volatile Node next; 46 47 // 等待的线程 48 volatile Thread thread; 49 50 ... 51 }
CLH队列 -- Craig, Landin, and Hagersten lock queue
CLH是一个非阻塞的 FIFO 队列。也就是说往里面插入或移除一个节点的时候，在并发条件下不会阻塞，而是通过自旋锁和 CAS 保证节点插入和移除的原子性。
源码导读：ReentrantLock 公平锁
1 public static void main(String[] args) { 2 ReentrantLock lock = new ReentrantLock(true); 3 4 lock.lock(); 5 6 lock.unlock(); 7 }
我们先看一下lock, 然后是 unlock
获得锁的流程和源码解读
1// ReentrantLock.FairSync 2 final void lock() { 3 acquire(1); 4 } 5 6// AQS 7 public final void acquire(int arg) { 8 // 1) 当前线程尝试获得锁，如果成功，结束 9 // 2）如果获得锁失败，当前线程加入等待队列，加入后不断循环监视上一个结点状态 10 // 如果上一个结点是头结点head ，-尝试获得锁，获得成功，跳出循环 11 // 否则，根据上一个结点的waitStatus，进行调整，包括挂起当前线程，或调整上一个结点为非取消状态的结点 12 // 3）最后当前线程发起中断 Thread.currentThread().interrupt() 13 if (!tryAcquire(arg) && 14 acquireQueued(addWaiter(Node.EXCLUSIVE), arg)) 15 selfInterrupt(); 16 } 17 18// ReentrantLock.FairSync 19// 当前线程尝试获得锁 20// 返回true：1.没有线程在等待锁；2.重入锁，线程本来就持有锁，也就可以理所当然可以直接获取 21 protected final boolean tryAcquire(int acquires) { 22 final Thread current = Thread.currentThread(); 23 int c = getState(); 24 if (c == 0) { 25 // 1）查看是否有等待已经在head后面等待锁了，如果有当前线程放弃抢占锁 26 // 2）如果没有线程等待，CAS 尝试修改 state, 如果成功，获得锁 27 // 3）设置的当前线程为 当前持有独占锁的线程 28 if (!hasQueuedPredecessors() && 29 compareAndSetState(0, acquires)) { 30 setExclusiveOwnerThread(current); 31 return true; 32 } 33 } 34 // 4）如果state不为0, 说明为可重入，只需要判断 当前线程 是否为 当前持有独占锁的线程 35 // 5）如果是，可重入，state增加 36 else if (current == getExclusiveOwnerThread()) { 37 int nextc = c + acquires; 38 if (nextc < 0) 39 throw new Error("Maximum lock count exceeded"); 40 setState(nextc); 41 return true; 42 } 43 //6） 其他情况，获得锁失败 44 return false; 45 } 46 47// AQS 48// 构造结点，采用CAS 加入等待队列的尾部 49 private Node addWaiter(Node mode) { 50 Node node = new Node(Thread.currentThread(), mode); 51 Node pred = tail; 52 //1） 如果等待队列不为空，则CAS 将当前线程加入等待队列的尾部 53 if (pred != null) { 54 node.prev = pred; 55 if (compareAndSetTail(pred, node)) { 56 pred.next = node; 57 return node; 58 } 59 } 60 // 2）如果等待队列为空，则CAS 新建初始化CLH队列，并当前线程加入等待队列的尾部 61 enq(node); 62 return node; 63 } 64 65// AQS 66// 如果获得锁失败，当前线程加入等待队列 67 final boolean acquireQueued(final Node node, int arg) { 68 boolean failed = true; 69 try { 70 boolean interrupted = false; 71 // 注意，这是一个循环，只有获得锁，或抛异常才会退出 72 // 如果上一个节点是头结点 head，则尝试获得锁 73 // 否则，如果当前线程需要挂起，则挂起等待锁的释放 74 for (;;) { 75 // 1）查看当前结点的上一个结点，如果是头结点head，尝试获得锁 76 final Node p = node.predecessor(); 77 if (p == head && tryAcquire(arg)) { 78 setHead(node); 79 p.next = null; // help GC 80 failed = false; 81 return interrupted; 82 } 83 // 2）如果当前结点的上一个结点不是头结点head，或获得锁失败，则判断一下是否需求挂起当前线程？ 84 // 3）如果需要挂起，则 park 挂起当前线程 85 if (shouldParkAfterFailedAcquire(p, node) && 86 parkAndCheckInterrupt()) 87 interrupted = true; 88 } 89 } finally { 90 // 4）如果获得锁失败，并且抛异常，则当前线程取消抢占 91 if (failed) 92 cancelAcquire(node); 93 } 94 } 95 96// AQS 97 private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) { 98 int ws = pred.waitStatus; 99 // 1）如果上一个结点的状态是 -1, 则当前结点 需要挂起 100 if (ws == Node.SIGNAL) 101 return true; 102 // 2）如果上一个结点的状态 大于0,表示取消抢占了， 那么循环往前找到一个结点的状态 <=0，然后将当前结点的上一个结点 设置为 找到的结点 103 if (ws > 0) { 104 do { 105 node.prev = pred = pred.prev; 106 } while (pred.waitStatus > 0); 107 pred.next = node; 108 } 109 // 3）如果上一个结点<=0且不等于-1, 那么结点状态为 0（加入等待队列的初始状态）, -2,-3， 110 // 此时需要 用CAS将上一个节点的waitStatus设置为Node.SIGNAL(也就是-1) 111 else { 112 compareAndSetWaitStatus(pred, ws, Node.SIGNAL); 113 } 114 return false; 115 } 116 117// AQS 118// 挂起当前线程，并测试返回当前线程是否中断状态 119 private final boolean parkAndCheckInterrupt() { 120 LockSupport.park(this); 121 return Thread.interrupted(); 122 }
释放锁的流程和源码解读
1// ReentrantLock 2 final void unlock() { 3 sync.release(1); 4 } 5 6// AQS 7 public final boolean release(int arg) { 8 // 1）释放锁，尝试修改state的值 9 if (tryRelease(arg)) { 10 Node h = head; 11 // 2）如果等待队列有线程，则将头部的结点状态修正，并唤醒头结点的下一个不是取消状态的线程 12 if (h != null && h.waitStatus != 0) 13 unparkSuccessor(h); 14 return true; 15 } 16 return false; 17 } 18 19// ReentrantLock 20// 判断当前线程 是否为 当前持有独占锁的线程 21// 如果是，修改state的值，包括可重入锁减一 22// 如果不是，抛出异常 23 protected final boolean tryRelease(int releases) { 24 int c = getState() - releases; 25 if (Thread.currentThread() != getExclusiveOwnerThread()) 26 throw new IllegalMonitorStateException(); 27 boolean free = false; 28 if (c == 0) { 29 free = true; 30 setExclusiveOwnerThread(null); 31 } 32 setState(c); 33 return free; 34 } 35 36// AQS 37 private void unparkSuccessor(Node node) { 38 /* 39 * If status is negative (i.e., possibly needing signal) try 40 * to clear in anticipation of signalling. It is OK if this 41 * fails or if status is changed by waiting thread. 42 */ 43 int ws = node.waitStatus; 44 // 如果head节点当前waitStatus<0, CAS 将其修改为0 45 if (ws < 0) 46 compareAndSetWaitStatus(node, ws, 0); 47 48 /* 49 * Thread to unpark is held in successor, which is normally 50 * just the next node. But if cancelled or apparently null, 51 * traverse backwards from tail to find the actual 52 * non-cancelled successor. 53 */ 54 // 唤醒后继节点，但是有可能后继节点取消了等待（waitStatus==1） 55 // 从队尾往前找，找到waitStatus<=0的所有节点中排在最前面的 56 Node s = node.next; 57 if (s == null || s.waitStatus > 0) { 58 s = null; 59 for (Node t = tail; t != null && t != node; t = t.prev) 60 if (t.waitStatus <= 0) 61 s = t; 62 } 63 if (s != null) 64 // 唤醒线程 65 LockSupport.unpark(s.thread); 66 }
by 斯武丶风晴 https://my.oschina.net/langxSpirit