AQS 锁核心类详解

private volatile int state;//共享变量，使用volatile修饰保证线程可见性

 1 //返回同步状态的当前值
 2 protected final int getState() {  
 3         return state;
 4 }
 5  // 设置同步状态的值
 6 protected final void setState(int newState) { 
 7         state = newState;
 8 }
 9 //原子地(CAS操作)将同步状态值设置为给定值update如果当前同步状态的值等于expect(期望值)
10 protected final boolean compareAndSetState(int expect, int update) {
11         return unsafe.compareAndSwapInt(this, stateOffset, expect, update);
12 }

1 isHeldExclusively()//该线程是否正在独占资源。只有用到condition才需要去实现它。
2 tryAcquire(int)//独占方式。尝试获取资源，成功则返回true，失败则返回false。
3 tryRelease(int)//独占方式。尝试释放资源，成功则返回true，失败则返回false。
4 tryAcquireShared(int)//共享方式。尝试获取资源。负数表示失败；0表示成功，但没有剩余可用资源；正数表示成功，且有剩余资源。
5 tryReleaseShared(int)//共享方式。尝试释放资源，成功则返回true，失败则返回false。

public abstract class AbstractQueuedSynchronizer extends AbstractOwnableSynchronizer implements java.io.Serializable

 1 public abstract class AbstractOwnableSynchronizer implements java.io.Serializable {
 2     
 3     // 版本序列号
 4     private static final long serialVersionUID = 3737899427754241961L;
 5     // 构造方法
 6     protected AbstractOwnableSynchronizer() { }
 7     // 独占模式下的线程
 8     private transient Thread exclusiveOwnerThread;
 9     
10     // 设置独占线程 
11     protected final void setExclusiveOwnerThread(Thread thread) {
12         exclusiveOwnerThread = thread;
13     }
14     
15     // 获取独占线程 
16     protected final Thread getExclusiveOwnerThread() {
17         return exclusiveOwnerThread;
18     }
19 }

 1 static final class Node {
 2     // 模式，分为共享与独占
 3     // 共享模式
 4     static final Node SHARED = new Node();
 5     // 独占模式
 6     static final Node EXCLUSIVE = null;        
 7     // 结点状态
 8     // CANCELLED，值为1，表示当前的线程被取消
 9     // SIGNAL，值为-1，表示当前节点的后继节点包含的线程需要运行，也就是unpark
10     // CONDITION，值为-2，表示当前节点在等待condition，也就是在condition队列中
11     // PROPAGATE，值为-3，表示当前场景下后续的acquireShared能够得以执行
12     // 值为0，表示当前节点在sync队列中，等待着获取锁
13     static final int CANCELLED =  1;
14     static final int SIGNAL    = -1;
15     static final int CONDITION = -2;
16     static final int PROPAGATE = -3;        
17 
18     // 结点状态
19     volatile int waitStatus;        
20     // 前驱结点
21     volatile Node prev;    
22     // 后继结点
23     volatile Node next;        
24     // 结点所对应的线程
25     volatile Thread thread;        
26     // 下一个等待者
27     Node nextWaiter;
28     
29     // 结点是否在共享模式下等待
30     final boolean isShared() {
31         return nextWaiter == SHARED;
32     }
33     
34     // 获取前驱结点，若前驱结点为空，抛出异常
35     final Node predecessor() throws NullPointerException {
36         // 保存前驱结点
37         Node p = prev; 
38         if (p == null) // 前驱结点为空，抛出异常
39             throw new NullPointerException();
40         else // 前驱结点不为空，返回
41             return p;
42     }
43     
44     // 无参构造方法
45     Node() {    // Used to establish initial head or SHARED marker
46     }
47     
48     // 构造方法
49         Node(Thread thread, Node mode) {    // Used by addWaiter
50         this.nextWaiter = mode;
51         this.thread = thread;
52     }
53     
54     // 构造方法
55     Node(Thread thread, int waitStatus) { // Used by Condition
56         this.waitStatus = waitStatus;
57         this.thread = thread;
58     }
59 }

 1 public interface Condition {
 2 
 3     // 等待，当前线程在接到信号或被中断之前一直处于等待状态
 4     void await() throws InterruptedException;
 5     
 6     // 等待，当前线程在接到信号之前一直处于等待状态，不响应中断
 7     void awaitUninterruptibly();
 8     
 9     //等待，当前线程在接到信号、被中断或到达指定等待时间之前一直处于等待状态 
10     long awaitNanos(long nanosTimeout) throws InterruptedException;
11     
12     // 等待，当前线程在接到信号、被中断或到达指定等待时间之前一直处于等待状态。此方法在行为上等效于: awaitNanos(unit.toNanos(time)) > 0
13     boolean await(long time, TimeUnit unit) throws InterruptedException;
14     
15     // 等待，当前线程在接到信号、被中断或到达指定最后期限之前一直处于等待状态
16     boolean awaitUntil(Date deadline) throws InterruptedException;
17     
18     // 唤醒一个等待线程。如果所有的线程都在等待此条件，则选择其中的一个唤醒。在从 await 返回之前，该线程必须重新获取锁。
19     void signal();
20     
21     // 唤醒所有等待线程。如果所有的线程都在等待此条件，则唤醒所有线程。在从 await 返回之前，每个线程都必须重新获取锁。
22     void signalAll();
23 }

 1 public abstract class AbstractQueuedSynchronizer extends AbstractOwnableSynchronizer
 2     implements java.io.Serializable {    
 3     // 版本号
 4     private static final long serialVersionUID = 7373984972572414691L;    
 5     // 头结点
 6     private transient volatile Node head;    
 7     // 尾结点
 8     private transient volatile Node tail;    
 9     // 状态
10     private volatile int state;    
11     // 自旋时间
12     static final long spinForTimeoutThreshold = 1000L;
13     
14     // Unsafe类实例
15     private static final Unsafe unsafe = Unsafe.getUnsafe();
16     // state内存偏移地址
17     private static final long stateOffset;
18     // head内存偏移地址
19     private static final long headOffset;
20     // state内存偏移地址
21     private static final long tailOffset;
22     // tail内存偏移地址
23     private static final long waitStatusOffset;
24     // next内存偏移地址
25     private static final long nextOffset;
26     // 静态初始化块
27     static {
28         try {
29             stateOffset = unsafe.objectFieldOffset
30                 (AbstractQueuedSynchronizer.class.getDeclaredField("state"));
31             headOffset = unsafe.objectFieldOffset
32                 (AbstractQueuedSynchronizer.class.getDeclaredField("head"));
33             tailOffset = unsafe.objectFieldOffset
34                 (AbstractQueuedSynchronizer.class.getDeclaredField("tail"));
35             waitStatusOffset = unsafe.objectFieldOffset
36                 (Node.class.getDeclaredField("waitStatus"));
37             nextOffset = unsafe.objectFieldOffset
38                 (Node.class.getDeclaredField("next"));
39 
40         } catch (Exception ex) { throw new Error(ex); }
41     }
42 }

