Java并发编程札记-(四)JUC锁-03AQS

AQS，AbstractQueuedSynchronizer的缩写，是JUC中非常重要的一个类。javadoc中对其的介绍是：

为实现依赖于先进先出 (FIFO) 等待队列的阻塞锁和相关同步器（信号量、事件，等等）提供一个框架。此类的设计目标是成为依靠单个原子 int 值来表示状态的大多数同步器的一个有用基础。

AbstractQueuedSynchronizer是CountDownLatch、ReentrantLock、RenntrantReadWriteLock、Semaphore等类实现的基础。在这几个类中都有这样名为Sync的一个内部类，比如在ReentrantLock中：

abstract static class Sync extends AbstractQueuedSynchronizer {
    ...
}

Sync的主要目的是作为一个同步器，同步器中必须定义更改状态的受保护方法，并定义哪种状态对此同步器意味着被获取或者被释放。完成这些后，Sync中的其他方法就可以实现所有的排队和阻塞机制。

结构示意图

下图是AQS的结构示意图：

如图所示，AQS维护了一个表示状态的STATE和一个FIFO线程等待队列。

STATE
STATE是共享资源，描述当前有多少线程获取了锁，对于互斥锁来说state<=1。。在源码中对应的属性为：

private volatile int state;

通过使用getState()、setState(int) 和/或 compareAndSetState(int, int) 方法可以检查和/或修改同步状态。

protected final int getState() {
return state;
}
protected final void setState(int newState) {
    state = newState;
}
protected final boolean compareAndSetState(int expect, int update) {
// See below for intrinsics setup to support this
return unsafe.compareAndSwapInt(this, stateOffset, expect, update);
}

AQS支持两种资源访问方式，独占模式（Exclusive）和共享模式（Share）。通常，AQS的子类只支持其中一种模式，但也有例外，如ReadWriteLock支持两种模式。只支持独占模式或者只支持共享模式的子类不必定义支持未使用模式的方法。
为了将此类用作同步器的基础，子类需要适当地重新定义以下方法：

• tryAcquire(int)。在独占模式下获取对象状态。
• tryRelease(int)。设置状态来反映独占模式下的一个释放。
• tryAcquireShared(int)。在共享模式下获取对象状态。
• tryReleaseShared(int)。设置状态来反映共享模式下的一个释放。
• isHeldExclusively()。如果对于当前线程，同步是以独占方式进行的，则返回 true。

在AQS中，这几个方法都抛出UnsupportedOperationException。

protected boolean tryAcquire(int arg) {
throw new UnsupportedOperationException();
}
protected boolean tryRelease(int arg) {
throw new UnsupportedOperationException();
}
protected int tryAcquireShared(int arg) {
throw new UnsupportedOperationException();
}
protected boolean tryReleaseShared(int arg) {
throw new UnsupportedOperationException();
}
protected boolean isHeldExclusively() {
throw new UnsupportedOperationException();
}

AQS的子类（锁或者同步器）需要进行重写这几个方法实现共享资源STATE的获取与释放方式。以ReentrantLock为例，ReentrantLock是独占锁，只有独占模式（Exclusive）一种资源访问方式，所以在NonfairSync和FairSync中只有tryAcquire(int)和tryRelease(int)方法。state初始化为0，表示未锁定状态。线程thread调用reentrantLock.lock()时，会调用tryAcquire(1)独占该锁并将state+1。此后，其他线程再tryAcquire()时就会失败，直到线程thread调用unlock()时，会调用tryRelease将state-1，这时state=0，这时其它线程才有机会获取该锁。当然，ReentrantLock是可重入的锁，在释放锁之前，线程thread自己是可以多次调用lock()重复获取此锁的，state会累加。释放锁时要重复调用unlock()释放锁，获取多少次就要释放多么次，这样才能保证state是能回到零态的。
FIFO线程等待队列
CHL队列是一个非阻塞的FIFO队列，通过自旋锁和CAS保证节点插入和移除的原子性。下面是队列的头结点和尾节点。

/**
 * CHL头节点
 */
private transient volatile Node head;

/**
 * CHL尾节点
 */
private transient volatile Node tail;

Node是队列的节点类。

