AQS

1
2
3
public abstract class AbstractQueuedSynchronizer
    extends AbstractOwnableSynchronizer
    implements java.io.Serializable ...

AQS即AbstractQueuedSynchronizer（抽象队列同步器），是其他同步器的一个抽象类
它继承了一个AOS（AbstractOwnableSynchronizer），AOS仅仅用于保存占有锁的线程

部分重要属性

AbstractQueuedSynchronizer部分属性及内部类如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
//同步队列的队列头
private transient volatile Node head;
//同步队列的队列尾
private transient volatile Node tail;
//状态值，代表的含义与具体的实现类相关
private volatile int state;
//与Lock的Condition相关，可以实现精确唤醒，每一个ConditionObject对象对应一条 条件队列
public class ConditionObject{
	//条件队列的队列头
	private transient Node firstWaiter;
	//条件队列的队列尾
	private transient Node lastWaiter;
}
//队列的每一个节点,对应一个线程
static final class Node {
	// 分别表示共享模式和排他模式
	static final Node SHARED = new Node();
    static final Node EXCLUSIVE = null;
    // 表示当前节点由于超时或者中断，任务已被取消
    static final int CANCELLED =  1;
    // 表示后继节点线程待唤醒，如果当前节点释放同步状态或者被取消，将有可能会唤醒后继节点
    static final int SIGNAL    = -1;
    // 表示当前节点为条件节点,
    static final int CONDITION = -2;
    // 表示releaseShared时应该传播到其他节点
    static final int PROPAGATE = -3;
    // 等待状态,初始值为0,也可取以上的值
    volatile int waitStatus;
    // 分别表示前一个,后一个节点
    volatile Node prev;
    volatile Node next;
    //Node节点对应的线程对象
    volatile Thread thread;
    // 指向下一个对应特殊Condition对象的节点或者SHARED的节点
    Node nextWaiter;
}

AQS可重写的方法

AQS类定义了几个可重写的方法，待子类重写。可以看到，AQS提供了独占（排他）、非独占（共享）两种模式，只需重写其中需要使用的即可。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
//分别以排他、共享模式尝试获取锁对象
protected boolean tryAcquire(int arg) {
    throw new UnsupportedOperationException();
}
protected int tryAcquireShared(int arg) {
    throw new UnsupportedOperationException();
}
//分别以排他、共享模式尝试释放锁对象
protected boolean tryRelease(int arg) {
    throw new UnsupportedOperationException();
}
protected boolean tryReleaseShared(int arg) {
    throw new UnsupportedOperationException();
}
//返回是否被线程独占
protected boolean isHeldExclusively() {
    throw new UnsupportedOperationException();
}

AQS中几个重要的方法

排他模式占用锁acquire

从acquire入手