1 protected AbstractQueuedSynchronizer() { } 

1 public final void acquire(int arg) {
2     if (!tryAcquire(arg) && acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
3         selfInterrupt();
4 }

 1 // 添加等待者
 2 private Node addWaiter(Node mode) {
 3     // 新生成一个结点，默认为独占模式
 4     Node node = new Node(Thread.currentThread(), mode);
 5     // Try the fast path of enq; backup to full enq on failure
 6     // 保存尾结点
 7     Node pred = tail;
 8     if (pred != null) { // 尾结点不为空，即已经被初始化
 9         // 将node结点的prev域连接到尾结点
10         node.prev = pred; 
11         if (compareAndSetTail(pred, node)) { // 比较pred是否为尾结点，是则将尾结点设置为node 
12             // 设置尾结点的next域为node
13             pred.next = node;
14             return node; // 返回新生成的结点
15         }
16     }
17     enq(node); // 尾结点为空(即还没有被初始化过)，或者是compareAndSetTail操作失败，则入队列
18     return node;
19 }

 1 private Node enq(final Node node) {
 2     for (;;) { // 无限循环，确保结点能够成功入队列
 3         // 保存尾结点
 4         Node t = tail;
 5         if (t == null) { // 尾结点为空，即还没被初始化
 6             if (compareAndSetHead(new Node())) // 头结点为空，并设置头结点为新生成的结点
 7                 tail = head; // 头结点与尾结点都指向同一个新生结点
 8         } else { // 尾结点不为空，即已经被初始化过
 9             // 将node结点的prev域连接到尾结点
10             node.prev = t; 
11             if (compareAndSetTail(t, node)) { // 比较结点t是否为尾结点，若是则将尾结点设置为node
12                 // 设置尾结点的next域为node
13                 t.next = node; 
14                 return t; // 返回尾结点
15             }
16         }
17     }
18 }

 1 // sync队列中的结点在独占且忽略中断的模式下获取(资源)
 2 final boolean acquireQueued(final Node node, int arg) {
 3     // 标志
 4     boolean failed = true;
 5     try {
 6         // 中断标志
 7         boolean interrupted = false;
 8         for (;;) { // 无限循环
 9             // 获取node节点的前驱结点
10             final Node p = node.predecessor(); 
11             if (p == head && tryAcquire(arg)) { // 前驱为头结点并且成功获得锁
12                 setHead(node); // 设置头结点
13                 p.next = null; // help GC
14                 failed = false; // 设置标志
15                 return interrupted; 
16             }
17             if (shouldParkAfterFailedAcquire(p, node) &&
18                 parkAndCheckInterrupt())
19                 interrupted = true;
20         }
21     } finally {
22         if (failed)
23             cancelAcquire(node);
24     }
25 }

 1 // 当获取(资源)失败后，检查并且更新结点状态
 2 private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
 3     // 获取前驱结点的状态
 4     int ws = pred.waitStatus;
 5     if (ws == Node.SIGNAL) // 状态为SIGNAL，为-1
 6         /*
 7             * This node has already set status asking a release
 8             * to signal it, so it can safely park.
 9             */
10         // 可以进行park操作
11         return true; 
12     if (ws > 0) { // 表示状态为CANCELLED，为1
13         /*
14             * Predecessor was cancelled. Skip over predecessors and
15             * indicate retry.
16             */
17         do {
18             node.prev = pred = pred.prev;
19         } while (pred.waitStatus > 0); // 找到pred结点前面最近的一个状态不为CANCELLED的结点
20         // 赋值pred结点的next域
21         pred.next = node; 
22     } else { // 为PROPAGATE -3 或者是0 表示无状态,(为CONDITION -2时，表示此节点在condition queue中) 
23         /*
24             * waitStatus must be 0 or PROPAGATE.  Indicate that we
25             * need a signal, but don't park yet.  Caller will need to
26             * retry to make sure it cannot acquire before parking.
27             */
28         // 比较并设置前驱结点的状态为SIGNAL
29         compareAndSetWaitStatus(pred, ws, Node.SIGNAL); 
30     }
31     // 不能进行park操作
32     return false;
33 }