static final class Node {
/** 表示节点处于共享模式 */
static final Node SHARED = new Node();
/** 标记节点处于独占模式 */
static final Node EXCLUSIVE = null;

/** waitStatus的值之一，表示线程被取消了 */
static final int CANCELLED =  1;
/** waitStatus的值之一，表示后继线程需要唤醒 */
static final int SIGNAL    = -1;
/** waitStatus的值之一，表示线程正等待某一等待条件 */
static final int CONDITION = -2;
/** waitStatus的值之一，表示下个acquireShared方法应该无条件地？？？*/
static final int PROPAGATE = -3;

/**
 * 节点的等待状态，以下值的一种:
 * SIGNAL: 节点的继任节点是（或者将要成为）BLOCKED状态（例如通过LockSupport.park()操作），因此一个节点一旦被释放（解锁）或者取消就需要唤醒（LockSupport.unpack()）它的继任节点。
 * CANCELLED: 节点操作因为超时或者对应的线程被interrupt。节点不应该留在此状态，一旦达到此状态将从CHL队列中踢出。
 * CONDITION: 表明节点对应的线程因为不满足一个条件（Condition）而被阻塞。
 * PROPAGATE: 一个releaseShared 应该扩散到其他节点。设置在doReleaseShared方法（只是头结点）中为了确保扩散继续，即使其他操作已经开始执行。
 * 0: 正常状态
 *
 * 非负值意味着节点不需要被唤醒。
 *
 * 这个值被初始化为0，可以使用CAS修改它。
 */
volatile int waitStatus;

/**
 * 前置节点。
 * 头结点不会被取消。
 * 被取消的线程在acquire方法中绝不会成功。
 * 线程只会被自己取消，而不是其他线程。
 */
volatile Node prev;

/**
 * 此节点的后继节点。
 */
volatile Node next;

/**
 * 节点中的线程
 */
volatile Thread thread;

/**
 * 在等待条件上的下个节点
 * 因为condition是独占模式，需要一个简单的队列来保存condition上的节点
 */
    Node nextWaiter;
    ...
    ...
    ...
}

下面开始学习AQS的源码实现。依次学习acquire、release、acquireShared、releaseShared方法。

acquire(int)

/**
 * 以独占模式获取对象，忽略中断。
 * 通过至少调用一次 tryAcquire(int) 来实现此方法，并在成功时返回。
 * 否则在成功之前，一直调用 tryAcquire(int) 将线程加入队列，线程可能重复被阻塞或不被阻塞。
 * 可以使用此方法来实现 Lock.lock() 方法。
 */