1
2
3
4
5
6
// 尝试 独占 锁对象
   public final void acquire(int arg) {
       if (!tryAcquire(arg) &&
           acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
           selfInterrupt();
   }

tryAcquire(arg)交由子类重写,暂不分析
现在分别分析addWaiter(Node mode)和acquireQueued(final Node node, int arg)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
// 添加进等待队列并返回
   private Node addWaiter(Node mode) {
       Node node = new Node(Thread.currentThread(), mode);
       // 如果已经存在等待队列,则直接添加,否则需要先构建等待队列
       Node pred = tail;
       if (pred != null) {
           node.prev = pred;
           // 如果CAS失败，将会在enq方法中，自旋加入队尾
           if (compareAndSetTail(pred, node)) {
               pred.next = node;
               return node;
           }
       }
       enq(node);
       return node;
   }
   //enq方法主要做两件事，①、初始化队列 ②、自旋，保证队尾节点的不丢不重有序添加
   // 初始化等待队列,在头结点处会形成一个空节点,而后向队列中加入node节点
   private Node enq(final Node node) {
       for (;;) {
           Node t = tail;
           if (t == null) { // Must initialize
               if (compareAndSetHead(new Node()))
                   tail = head;
           } else {
               node.prev = t;
               if (compareAndSetTail(t, node)) {
                   t.next = node;
                   return t;
               }
           }
       }
   }
   // 加入等待队列后,如果条件满足,可以再尝试一遍获取锁对象,还是获取不到就进行阻塞,等待唤醒
   final boolean acquireQueued(final Node node, int arg) {
       boolean failed = true;
       try {
           boolean interrupted = false;
           for (;;) {
               final Node p = node.predecessor();
               // 如果是头结点的后继结点,则可以再次尝试获取一次锁对象（其实应该会是两次，解释见下）
               if (p == head && tryAcquire(arg)) {
                   setHead(node);
                   p.next = null; // help GC
                   failed = false;
                   return interrupted;
               }
               //判断是否应该阻塞,如果应该,则阻塞
               if (shouldParkAfterFailedAcquire(p, node) &&
                   parkAndCheckInterrupt())
                   interrupted = true;
           }
       } finally {
           if (failed)
               cancelAcquire(node);
       }
   }

接下来,分析shouldParkAfterFailedAcquire(Node pred, Node node)、parkAndCheckInterrupt()与cancelAcquire(Node node)
对于shouldParkAfterFailedAcquire(Node pred, Node node)方法,每个节点第一次调用时,前置节点的waitStatus为0。所以，第一次会返回false，acquireQueued(final Node node, int arg)方法中的if (p == head && tryAcquire(arg))在阻塞前会运行两次。

此外，if (p == head && tryAcquire(arg)) 保证了FIFO，避免了过早通知（节点在中断时，可能会唤醒后继节点）。
对于，取消节点的过程，下文有讲解

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
// 返回当前线程是否应该阻塞(由前置节点的waitStatus决定),可以查看本文开头关于Node属性的解释
private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
       int ws = pred.waitStatus;
       // 如果前置节点的waitStatus为SIGNAL(-1),则返回true
       if (ws == Node.SIGNAL)
           return true;
       // 如果>0(其实就是1,代表被取消),则跳过被取消的节点
       if (ws > 0) {
           do {
               node.prev = pred = pred.prev;
           } while (pred.waitStatus > 0);
           pred.next = node;
       } else {
           compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
       }
       return false;
   }
   // 阻塞线程,返回是否中断
   private final boolean parkAndCheckInterrupt() {
       LockSupport.park(this);
       return Thread.interrupted();
   }
   // 取消节点任务
   private void cancelAcquire(Node node) {
       if (node == null)
           return;
       // 被取消的节点中的thread置空
       node.thread = null;
       Node pred = node.prev;
       // 向前找到一个非取消的节点记为pred，修改node的前驱指针指向pred
       while (pred.waitStatus > 0)
           node.prev = pred = pred.prev;
       Node predNext = pred.next;
       // 改状态
       node.waitStatus = Node.CANCELLED;
       // 如果node为尾节点，则将pred设置为尾节点
       if (node == tail && compareAndSetTail(node, pred)) {
           compareAndSetNext(pred, predNext, null);
       } else {
           int ws;
           /*
	     * ① pred != head  前置节点不是头节点
	     * ② ((ws = pred.waitStatus) == Node.SIGNAL ||
                (ws <= 0 && compareAndSetWaitStatus(pred, ws, Node.SIGNAL)))  前置节点没有被取消，并且成功设置为SIGNAL
            * ③ pred.thread != null		前置节点中的thread不为空。
            * 									目前，笔者已知：设置头节点的时候，会将头节点的thread置空；
            * 									取消节点的第一步也会将节点中线程置空。
            * 同时满足以上三个条件，则可以进入下一步，跳过所有连续的被取消的节点，否则会唤醒后续节点
	     */
           if (pred != head &&
               ((ws = pred.waitStatus) == Node.SIGNAL ||
                (ws <= 0 && compareAndSetWaitStatus(pred, ws, Node.SIGNAL))) &&
               pred.thread != null) {
               Node next = node.next;
               // 如果node的后置节点不为空，且未被取消，则将pred的后置节点设置为next节点。
               if (next != null && next.waitStatus <= 0)
               	// 此处就跳过了pred与next中间所有的节点（被取消了）
                   compareAndSetNext(pred, predNext, next);
           } else {
           	// 唤醒后继节点，此处可能会产生过早通知（前置节点非头节点，却被唤醒）
               unparkSuccessor(node);
           }
           node.next = node; // help GC
       }
   }

