Java并发编程之JUC并发核心AQS同步队列原理剖析

Java并发编程之JUC并发核心AQS同步队列原理剖析

目录一、AQS介绍二、AQS中的队列1、同步等待队列2、条件等待队列3、AQS队列节点Node三、同步队列源码分析1、同步队列分析2、同步队列——独占模式源码分析3、同步队列——共享模式源码分析

一、AQS介绍

AQS具备特性

阻塞等待队列

共享/独占

公平/非公平

可重入

允许中断

它维护了一个volatile int state（代表共享资源）和一个FIFO线程等待队列（多线程争用资源被阻塞时会进入此队列）。state的访问方式有三种:

getState() 获取state

setState() 设置state

compareAndSetState() 通过CAS的方式设置state值

AQS有两种资源共享方式：Exclusive（独占式） 和 Share（共享式）。所谓独占式是指依据AQS中的state控制状态，只有一个线程能够进行工作（其它参与调度的线程会进入阻塞状态，如ReentrantLock）；共享式是指，依据AQS中的state控制状态，可以有多个满足条件的线程同时执行（如Semaphore/CountDownLatch）。

AQS定义两种队列

同步等待队列（基于双向链表实现）

条件等待队列（基于单向链表实现）

不同的自定义同步器争用共享资源的方式也不同。自定义同步器在实现时只需要实现共享资源state的获取与释放方式即可，至于具体线程等待队列的维护（如获取资源失败入队/唤醒出队等），AQS已经在顶层实现好了。一般通过定义内部类Sync继承AQS将同步器所有调用都映射到Sync对应的方法。

自定义同步器实现时主要实现以下几种方法：

isHeldExclusively()：该线程是否正在独占资源，如果返回true，则表示当前线程正在独占资源。只有用到condition才需要去实现它。

tryAcquire(int)：独占方式。尝试获取资源，成功则返回true，失败则返回false。

tryRelease(int)：独占方式。尝试释放资源，成功则返回true，失败则返回false。

tryAcquireShared(int)：共享方式。尝试获取资源。负数表示失败；0表示成功，但没有剩余可用资源；正数表示成功，且有剩余资源。

tryReleaseShared(int)：共享方式。尝试释放资源，如果释放后允许唤醒后续等待结点返回true，否则返回false。

实现自定义同步组件时，将会调用同步器提供的模板方法，这些（部分）模板方法与描述如下。同步器提供的模板方法基本上分为3类：独占式获取与释放同步状态、共享式获取与释放同步状态和查询同步队列中的等待线程情况。自定义同步组件将使用同步器提供的模板方法来实现自己的同步语义。

acquire(int arg)：独占式获取同步状态，如果当前线程获取同步状态成功，则由该方法返回，否则，将会进入同步队列等待，该方法将会调用可重写的tryAcquire(int arg)方法；

acquireInterruptibly(int arg)：与acquire(int arg)相同，但是该方法响应中断，当前线程为获取到同步状态而进入到同步队列中，如果当前线程被中断，则该方法会抛出InterruptedException异常并返回；

tryAcquireNanos(int arg,long nanos)：超时获取同步状态，如果当前线程在nanos时间内没有获取到同步状态，那么将会返回false，已经获取则返回true；

acquireShared(int arg)：共享式获取同步状态，如果当前线程未获取到同步状态，将会进入同步队列等待，与独占式的主要区别是在同一时刻可以有多个线程获取到同步状态；

acquireSharedInterruptibly(int arg)：共享式获取同步状态，响应中断；

tryAcquireSharedNanos(int arg, long nanosTimeout)：共享式获取同步状态，增加超时限制；

release(int arg)：独占式释放同步状态，该方法会在释放同步状态之后，将同步队列中第一个节点包含的线程唤醒；

releaseShared(int arg)：共享式释放同步状态；

二、AQS中的队列

1、同步等待队列