1 // 进行park操作并且返回该线程是否被中断
2 private final boolean parkAndCheckInterrupt() {
3     // 在许可可用之前禁用当前线程，并且设置了blocker
4     LockSupport.park(this);
5     return Thread.interrupted(); // 当前线程是否已被中断，并清除中断标记位
6 }

 1 // 释放后继结点
 2 private void unparkSuccessor(Node node) {
 3     /*
 4         * If status is negative (i.e., possibly needing signal) try
 5         * to clear in anticipation of signalling.  It is OK if this
 6         * fails or if status is changed by waiting thread.
 7         */
 8     // 获取node结点的等待状态
 9     int ws = node.waitStatus;
10     if (ws < 0) // 状态值小于0，为SIGNAL -1 或 CONDITION -2 或 PROPAGATE -3
11         // 比较并且设置结点等待状态，设置为0
12         compareAndSetWaitStatus(node, ws, 0);
13 
14     /*
15         * Thread to unpark is held in successor, which is normally
16         * just the next node.  But if cancelled or apparently null,
17         * traverse backwards from tail to find the actual
18         * non-cancelled successor.
19         */
20     // 获取node节点的下一个结点
21     Node s = node.next;
22     if (s == null || s.waitStatus > 0) { // 下一个结点为空或者下一个节点的等待状态大于0，即为CANCELLED
23         // s赋值为空
24         s = null; 
25         // 从尾结点开始从后往前开始遍历
26         for (Node t = tail; t != null && t != node; t = t.prev)
27             if (t.waitStatus <= 0) // 找到等待状态小于等于0的结点，找到最前的状态小于等于0的结点
28                 // 保存结点
29                 s = t;
30     }
31     if (s != null) // 该结点不为为空，释放许可
32         LockSupport.unpark(s.thread);
33 }

 1 public final boolean release(int arg) {
 2     if (tryRelease(arg)) { // 释放成功
 3         // 保存头结点
 4         Node h = head; 
 5         if (h != null && h.waitStatus != 0) // 头结点不为空并且头结点状态不为0
 6             unparkSuccessor(h); //释放头结点的后继结点
 7         return true;
 8     }
 9     return false;
10 }

 1 // java.util.concurrent.locks.ReentrantLock
 2 
 3 static final class NonfairSync extends Sync {
 4     ...
 5     final void lock() {
 6         if (compareAndSetState(0, 1))
 7             setExclusiveOwnerThread(Thread.currentThread());
 8         else
 9             acquire(1);
10     }
11   ...
12 }

1 // java.util.concurrent.locks.AbstractQueuedSynchronizer
2 
3 public final void acquire(int arg) {
4     if (!tryAcquire(arg) && acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
5         selfInterrupt();
6 }

1 // java.util.concurrent.locks.AbstractQueuedSynchronizer
2 
3 protected boolean tryAcquire(int arg) {
4     throw new UnsupportedOperationException();
5 }