大致流程如图：
请添加图片描述

排他模式释放锁release

现在，分析释放锁的过程
从release入手

1
2
3
4
5
6
7
8
9
10
public final boolean release(int arg) {
    if (tryRelease(arg)) {
        Node h = head;
        if (h != null && h.waitStatus != 0)
            unparkSuccessor(h);
        // 头结点为空或者头结点的waitStatus为0,则跳过unparkSuccessor(h)方法
        return true;
    }
    return false;
}

tryRelease(arg)由子类重写，暂不分析
我们进一步分析unparkSuccessor(Node node)方法

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
private void unparkSuccessor(Node node) {
    int ws = node.waitStatus;
    // 如果当前节点没有被取消，则将waitStatus设置为0
    if (ws < 0)
        compareAndSetWaitStatus(node, ws, 0);
    Node s = node.next;
    // 如果后继节点等于空或者被取消了，则从队尾开始寻找一个未被取消的非node节点的节点
    if (s == null || s.waitStatus > 0) {
        s = null;
        for (Node t = tail; t != null && t != node; t = t.prev)
            if (t.waitStatus <= 0)
                s = t;
    }
    if (s != null)
        LockSupport.unpark(s.thread);
}

所以，release方法释放锁后，如果有下一个节点并且下一个节点没有被取消，则唤醒下一个节点所对应的线程
如果有下一个节点，但是下一个节点被取消了，则从队尾开始找一个没有被取消的节点，唤醒对应的线程
大致流程如图：
请添加图片描述

共享模式占有锁

同样从acquireShared共享模式占有锁方法开始。其中tryAcquireShared(arg)方法是模板方法，交由子类重写。
对于tryAcquireShared(arg)方法
当其返回正数时，表示获取锁成功，且后续共享模式获取锁可能成功；
返回0时，表示获取锁成功，但后续共享模式获取锁不能成功；
返回负数，则表示获取锁失败

1
2
3
4
public final void acquireShared(int arg) {
       if (tryAcquireShared(arg) < 0)
           doAcquireShared(arg);
   }

进一步分析doAcquireShared(int arg)方法

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
private void doAcquireShared(int arg) {
        final Node node = addWaiter(Node.SHARED);
        boolean failed = true;
        try {
            boolean interrupted = false;
            for (;;) {
                final Node p = node.predecessor();
                if (p == head) {
                    int r = tryAcquireShared(arg);
                    if (r >= 0) {
                        setHeadAndPropagate(node, r);
                        p.next = null; // help GC
                        if (interrupted)
                            selfInterrupt();
                        failed = false;
                        return;
                    }
                }
                if (shouldParkAfterFailedAcquire(p, node) &&
                    parkAndCheckInterrupt())
                    interrupted = true;
            }
        } finally {
            if (failed)
                cancelAcquire(node);
        }
    }

对比独占模式获取锁。