AQS当中的同步等待队列也称CLH队列，CLH队列是Craig、Landin、Hagersten三人发明的一种基于双向链表数据结构的队列，是FIFO先入先出线程等待队列，Java中的CLH队列是原CLH队列的一个变种，线程由原自旋机制改为阻塞机制。

这种结构的特点是每个数据结构都有两个指针，分别指向直接的后继节点和直接前驱节点。所以双向链表可以从任意一个节点开始很方便的访问前驱和后继。每个 Node 其实是由线程封装，当线程争抢锁失败后会封装成 Node 加入到 ASQ 队列中去；当获取锁的线程释放锁以后，会从队列中唤醒一个阻塞的节点线程 。

2、条件等待队列

条件等待队列是单向链表实现的，此时Node（下面会介绍）中pre和next都为null。Condition是一个多线程间协调通信的工具类，使得某个、或者某些线程一起等待某个条件（Condition），只有当该条件具备时，这些等待线程才会被唤醒，从而重新争夺锁。

3、AQS队列节点Node

同步队列中的节点（Node）用来保存获取同步状态失败的线程引用、等待状态以及前驱和后继节点，节点的属性类型与名称等，Node类基本属性定义如下所示，它是在AbstractQueuedSynchronizer中的一个内部类。

注意：如果Node在条件队列当中，Node必须是独占模式，不能是共享模式。

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;

static final int PROPAGATE = -3;

volatile int waitStatus;

volatile Node prev;

volatile Node next;

volatile Thread thread;

Node nextWaiter;

}

Node pre：前驱节点，当前节点加入到同步队列中被设置（尾部添加）

Node next：后继节点

Thread thread：节点同步状态的线程

Node nextWaiter：等待队列中的后继节点，如果当前节点是共享的，那么这个字段是一个SHARED常量，也就是说节点类型(独占和共享)和等待队列中的后继节点共用同一个字段

int waitStatus：等待状态，标记当前节点的信号量状态 (1,0,-1,-2,-3)5种状态，使用CAS更改状态，volatile保证线程可见性，高并发场景下，即被一个线程修改后，状态会立马让其他线程可见，五种状态分别为：

CANCELLED，值为1，在同步队列中等待的线程等待超时或者被中断，需要从同步队列中取消等待，节点进入该状态后将不会变化

SIGNAL，值为-1，后继节点的线程处于等待状态，而当前的节点如果释放了同步状态或者被取消，将会通知后继节点，使后继节点的线程得以运行。

CONDITION，值为-2，节点在等待队列中，节点的线程等待在Condition上，当其他线程对Condition调用了signal()方法后，该节点会从等待队列中转移到同步队列中，加入到同步状态的获取中

PROPAGATE ，值为-3，表示下一次共享式同步状态获取将会被无条件地传播下去

INITIAL，值为0，初始状态

三、同步队列源码分析

1、同步队列分析

同步器拥有首节点（head）和尾节点（tail），没有成功获取同步状态的线程将会成为节点加入该队列的尾部，同步队列的基本结构如图所示。

同步器包含了两个节点类型的引用，一个指向头节点，而另一个指向尾节点。同步器提供了一个基于CAS的设置尾节点的方法：compareAndSetTail(Node expect,Node update)，它需要传递当前线程“认为”的尾节点和当前节点，只有设置成功后，当前节点才正式与之前的尾节点建立关联。 涉及两个变化：

1. 新的线程封装成 Node 节点追加到同步队列中，设置 prev 节点以及修改当前节点的前置节点的 next 节点指向自己

2. 通过 CAS 讲 tail 重新指向新的尾部节点

head节点表示获取锁成功的节点，当头结点在释放同步状态时，会唤醒后继节点，如果后继节点获得锁成功，会把自己设置为头结点，节点的变化过程如下

同步队列遵循FIFO，首节点是获取同步状态成功的节点，首节点的线程在释放同步状态时，将会唤醒后继节点，而后继节点将会在获取同步状态成功时将自己设置为首节点，该过程如下图所示。涉及两个变化：

1. 修改 head 节点指向下一个获得锁的节点

