java并发之AbstractQueuedSynchronizer同步器

             java并发之AbstractQueuedSynchronizer

一、概述

谈到java的并发编程&＃xff0c;就离不开JUC,JUC指的是java.util.concurrent包&＃xff0c;里面包含了需要java并发的工具类&＃xff0c;而谈到JUC&＃xff0c;就不得不谈AbstractQueuedSynchronizer&＃xff08;AQS&＃xff09;&＃xff01;

AQS:抽象的队列式的同步器&＃xff0c;AQS定义了一套多线程访问共享资源的同步器框架&＃xff0c;许多同步类实现都依赖于它&＃xff0c;如常用的ReentrantLock/Semaphore/CountDownLatch...&＃xff0c;

二、框架解读

AQS内部维护了一个volatile int state(代表资源)变量和一个变种的CLH(FIFO)队列(多线程竞争阻塞时进入队列)&＃xff0c;CLH队列一般用于自旋锁。

state代表一种资源&＃xff0c;AQS提供了3个操作state的方法

• getState
• setState
• compareAndSetState

AQS定义了两种获取资源的方式&＃xff1a;exclusive(独占)与SHARED(共享)&＃xff0c;独占模式获取资源时(如ReentrantLock)&＃xff0c;其他threads无法获取成功&＃xff0c;共享模式获取资源时(如CountDownLatch)&＃xff0c;其他threads可能获取成功。

　不同的自定义同步器争用共享资源的方式也不同。自定义同步器在实现时只需要实现共享资源state的获取与释放方式即可&＃xff0c;至于具体线程等待队列的维护&＃xff08;如获取资源失败入队/唤醒出队等&＃xff09;&＃xff0c;AQS已经在顶层实现好了。自定义同步器实现时主要实现以下几种方法&＃xff1a;

• isHeldExclusively()&＃xff1a;该线程是否正在独占资源。只有用到condition才需要去实现它。
• tryAcquire(int)&＃xff1a;独占方式。尝试获取资源&＃xff0c;成功则返回true&＃xff0c;失败则返回false。
• tryRelease(int)&＃xff1a;独占方式。尝试释放资源&＃xff0c;成功则返回true&＃xff0c;失败则返回false。
• tryAcquireShared(int)&＃xff1a;共享方式。尝试获取资源。负数表示失败&＃xff1b;0表示成功&＃xff0c;但没有剩余可用资源&＃xff1b;正数表示成功&＃xff0c;且有剩余资源。
• tryReleaseShared(int)&＃xff1a;共享方式。尝试释放资源&＃xff0c;如果释放后允许唤醒后续等待结点返回true&＃xff0c;否则返回false。

　　一般来说&＃xff0c;自定义同步器要么是独占方法&＃xff0c;要么是共享方式&＃xff0c;他们也只需实现tryAcquire-tryRelease、tryAcquireShared-tryReleaseShared中的一种即可。但AQS也支持自定义同步器同时实现独占和共享两种方式&＃xff0c;如ReentrantReadWriteLock。

下面我们就以ReentrantLock类来解读下AQS的内部逻辑

三、源码理解

先看下内部类Node&＃xff0c;线程获取资源阻塞后会被封装成Node类&＃xff0c;CLH队列中的node关系如下

&＃43;------&＃43; prev &＃43;-----&＃43; &＃43;-----&＃43;head | | <---- | | <---- | | tail&＃43;------&＃43; &＃43;-----&＃43; &＃43;-----&＃43;

入队只需要拼接tail&＃xff0c;出队需要设置head&＃xff0c;prev代表前驱结点&＃xff0c;next代表后继节点&＃xff0c;初始化队列时&＃xff0c;head&＃61;tail。当前节点所包含的线程依赖于前驱结点的唤醒。CLH队列需要一个虚拟的head&＃xff0c;AQS并没有在构造CLH队列时就初始化head&＃xff0c;而是在竞争出现时构造一个node&＃xff0c;然后让head、tail指向这个node

Node类内部有个waitStatus变量代表结点的状态&＃xff0c;有如下几种取值。

/** waitStatus value to indicate thread has cancelled */ 线程被取消static final int CANCELLED &＃61; 1;/** waitStatus value to indicate successor&＃39;s thread needs unparking */ 后继节点需要唤醒static final int SIGNAL &＃61; -1;/** waitStatus value to indicate thread is waiting on condition */ 后继节点在等待条件static final int CONDITION &＃61; -2;/*** waitStatus value to indicate the next acquireShared should* unconditionally propagate */static final int PROPAGATE &＃61; -3;