1
2
3
4
5
6
// 尝试 独占 锁对象
   public final void acquire(int arg) {
       if (!tryAcquire(arg) &&
           acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
           selfInterrupt();
   }

同样是三步：
①尝试获取锁
②加入等待队列
③基本相同的try(自旋)-finally操作
过程基本相同，只分析其中的setHeadAndPropagate(Node node, int propagate)方法

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
private void setHeadAndPropagate(Node node, int propagate) {
        Node h = head; // Record old head for check below
        setHead(node);
        /*
         * 如果满足以下条件其中之一即可
         * ① propagate > 0 其中，propagate为tryAcquireShared(arg)的返回值
         * ② h == null 表示共享地占有锁之前，没有其他的节点
         * ③ h.waitStatus < 0 情况一：h.waitStatus == Node.SIGNAL 即-1，表示后继节点待唤醒
         * 					  情况二：h.waitStatus == PROPAGATE 即-3，表示唤醒可以向后传播
         * ④ (h = head) == null || h.waitStatus < 0) 表示node节点为空，或者其waitStatus < 0
         */
        if (propagate > 0 || h == null || h.waitStatus < 0 ||
            (h = head) == null || h.waitStatus < 0) {
            Node s = node.next;
            // 下一个节点是共享模式下占有锁产生的节点，则唤醒下一个节点
            if (s == null || s.isShared())
                doReleaseShared();
        }
    }

其中，doReleaseShared()是共享模式下释放锁

共享模式释放锁

独占模式下，释放锁没有并发风险；
而共享模式下，释放锁会有并发风险，所以需要自旋+CAS进行控制

1
2
3
4
5
6
7
public final boolean releaseShared(int arg) {
       if (tryReleaseShared(arg)) {
           doReleaseShared();
           return true;
       }
       return false;
   }

进一步分析doReleaseShared()方法

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
private void doReleaseShared() {
		/*
		* ① 头结点的waitStatus == Node.SIGNAL 即-1，并且唤醒后继节点
		* ② 头结点的waitStatus == 0 ，则改为Node.PROPAGATE
		*/
        for (;;) {
            Node h = head;
            if (h != null && h != tail) {
                int ws = h.waitStatus;
                if (ws == Node.SIGNAL) {
                    if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))
                        continue;            // loop to recheck cases
                    // 唤醒后继节点
                    unparkSuccessor(h);
                }
                else if (ws == 0 &&
                         !compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
                    continue;                // loop on failed CAS
            }
            if (h == head)                   // loop if head changed
                break;
        }
    }

后继节点对应的线程被唤醒后，会尝试共享模式获取锁。如果成功，则头指针往后移动，唤醒下一个节点，重复以上步骤。最终，实现后面所有的连续的共享模式节点成功获取锁。

具体的应用有ReentrantLock、ReentrantReadWriteLock、CountDownLatch、CyclicBarrier、Semaphore等等。
学习以上类的时候，我们有两个关注点
①AQS中state字段的含义
②子类对AQS中几个可重写方法的重写

ReentrantLock

ReentrantLock是可重入锁，重写了三个模板方法，实现了公平，非公平两种独占锁的方式。
FairSync和NonfairSync都继承Sync，Sync继承AQS
在这里插入图片描述

可重入的底层

①state字段表示当前线程锁占有的层数（可重入锁可重入的关键）
每进入一层，该值就+1；每出一层，该值就-1
为0时，代表没有线程占用
代码见下文

公平、非公平底层

直接查看关键代码

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
// 公平地尝试获取锁
protected final boolean tryAcquire(int acquires) {
          final Thread current = Thread.currentThread();
          int c = getState();
          if (c == 0) {
          	// 这里会判断是否有等待队列，这里就是公平的体现
              if (!hasQueuedPredecessors() &&
                  compareAndSetState(0, acquires)) {
                  // 此方法是AOS中的（本文开头已简单介绍），用于设置其中的thread对象
                  setExclusiveOwnerThread(current);
                  return true;
              }
          }
          // 可重入的关键
          else if (current == getExclusiveOwnerThread()) {
              int nextc = c + acquires;
              if (nextc < 0)
                  throw new Error("Maximum lock count exceeded");
              setState(nextc);
              return true;
          }
          return false;
      }
      // 非公平地尝试获取锁
      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;
      }

对于其他的应用暂不做考虑了。

由于笔者能力有限，本文中难免会有一些错误或者不准确的地方，恳请读者批评指正。

看到这里，相信你对AQS已经有了较深的理解。如果本文对你有所帮助，不妨点赞支持一下！谢谢你的阅读与支持！

推荐阅读

《The java.util.concurrent Synchronizer Framework》 JUC同步器框架（AQS框架）原文翻译