2. 新的获得锁的节点，将 prev 的指针指向 null

设置首节点是通过获取同步状态成功的线程来完成的，由于只有一个线程能够成功获取到同步状态，因此设置头节点的方法并不需要使用CAS来保证，它只需要将首节

点设置成为原首节点的后继节点并断开原首节点的next引用即可。

2、同步队列——独占模式源码分析

acquire方法（独占获取）源码分析

通过调用同步器的acquire(int arg)方法可以获取同步状态，该方法对中断不敏感，也就是由于线程获取同步状态失败后进入同步队列中，后续对线程进行中断操作时，线程不会从同步队列中移出。

public final void acquire(int arg) {

if (!tryAcquire(arg) &&

acquireQueued(addWaiter(Node.EXCLUSIVE), arg))

selfInterrupt();

}

上面方法中首先调用自定义同步器实现的tryAcquire(int arg)方法（重写该方法），该方法保证线程安全的获取同步状态，如果同步状态获取失败，则构造同步节点（独占式Node.EXCLUSIVE，同一时刻只能有一个线程成功获取同步状态）并通过addWaiter(Node node)方法将该节点加入到同步队列的尾部，最后调用acquireQueued(Node node,int arg)方法，使得该节点以“死循环”的方式获取同步状态。如果获取不到则阻塞节点中的线程，而被阻塞线程的唤醒主要依靠前驱节点的出队或阻塞线程被中断来实现。

下面看下addWaiter方法的实现：把当前线程构建为Node节点；判断尾结点是否为空，通过CAS的方式将当前节点放到队列尾部；如果尾结点不为空或者前面CAS插入尾结点失败，调用enq方法，通过自旋的方式插入尾结点。

private Node addWaiter(Node mode) {

// 1. 将当前线程构建成Node类型

Node node = new Node(Thread.currentThread(), mode);

// Try the fast path of enq; backup to full enq on failure

Node pred = tail;

// 2. 1当前尾节点是否为null？

if (pred != null) {

// 2.2 将当前节点尾插入的方式

node.prev = pred;

// 2.3 CAS将节点插入同步队列的尾部

if (compareAndSetTail(pred, node)) {

pred.next = node;

return node;

}

}

enq(node);

return node;

}

下面看下enq方法的实现：判断尾结点是否为空，如果为空，则通过CAS的方式创建一个空的头结点（Thread为空），并将尾结点也指向头结点；如果尾结点不为空或者上面CAS创建头结点失败，将当前队列的前驱指针指向原来的尾结点，通过CAS的方式将当前节点放到队列尾部，将原来尾结点的后继指针指向当前节点；如果前面都失败了，进行下一次循环。当前线程构造的node节点通过addWaiter方法执行入队之后，其waitStatus为0，头结点的waitStatus也是0，此时是下面这种结构

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;

//set尾部节点

if (compareAndSetTail(t, node)) {//当前节点置为尾部

t.next = node; //前驱节点的next指针指向当前节点

return t;

}

}

}

}

通过addWaiter 方法把线程添加到链表后， 会接着把 Node 作为参数传递给acquireQueued 方法，去竞争锁

1. 获取当前节点的 prev 节点

2. 如果 prev 节点为 head 节点，那么它就有资格去争抢锁，调用 tryAcquire 抢占锁

3. 抢占锁成功以后，把获得锁的节点设置为 head ，并且移除原来的初始head节点，通过setHead方法和p.next=null来将新的head（获取锁的线程节点）与原head断开连接，并将新的head的thread设为null。由此可见head节点的waitStatus都为0

4. 如果当前节点的前驱不是head或者当前节点是head但是获取锁失败，则根据前驱节点的waitStatus（SIGNAL）决定是否需要挂起线程

/**

* 已经在队列当中的Thread节点，准备阻塞等待获取锁

*/

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)) {//如果前驱结点是头结点，才tryAcquire，其他结点是没有机会tryAcquire的。

setHead(node);//获取同步状态成功，将当前结点设置为头结点。

p.next = null; // help GC

failed = false;

return interrupted;

}