public final void acquire(int arg) {
if (!tryAcquire(arg) &&
        acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
        selfInterrupt();
}

在学习tryAcquire(int)等方法之前，先来总体了解下acquire(int)的执行流程：

1. tryAcquire(int)尝试直接去获取资源，如果成功，则acquire的任务就完成了。
2. 如果步骤1失败，addWaiter()根据当前线程创建节点，将当前节点插入等待队列，标记为独占模式，并返回节点。
3. 在步骤2中，节点已经插入到队列末尾，acquireQueued()自旋尝试获取资源。
   
   1. 如果节点的前驱是队列的头结点，这时可以调用tryAcquire(arg)尝试获取资源，如果成功则返回中断位结束acquire(int)。
   2. 如果步骤3.1失败，调用shouldParkAfterFailedAcquire(Node, Node)检测当前节点是否应该park()。
      
      1. 如果前置节点的状态是Node.SIGNAL，说明前置节点释放资源后会通知自己。此时当前节点就可以park()，返回true。
      2. 如果前置节点的状态是Node.CANCELLED，说明前置节点已经放弃获取资源了，此时一直往前找，直到找到最近的一个处于正常等待状态的节点，并排在它的后边。返回false，acquireQueued()自旋，回到3.1步骤。
      3. 如果前置节点处于其他状态，利用CAS将前置节点的状态置为SIGNAL，让它释放资源后通知自己（有可能失败，如果前置节点刚刚释放资源，状态就不是SIGNAL了，这时就会失败）。返回false，acquireQueued()自旋，回到3.1步骤。
   3. 如果shouldParkAfterFailedAcquire(Node, Node)返回true，调用parkAndCheckInterrupt()中断当前节点中的线程。acquireQueued(Node,int)返回true。
4. 回到acquire(int)，acquireQueued(Node,int)如果返回true，调用selfInterrupt()，中断当前线程。

1.tryAcquire(int)
前面已经提到，在AQS中，tryAcquire(int)抛出UnsupportedOperationException。

protected boolean tryAcquire(int arg) {
throw new UnsupportedOperationException();
}

AQS的子类（锁或者同步器）需要进行重写这个方法实现共享资源STATE的获取。以公平的ReentrantLock为例

/**
 * Fair version of tryAcquire. Don't grant access unless
 * recursive call or no waiters or is first.
 */
protected final boolean tryAcquire(int acquires) {
final Thread current = Thread.currentThread();
int c = getState();
if (c == 0) {
if (!hasQueuedPredecessors() &&
            compareAndSetState(0, acquires)) {
            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;
}

从代码中可以看出，如果共享资源state为0，说明该资源还没有被其他线程占用，此时tryAcquire(int)获取资源成功。如果state不为0，且资源的占用者就是当前线程，此时可以重复获取资源（印证了ReentrantLock是可重入的锁），tryAcquire(int)获取资源成功。
2.addWaiter(Node)

private Node addWaiter(Node mode) {
//将当前线程封装成节点，并指定资源访问模式，如独占模式（Exclusive）和共享模式（Share）
    Node node = new Node(Thread.currentThread(), mode);
// 尝试快速方式入队
    Node pred = tail;
if (pred != null) {
        node.prev = pred;
//可以看出，设置尾节点是通过CAS进行的
if (compareAndSetTail(pred, node)) {
            pred.next = node;
return node;
        }
    }
//如果快速方式入队，则通过enq(Node)入队
    enq(node);
//返回当前线程所在的节点
return node;
}

此方法用于将当前线程加入到等待队列的队尾，并返回当前线程所在的节点。
3.acquireQueued(Node, int)

final boolean acquireQueued(final Node node, int arg) {
boolean failed = true;
try {
boolean interrupted = false;
//自旋尝试获取资源
for (;;) {
final Node p = node.predecessor();
//如果节点的前驱是队列的头结点，这时可以调用tryAcquire(arg)尝试获取资源，如果成功则返回中断位结束acquire(int)。
if (p == head && tryAcquire(arg)) {
                setHead(node);
                p.next = null; // help GC
                failed = false;
return interrupted;
            }
//调用shouldParkAfterFailedAcquire(Node, Node)检测当前节点是否应该park()
//如果shouldParkAfterFailedAcquire(Node, Node)返回true，调用parkAndCheckInterrupt()中断当前节点中的线程
if (shouldParkAfterFailedAcquire(p, node) &&
                parkAndCheckInterrupt())
                interrupted = true;
        }
    } finally {
if (failed)
            cancelAcquire(node);
    }
}

3.1shouldParkAfterFailedAcquire(Node, Node)

private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
int ws = pred.waitStatus;
if (ws == Node.SIGNAL)
/*
 * 如果前置节点的状态是Node.SIGNAL，说明前置节点释放资源后会通知自己。
 * 此时当前节点就可以park()，返回true。
 */
return true;
if (ws > 0) {
/*
 * 如果前置节点的状态是Node.CANCELLED，说明前置节点已经放弃获取资源了，
 * 此时一直往前找，直到找到最近的一个处于正常等待状态的节点，
 * 并排在它的后边，返回false。
 */
        do {
            node.prev = pred = pred.prev;
        } while (pred.waitStatus > 0);
        pred.next = node;
    } else {
/*
 * 如果前置节点处于其他状态，利用CAS将前置节点的状态置为SIGNAL，
 * 让它释放资源后通知自己
 * （有可能失败，如果前置节点刚刚释放资源，状态就不是SIGNAL了，这时就会失败）。
 * 返回false
 */
        compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
    }
return false;
}

3.2parkAndCheckInterrupt()
如果shouldParkAfterFailedAcquire(Node, Node)返回true，调用parkAndCheckInterrupt()中断当前节点中的线程。

private final boolean parkAndCheckInterrupt() {
    LockSupport.park(this);
return Thread.interrupted();
}

acquire(int)总结
acquire(int)到这里就结束了，做下总结。acquire(int)尝试获取资源，如果获取失败，将线程插入等待队列。插入等待队列后，acquire(int)并没有放弃获取资源，而是根据前置节点状态状态判断是否应该继续获取资源，如果前置节点是头结点，继续尝试获取资源，如果前置节点是SIGNAL状态，就中断当前线程，否则继续尝试获取资源。直到当前线程被park()或者获取到资源，acquire(int)结束。

release(int)

/**
 * 以独占模式释放对象。
 * 如果 tryRelease(int) 返回 true，则通过消除一个或多个线程的阻塞来实现此方法。
 * 可以使用此方法来实现 Lock.unlock() 方法
 */
public final boolean release(int arg) {
if (tryRelease(arg)) {
        Node h = head;
if (h != null && h.waitStatus != 0)
            unparkSuccessor(h);
return true;
    }
return false;
}

release(int)会释放指定量的资源，如果资源被彻底释放了，即state=0，它会唤醒等待队列里的其他线程来获取资源。
在学习tryRelease(int)等方法之前，先来总体了解下release(int)的执行流程：

1. tryRelease(int)尝试直接去释放资源
   
   1. 如果将资源完全释放了，调用unparkSuccessor(Node)唤醒头结点的后继节点。返回true。
2. 如果tryRelease(int)没有将资源完全释放，返回false。

1.tryRelease(int)
前面已经提到，在AQS中，tryRelease(int)抛出UnsupportedOperationException。

protected boolean tryRelease(int arg) {
throw new UnsupportedOperationException();
}

AQS的子类（锁或者同步器）需要进行重写这个方法实现共享资源STATE的释放。以公平的ReentrantLock为例

protected final boolean tryRelease(int releases) {
//释放量为releases的资源
int c = getState() - releases;
//如果当前线程不是独占模式的，抛出异常
if (Thread.currentThread() != getExclusiveOwnerThread())
throw new IllegalMonitorStateException();
boolean free = false;
//如果当前线程占有的资源已经全部被释放了，返回true
if (c == 0) {
        free = true;
        setExclusiveOwnerThread(null);
    }
    setState(c);
return free;
}

要注意到是，如果资源已经全部被释放了，tryRelease(int)需要返回true。
1.1unparkSuccessor(Node)

private void unparkSuccessor(Node node) {
/*
 * If status is negative (i.e., possibly needing signal) try
 * to clear in anticipation of signalling. It is OK if this
 * fails or if status is changed by waiting thread.
 */
int ws = node.waitStatus;
//如果节点状态小于0，将节点状态置为0，即正常状态
if (ws < 0)
        compareAndSetWaitStatus(node, ws, 0);

//找到符合条件的需要唤醒的节点
    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);
}

此方法用户唤醒指定节点的后继节点。

acquireShared(int)

/**
 * 以共享模式获取对象，忽略中断。
 */
public final void acquireShared(int arg) {
if (tryAcquireShared(arg) < 0)
        doAcquireShared(arg);
}

在学习tryAcquireShared(int)等方法之前，先来总体了解下acquireShared(int)的执行流程：

1. tryAcquireShared(int)尝试直接去获取资源，如果成功，acquireShared(int)就结束了。
2. 如果步骤1失败了，调用doAcquireShared(Node)将线程加入等待队列，直到获取到资源为止。

1.tryAcquireShared(int)
前面已经提到，在AQS中，tryAcquireShared(int)抛出UnsupportedOperationException。

protected boolean tryAcquireShared(int arg) {
throw new UnsupportedOperationException();
}

AQS的子类（锁或者同步器）需要进行重写这个方法实现共享资源STATE的获取。以ReentrantReadWriteLock为例

protected final int tryAcquireShared(int unused) {
    Thread current = Thread.currentThread();
int c = getState();
//如果独占锁（写锁）已经被其他线程获取，tryAcquireShared(int)失败
if (exclusiveCount(c) != 0 &&
        getExclusiveOwnerThread() != current)
return -1;
int r = sharedCount(c);
//如果线程不应该阻塞，且大小没有饱和，且CAS成功，获取资源成功，返回1
if (!readerShouldBlock() &&
        r < MAX_COUNT &&
        compareAndSetState(c, c + SHARED_UNIT)) {
if (r == 0) {
            firstReader = current;
            firstReaderHoldCount = 1;
        } else if (firstReader == current) {
            firstReaderHoldCount++;
        } else {
            HoldCounter rh = cachedHoldCounter;
if (rh == null || rh.tid != getThreadId(current))
                cachedHoldCounter = rh = readHolds.get();
else if (rh.count == 0)
                readHolds.set(rh);
            rh.count++;
        }
return 1;
    }
//如果线程应该阻塞，或大小饱和，或CAS失败，自旋获取资源，如果成功返回1，否则返回-1
return fullTryAcquireShared(current);
}

2.doAcquireShared(arg)

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);
    }
}

可以看出，acquireShared(int)和acquire(int)大体上是一样的，只不过比acquire(int)多了“如果还有剩余资源，唤醒后继节点的线程去获取资源”这一逻辑。

releaseShared(int)

public final boolean releaseShared(int arg) {
if (tryReleaseShared(arg)) {
        doReleaseShared();
return true;
    }
return false;
}

releaseShared(int)以共享模式释放对象。releaseShared(int)与release(int)大体上比较像，区别在于releaseShared(int)在释放一部分资源后就可以通知其他线程获取资源，而release(int)要等资源释放完后才可以。

总结

在AQS的设计中，在父类AQS中实现了对等待队列的默认实现，子类中几乎不用修改该部分功能。而state在子类中根据需要被赋予了不同的意义，子类通过对state的不同操作来提供不同的同步器功能，进而对封装的工具类提供不同的功能。
AQS在JUC里是一个非常重要也是非常复杂的一个类，本章只是对其做简单的介绍，在后面的章节中会介绍AQS的子类和其他的特性。
本文就讲到这里，想了解Java并发编程更多内容请参考：

• Java并发编程札记-目录