acquire()方法解读&＃xff0c;首先调用tryAcquire方法尝试获取资源&＃xff0c;获取资源失败节点入队&＃xff0c;在队列中继续尝试获取资源

/**独占模式下获取资源&＃xff0c;方法实现中的tryAcquire由子类实现* Acquires in exclusive mode, ignoring interrupts. Implemented* by invoking at least once {&＃64;link #tryAcquire},* returning on success. Otherwise the thread is queued, possibly* repeatedly blocking and unblocking, invoking {&＃64;link* #tryAcquire} until success. This method can be used* to implement method {&＃64;link Lock#lock}.** &＃64;param arg the acquire argument. This value is conveyed to* {&＃64;link #tryAcquire} but is otherwise uninterpreted and* can represent anything you like.*/尝试获取资源&＃xff0c;失败后入队,继续尝试获取资源public final void acquire(int arg) {if (!tryAcquire(arg) &&acquireQueued(addWaiter(Node.EXCLUSIVE), arg))selfInterrupt();}

/*** Creates and enqueues node for current thread and given mode.** &＃64;param mode Node.EXCLUSIVE for exclusive, Node.SHARED for shared* &＃64;return the new node* 将当前线程包装成node&＃xff0c;并且入队&＃xff0c;如果队列为空&＃xff0c;初始化一个head&＃xff0c;并且head&＃61;tail&＃xff0c;队列不为空则直接入队*/private Node addWaiter(Node mode) {Node node &＃61; new Node(Thread.currentThread(), mode);// Try the fast path of enq; backup to full enq on failureNode pred &＃61; tail;if (pred !&＃61; null) {node.prev &＃61; pred;if (compareAndSetTail(pred, node)) {pred.next &＃61; node;return node;}}enq(node);return node;}

/*** Inserts node into queue, initializing if necessary. See picture above.* &＃64;param node the node to insert* &＃64;return node&＃39;s predecessor* node入队的具体实现*/private Node enq(final Node node) {for (;;) {Node t &＃61; tail; //t指向tailif (t &＃61;&＃61; null) { // Must initialize//如果tail为空&＃xff0c;则初始化一个node&＃xff0c;并且tail&＃61;headif (compareAndSetHead(new Node()))tail &＃61; head;} else {node.prev &＃61; t;//当前节点前驱指向之前的tail&＃xff0c;并且设置自己为tailif (compareAndSetTail(t, node)) {t.next &＃61; node;return t;}}}}

前面说过尝试获取资源失败会入队等待重新尝试获取资源

/*** Acquires in exclusive uninterruptible mode for thread already in* queue. Used by condition wait methods as well as acquire.** &＃64;param node the node* &＃64;param arg the acquire argument* &＃64;return {&＃64;code true} if interrupted while waiting*/final boolean acquireQueued(final Node node, int arg) {boolean failed &＃61; true;//标记是否获取资源成功try {boolean interrupted &＃61; false;//标记获取资源过程中是否被中断for (;;) {//自旋等等获取资源final Node p &＃61; node.predecessor();//获取前驱结点if (p &＃61;&＃61; head && tryAcquire(arg)) {//如果前驱结点是头部结点&＃xff0c;就尝试获取资源&＃xff0c;因为头部节点可能会释放资源&＃xff0c;然后唤醒后继节点setHead(node);//如果获取资源成功&＃xff0c;则将自己置为headp.next &＃61; null; // help GC 表示当前线程获取资源成功&＃xff0c;则将前驱结点出队failed &＃61; false;return interrupted;//返回获取资源过程中是否被中断}//获取资源失败后是否应该挂起当前线程if (shouldParkAfterFailedAcquire(p, node) &&parkAndCheckInterrupt())//当前线程找到安全点后&＃xff0c;就可以暂时挂起&＃xff0c;等等前驱结点的唤醒interrupted &＃61; true;}} finally {if (failed)cancelAcquire(node);}}

/*** Checks and updates status for a node that failed to acquire.* Returns true if thread should block. This is the main signal* control in all acquire loops. Requires that pred &＃61;&＃61; node.prev** &＃64;param pred node&＃39;s predecessor holding status* &＃64;param node the node* &＃64;return {&＃64;code true} if thread should block*/private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {int ws &＃61; pred.waitStatus;//前驱结点的waitStatusif (ws &＃61;&＃61; Node.SIGNAL)//如果前驱结点waitStatus是SIGNAL&＃xff0c;表示当前节点还需要等待唤醒&＃xff0c;需要挂起/** This node has already set status asking a release* to signal it, so it can safely park.*/return true;if (ws > 0) {//前驱结点waitStatus为非负数表示前驱节点无需唤醒(可能被取消、等等条件)&＃xff0c;就继续寻找正常前驱结点挂靠在后边&＃xff0c;无需唤醒的结点就断了链接&＃xff0c;成为独立的node&＃xff0c;将会被GC回收/** Predecessor was cancelled. Skip over predecessors and* indicate retry.*/do {node.prev &＃61; pred &＃61; pred.prev;} while (pred.waitStatus > 0);pred.next &＃61; node;} else {/** waitStatus must be 0 or PROPAGATE. Indicate that we* need a signal, but don&＃39;t park yet. Caller will need to* retry to make sure it cannot acquire before parking.*/如果前驱结点是正常的&＃xff0c;就设置前驱结点waitStatus为signal&＃xff0c;表示当前结点需要收到通知compareAndSetWaitStatus(pred, ws, Node.SIGNAL);}return false;}


/**挂起当前线程&＃xff0c;唤醒被挂起的线程需要unPark()或者中断* Convenience method to park and then check if interrupted** &＃64;return {&＃64;code true} if interrupted*/private final boolean parkAndCheckInterrupt() {LockSupport.park(this);return Thread.interrupted();}

现在让我们再回到acquireQueued()&＃xff0c;总结下该函数的具体流程&＃xff1a;

1. 结点进入队尾后&＃xff0c;检查状态&＃xff0c;找到安全休息点&＃xff1b;
2. 调用park()进入waiting状态&＃xff0c;等待unpark()或interrupt()唤醒自己&＃xff1b;
3. 被唤醒后&＃xff0c;看自己是不是有资格能拿到号。如果拿到&＃xff0c;head指向当前结点&＃xff0c;并返回从入队到拿到号的整个过程中是否被中断过&＃xff1b;如果没拿到&＃xff0c;继续流程1。

acquireQueued()分析完之后&＃xff0c;我们接下来再回到acquire()&＃xff01;再贴上它的源码吧&＃xff1a;

/*** Acquires in exclusive mode, ignoring interrupts. Implemented* by invoking at least once {&＃64;link #tryAcquire},* returning on success. Otherwise the thread is queued, possibly* repeatedly blocking and unblocking, invoking {&＃64;link* #tryAcquire} until success. This method can be used* to implement method {&＃64;link Lock#lock}.** &＃64;param arg the acquire argument. This value is conveyed to* {&＃64;link #tryAcquire} but is otherwise uninterpreted and* can represent anything you like.*/public final void acquire(int arg) {if (!tryAcquire(arg) &&acquireQueued(addWaiter(Node.EXCLUSIVE), arg))selfInterrupt();}

再来总结下它的流程吧&＃xff1a;

1. 调用自定义同步器的tryAcquire()尝试直接去获取资源&＃xff0c;如果成功则直接返回&＃xff1b;
2. 没成功&＃xff0c;则addWaiter()将该线程加入等待队列的尾部&＃xff0c;并标记为独占模式&＃xff1b;
3. acquireQueued()使线程在等待队列中休息&＃xff0c;有机会时&＃xff08;轮到自己&＃xff0c;会被unpark()&＃xff09;会去尝试获取资源。获取到资源后才返回。如果在整个等待过程中被中断过&＃xff0c;则返回true&＃xff0c;否则返回false。
4. 如果线程在等待过程中被中断过&＃xff0c;它是不响应的。只是获取资源后才再进行自我中断selfInterrupt()&＃xff0c;将中断补上。

整个流程如图所示&＃xff0c;

说完了acquire()方法&＃xff0c;下面来说说release()方法

/*** Releases in exclusive mode. Implemented by unblocking one or* more threads if {&＃64;link #tryRelease} returns true.* This method can be used to implement method {&＃64;link Lock#unlock}.** &＃64;param arg the release argument. This value is conveyed to* {&＃64;link #tryRelease} but is otherwise uninterpreted and* can represent anything you like.* &＃64;return the value returned from {&＃64;link #tryRelease}*/public final boolean release(int arg) {if (tryRelease(arg)) {Node h &＃61; head;//获取头节点if (h !&＃61; null && h.waitStatus !&＃61; 0)//如果头节点不为空&＃xff0c;并且waitStatus不为空则唤醒后继节点unparkSuccessor(h);return true;}return false;}

/*** Wakes up node&＃39;s successor, if one exists.*如果后继节点存在&＃xff0c;则唤醒* &＃64;param node the 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 &＃61; node.waitStatus;if (ws <0)compareAndSetWaitStatus(node, ws, 0);/** Thread to unpark is held in successor, which is normally* just the next node. But if cancelled or apparently null,* traverse backwards from tail to find the actual* non-cancelled successor.*/Node s &＃61; node.next;if (s &＃61;&＃61; null || s.waitStatus > 0) {//waitStatus大于0&＃xff0c;说明节点被cancelled&＃xff0c;从tail开始寻找需要唤醒的后继节点s &＃61; null;for (Node t &＃61; tail; t !&＃61; null && t !&＃61; node; t &＃61; t.prev)if (t.waitStatus <&＃61; 0)s &＃61; t;}if (s !&＃61; null)//后继节点不为空则唤醒&＃xff0c;唤醒之后就可以继续尝试获取资源LockSupport.unpark(s.thread);}

逻辑并不复杂。它调用tryRelease()来释放资源。有一点需要注意的是&＃xff0c;它是根据tryRelease()的返回值来判断该线程是否已经完成释放掉资源了&＃xff01;所以自定义同步器在设计tryRelease()的时候要明确这一点&＃xff01;&＃xff01;

这里的唤醒线程需要结合acquireQueued()方法一起来理解&＃xff0c;可以看到这里线程被unParked后就可以继续自旋尝试获取资源。

release()是独占模式下线程释放共享资源的顶层入口。它会释放指定量的资源&＃xff0c;如果彻底释放了&＃xff08;即state&＃61;0&＃xff09;,它会唤醒等待队列里的其他线程来获取资源。

既然源码看完了&＃xff0c;我们来看下怎么使用AQS,以ReentrantLock为例子。ReentrantLock是一个可重入锁&＃xff0c;ReentrantLock的内部类Sync实现了AQS接口&＃xff0c;这是可重入锁的同步控制器&＃xff0c;使用AQS的state字段代表锁的持有数。可重入锁有两种模式&＃xff0c;公平和非公平模式&＃xff0c;默认为非公平模式&＃xff0c;先看下fair模式。

/*** Creates an instance of {&＃64;code ReentrantLock} with the* given fairness policy.** &＃64;param fair {&＃64;code true} if this lock should use a fair ordering policy*/构造方法public ReentrantLock(boolean fair) {sync &＃61; fair ? new FairSync() : new NonfairSync();}

/*** Sync object for fair locks*/static final class FairSync extends Sync {private static final long serialVersionUID &＃61; -3000897897090466540L;final void lock() {acquire(1);}/*** Fair version of tryAcquire. Don&＃39;t grant access unless* recursive call or no waiters or is first.*/protected final boolean tryAcquire(int acquires) {final Thread current &＃61; Thread.currentThread();//获取当前线程int c &＃61; getState();if (c &＃61;&＃61; 0) {//state&＃61;&＃61;0表示资源空闲if (!hasQueuedPredecessors() &&//是否有前驱结点compareAndSetState(0, acquires)) {//设置资源setExclusiveOwnerThread(current);//设置锁的拥有者为当前线程&＃xff0c;重入标志return true;}}else if (current &＃61;&＃61; getExclusiveOwnerThread()) {//如果资源不是空闲的&＃xff0c;并且锁的拥有者为当前线程&＃xff0c;则表示可重入int nextc &＃61; c &＃43; acquires;if (nextc <0)throw new Error("Maximum lock count exceeded");setState(nextc);return true;}return false;}}

tryRelease()

ReentrantLock的tryRelease方法&＃xff0c;重入多少次就释放多少次protected final boolean tryRelease(int releases) {int c &＃61; getState() - releases;if (Thread.currentThread() !&＃61; getExclusiveOwnerThread())throw new IllegalMonitorStateException();boolean free &＃61; false;if (c &＃61;&＃61; 0) {//state表示释放完毕&＃xff0c;否则锁还是未释放free &＃61; true;setExclusiveOwnerThread(null);}setState(c);return free;}

再来看下nonFair模式

/*** Sync object for non-fair locks非公平模式下&＃xff0c;首先直接参与获取资源而不用排队&＃xff0c;这就是非公平的含义*/static final class NonfairSync extends Sync {private static final long serialVersionUID &＃61; 7316153563782823691L;/*** Performs lock. Try immediate barge, backing up to normal* acquire on failure.*/final void lock() {if (compareAndSetState(0, 1))setExclusiveOwnerThread(Thread.currentThread());elseacquire(1);}protected final boolean tryAcquire(int acquires) {return nonfairTryAcquire(acquires);}}

看懂了AQS的逻辑&＃xff0c;再去理解JUC下面的一些包就容易多了

上面的acquire(),release()方法都是独占模式下的资源操作

下面说说共享模式的acquireShared()&＃xff0c;releaseShared()