/**

* 如果前驱节点不是Head，通过shouldParkAfterFailedAcquire判断是否应该阻塞

* 前驱节点信号量为-1，当前线程可以安全被parkAndCheckInterrupt用来阻塞线程

*/

if (shouldParkAfterFailedAcquire(p, node) &&

parkAndCheckInterrupt())

interrupted = true;

}

} finally {

if (failed)

cancelAcquire(node);

}

}

下面是setHead方法的实现

private void setHead(Node node) {

head = node;

node.thread = null;

node.prev = null;

}

在acquireQueued(final Node node,int arg)方法中，当前线程在“死循环”中尝试获取同步状态，而只有前驱节点是头节点才能够尝试获取同步状态，这是为什么？原因有两个，如下。

第一，头节点是成功获取到同步状态的节点，而头节点的线程释放了同步状态之后，将会唤醒其后继节点，后继节点的线程被唤醒后需要检查自己的前驱节点是否是头节点。

第二，维护同步队列的FIFO原则。该方法中，节点自旋获取同步状态的行为如图所示

如果前驱节点不是Head，通过shouldParkAfterFailedAcquire判断是否应该阻塞：如果前驱节点waitStatus为-1（SIGNAL的状态），当前线程可以安全的被parkAndCheckInterrupt用来阻塞线程；通过循环扫描链表把 CANCELLED 状态的节点移除；如果前驱节点waitStatus不是-1，则通过CAS将前驱节点的waitStatus改为-1。

第一次循环进入shouldParkAfterFailedAcquire方法时head节点为0，会将其改为SIGNAL，此时会返回false，那么外层的方法acquireQueued方法会执行第二次循环进入shouldParkAfterFailedAcquire方法，此时会返回true，当前线程可以被阻塞，则调用parkAndCheckInterrupt（）方法阻塞当前线程，其底层调用的是UnSafe类里面的park方法。

shouldParkAfterFailedAcquire方法会将前驱节点的waitStatus改为SIGNAL，因为只有前驱节点的状态是SIGNAL后继节点才可以被阻塞，次数除了tail节点的状态是0，其他的都是-1。

private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {

int ws = pred.waitStatus;

if (ws == Node.SIGNAL)

/*

* 若前驱结点的状态是SIGNAL，意味着当前结点可以被安全地park

*/

return true;

if (ws > 0) {

/*

* 前驱节点状态如果被取消状态，将被移除出队列

*/

do {

node.prev = pred = pred.prev;

} while (pred.waitStatus > 0);

pred.next = node;

} else {

/*

* 当前驱节点waitStatus为 0 or PROPAGATE状态时

* 将其设置为SIGNAL状态，然后当前结点才可以可以被安全地park

*/

compareAndSetWaitStatus(pred, ws, Node.SIGNAL);

}

return false;

}

通过分析acquireQueued方法可以得出结论：头结点是获取同步状态成功的节点，头结点的所有有效后继节点线程都会被阻塞，释放锁后需要挨个唤醒头结点的后续线程节点。

独占式同步状态获取流程，也就是acquire(int arg)方法调用流程，如图所示。

release方法（独占释放）源码分析

当前线程获取同步状态并执行了相应逻辑之后，就需要释放同步状态，使得后续节点能够继续获取同步状态。通过调用同步器的release(int arg)方法可以释放同步状态，该方法在tryRelease方法释放了同步状态之后，会唤醒其后继节点（进而使后继节点重新尝试获取同步状态）。

public final boolean release(int arg) {

if (tryRelease(arg)) {//释放一次锁

Node h = head;

if (h != null && h.waitStatus != 0)

unparkSuccessor(h);//唤醒后继结点

return true;

}

return false;

}

该方法执行时，会唤醒头节点的后继节点线程，unparkSuccessor(Node node)方法底层使用UnSafe的unpark方法来唤醒处于等待状态的线程。

private void unparkSuccessor(Node node) {

//获取wait状态

int ws = node.waitStatus;

if (ws < 0)

compareAndSetWaitStatus(node, ws, 0);// 将等待状态waitStatus设置为初始值0

/**

* 若后继结点为空，或状态为CANCEL（已失效），则从后尾部往前遍历找到最前的一个处于正常阻塞状态的结点

* 进行唤醒

*/

Node s = node.next;

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);//唤醒线程

}

上面方法中判断node的后继节点是空或者waitStatus是撤销状态，会从tail往前遍历找到一个离node节点最近的节点，这是为什么呢？ 原因在于上面acquire时调用的enq入队方法：先compareAndSetTail(t, node)设置尾结点，然后t.next=node将前驱节点的next指针指向当前节点，如果t.next=node还没有执行的话，链表还没有建立完整，从前向后遍历时会出现遍历到t时找不到t的后继节点，从后往前遍历则不会出现这种情况。

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;

//set尾部节点

if (compareAndSetTail(t, node)) {//当前节点置为尾部

t.next = node; //前驱节点的next指针指向当前节点

return t;

}

}

}

}

分析了独占式同步状态获取和释放过程后，适当做个总结：在获取同步状态时，同步器维护一个同步队列，获取状态失败的线程都会被加入到队列中（都是头结点的后继节点）并在队列中进行自旋；移出队列（或停止自旋）的条件是前驱节点为头节点且成功获取了同步状态。在释放同步状态时，同步器调用tryRelease(int arg)方法释放同步状态，然后唤醒头节点的后继节点。

3、同步队列——共享模式源码分析

同步队列共享模式与独占模式异同点：

共享模式跟独占模式的相同点：获取同步状态失败的线程会被包装成node节点添加到队列尾部

共享模式跟独占模式的不同点：独占模式中同步状态是独占的，只有一个线程可以获取到同步资源，因此释放同步资源时会唤醒head节点后面的一个节点；而共享模式因为多个线程可以共享同步资源，所以唤醒线程时会唤醒head节点后面的所有有效节点。

acquireShared方法（共享获取）源码分析

acquireShared方法会先调用tryAcquireShared获取同步状态，如果返回值小于0表示获取失败，需要进行排队；如果获取成功，则可以向下执行。

public final void acquireShared(int arg) {

if (tryAcquireShared(arg) < 0)//返回值小于0，获取同步状态失败，排队去；获取同步状态成功，直接返回去干自己的事儿。

doAcquireShared(arg);

}

获取失败时会调用doAcquireShared方法进行排队：

先通过addWaiter方法入队，上面已经介绍了，这里就不再重复分析了，唯一不同的就是添加的是一个共享模式的node

判断当前节点的前驱节点是否是head，如果是的话再次调用tryAcquireShared获取同步资源，如果获取成功，则将当前node设置为head并且唤醒等待的线程节点

如果当前节点的前驱节点不是head或者是head但是获取同步资源失败，则跟上面的共享模式一样调用shouldParkAfterFailedAcquire方法将node的前驱节点设置为SIGNAL（-1）状态，然后阻塞当前线程。

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); //非公平锁实现，再尝试获取锁

//state==0时tryAcquireShared会返回>=0(CountDownLatch中返回的是1)。

// state为0说明共享次数已经到了，可以获取锁了

if (r >= 0) {//r>0表示state==0,前继节点已经释放锁，锁的状态为可被获取

//这一步设置node为head节点设置node.waitStatus->Node.PROPAGATE，然后唤醒node.thread

setHeadAndPropagate(node, r);

p.next = null; // help GC

if (interrupted)

selfInterrupt();

failed = false;

return;

http://}

}

//前继节点非head节点，将前继节点状态设置为SIGNAL，通过park挂起node节点的线程

if (shouldParkAfterFailedAcquire(p, node) &&

parkAndCheckInterrupt())

interrupted = true;

}

} finally {

if (failed)

cancelAcquire(node);

}

}

下面看下setHeadAndPropagate方法：

调用setHead方法将当前节点设置head

判断如果需要执行唤醒，通过上面的分析这里会调用doReleaseShared执行唤醒

/**

* 把node节点设置成head节点，且Node.waitStatus->Node.PROPAGATE

*/

private void setHeadAndPropagate(Node node, int propagate) {

Node h = head; //h用来保存旧的head节点

setHead(node);//head引用指向node节点

/* 这里意思有两种情况是需要执行唤醒操作

* 1.propagate > 0 表示调用方指明了后继节点需要被唤醒

* 2.头节点后面的节点需要被唤醒（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())//node是最后一个节点或者 node的后继节点是共享节点

/* 如果head节点状态为SIGNAL，唤醒head节点线程，重置head.waitStatus->0

* head节点状态为0(第一次添加时是0)，设置head.waitStatus->Node.PROPAGATE表示状态需要向后继节点传播

*/

doReleaseShared();

}

}

看下doReleaseShared方法的实现：

判断head!=null && head!=tail，然后判断head的waitStatus如果是SIGNAL则会使用CAS的方式将其改为0，这里没有直接改成PROPAGATE，是因为unparkSuccessor(h)中，如果ws < 0会设置为0，所以ws先设置为0，再设置为PROPAGATE，这里需要控制并发，因为入口有setHeadAndPropagate跟release两个，避免两次unpark。head状态为SIGNAL，且成功设置为0之后唤醒head.next节点线程，此时head、head.next的线程都唤醒了，head.next会去竞争锁，成功后head会指向获取锁的节点，也就是head发生了变化。head发生变化后h==head会不成立了，此时会重新循环，继续唤醒重新获取的新head的下一个节点。

如果本身头节点的waitStatus是0，将其设置为PROPAGATE状态。意味着需要将状态向后一个节点传播。

最后判断如果h==head，即已经没有要唤醒的节点了，跳出循环向下执行。如果h!=head，则说明head指针发生了变化，head已经指向了新唤醒的线程node，继续执行下次循环，获取新的head，唤醒head的后续节点。

private void doReleaseShared() {

for (;;) {

Node h = head;

if (h != null && h != tail) {

int ws = h.waitStatus;

if (ws == Node.SIGNAL) {//head是SIGNAL状态

/* head状态是SIGNAL，重置head节点waitStatus为0，这里不直接设为Node.PROPAGATE,

* 是因为unparkSuccessor(h)中，如果ws < 0会设置为0，所以ws先设置为0，再设置为PROPAGATE

* 这里需要控制并发，因为入口有setHeadAndPropagate跟release两个，避免两次unpark

*/

if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))

continue; //设置失败，重新循环

/* head状态为SIGNAL，且成功设置为0之后,唤醒head.next节点线程

* 此时head、head.next的线程都唤醒了，head.next会去竞争锁，成功后head会指向获取锁的节点，

* 也就是head发生了变化。看最底下一行代码可知，head发生变化后会重新循环，继续唤醒head的下一个节点

*/

unparkSuccessor(h);

/*

* 如果本身头节点的waitStatus是出于重置状态（waitStatus==0）的，将其设置为“传播”状态。

* 意味着需要将状态向后一个节点传播

*/

}

else if (ws == 0 &&

!compareAndSeqDefgbwtWaitStatus(h, 0, Node.PROPAGATE))

continue; // loop on failed CAS

}

if (h == head) //如果head变了，重新循环

break;

}

}

releaseShared方法（共享释放）源码分析

调用tryReleaseShared方法释放资源成功时会调用doReleaseShared方法执行唤醒逻辑，上面已经分析过了。

public final boolean releaseShared(int arg) {

if (tryReleaseShared(arg)) {

doReleaseShared();

return true;

}

return false;

}

版权声明：本文内容由网络用户投稿，版权归原作者所有，本站不拥有其著作权，亦不承担相应法律责任。如果您发现本站中有涉嫌抄袭或描述失实的内容，请联系我们jiasou666@gmail.com 处理，核实后本网站将在24小时内删除侵权内容。