JUC并发—6.AQS源码分析二
大纲1.ReentractReadWriteLock的基本原理
2.基于AQS实现的ReentractReadWriteLock
3.ReentractReadWriteLock如何竞争写锁
4.ReentractReadWriteLock如何竞争读锁
5.ReentractReadWriteLock的公平锁和非公平锁
6.ReentrantReadWriteLock中的锁降级
7.Condition的说明介绍
8.Condition的源码实现
1.ReentractReadWriteLock的基本原理
(1)读锁和写锁关系
表面上读锁和写锁是两把锁，但实际上只是同一把锁的两个视图。读锁和写锁在初始化的时候会共用一个Sync，也就是同一把锁、两类线程。其中读线程和读线程不互斥，读线程和写线程互斥，写线程和写线程互斥。
(2)锁状态设计
和独占锁一样，读写锁也是使用state变量来表示锁的状态。只是将state变量拆成两半：高16位表示读锁状态，低16位表示写锁状态。
读写锁是通过位运算来快速确定读和写的状态的。假设当前state = s，则写状态等于s & ((1 << 16) - 1)，读状态等于s >>> 16。当写状态增加1时，state = s + 1。当读状态加1时，state = s + (1 << 16)。
将一个int型的state变量拆成两半，而不是用两个int型变量分别表示读锁和写锁的状态，是因为无法用一次CAS同时操作两个int型变量。
1 << 0 等于 1，(1 << 0) - 1 = 01 << 1 等于 10，(1 << 1) - 1 = 011 << 2 等于 100，(1 << 2) - 1 = 0111 << 4 等于 1000，(1 << 4) - 1 = 01111 << 8 等于 100000000，(1 << 8) - 1 = 0111111111 << 16 等于 10000000000000000，(1 << 16) - 1 = 01111111111111111//所以s & ((1 << 16) - 1)相当于将s的高16位全部抹去，只剩下低16位//若s = 11111，则s >>> 2 = 00111//所以s >>> 16，就是无符号补0右移16位(3)写锁的获取与释放
写锁是一个可重入的排他锁，它只能被一个线程同时获取。如果当前线程已获取写锁，则增加写状态：s + 1。如果当前线程在获取写锁时，读锁已被获取或者自己不是已获写锁的线程，则进入等待状态。
(4)读锁的获取与释放
读锁是一个可重入的共享锁，它能被多个线程同时获取。在写状态为0时，读锁总会被成功获取，而所做的也只是增加读状态。如果当前线程已获取读锁，则增加读状态：state = s + (1 << 16)。如果当前线程在获取读锁时，写锁已被其他其他线程获取，则该线程进入等待状态。
2.基于AQS实现的ReentractReadWriteLock
(1)ReentractReadWriteLock的构造函数
(2)ReentractReadWriteLock的读锁和写锁
(3)ReentractReadWriteLock基于AQS的两对模版方法
(1)ReentractReadWriteLock的构造函数
readerLock变量表示读锁，writerLock变量表示写锁，sync变量表示公平锁还是非公平锁。
public class ReentrantReadWriteLock implements ReadWriteLock, java.io.Serializable { //Inner class providing readlock private final ReentrantReadWriteLock.ReadLock readerLock; //Inner class providing writelock private final ReentrantReadWriteLock.WriteLock writerLock; //Performs all synchronization mechanics final Sync sync; //Creates a new ReentrantReadWriteLock with default (nonfair) ordering properties. public ReentrantReadWriteLock() { this(false); } //Creates a new ReentrantReadWriteLock with the given fairness policy. //@param fair {@code true} if this lock should use a fair ordering policy public ReentrantReadWriteLock(boolean fair) { sync = fair ? new FairSync() : new NonfairSync(); readerLock = new ReadLock(this); writerLock = new WriteLock(this); } //获取写锁 public ReentrantReadWriteLock.WriteLock writeLock() { return writerLock; } //获取读锁 public ReentrantReadWriteLock.ReadLockreadLock(){ return readerLock; } ...}(2)ReentractReadWriteLock的读锁和写锁
读锁和写锁都会通过内部类Sync来实现获取锁和释放锁的功能。加写锁会调用AQS的acquire()方法，然后调用tryAcquire()方法。加读锁会调用AQS的acquireShared()方法，然后调用acquireShared()方法。
public class ReentrantReadWriteLock implements ReadWriteLock, java.io.Serializable { ... //Performs all synchronization mechanics final Sync sync; //The lock returned by method {@link ReentrantReadWriteLock#readLock}. public static class ReadLock implements Lock, java.io.Serializable { private final Sync sync; protected ReadLock(ReentrantReadWriteLock lock) { sync = lock.sync; } //Acquires the read lock. //Acquires the read lock if the write lock is not held by another thread and returns immediately. //If the write lock is held by another thread then the current thread becomes disabled //for thread scheduling purposes and lies dormant until the read lock has been acquired. public void lock() { sync.acquireShared(1); } //Acquires the read lock unless the current thread is Thread#interrupt interrupted. public void lockInterruptibly() throws InterruptedException { sync.acquireSharedInterruptibly(1); } //Acquires the read lock only if the write lock is not held by another thread at the time of invocation. public boolean tryLock() { return sync.tryReadLock(); } //Acquires the read lock if the write lock is not held by another thread //within the given waiting time and the current thread has not been Thread#interrupt interrupted. public boolean tryLock(long timeout, TimeUnit unit) throws InterruptedException { return sync.tryAcquireSharedNanos(1, unit.toNanos(timeout)); } //Attempts to release this lock. //If the number of readers is now zero then the lock is made available for write lock attempts. public void unlock() { sync.releaseShared(1); } ... } //The lock returned by method {@link ReentrantReadWriteLock#writeLock}. public static class WriteLock implements Lock, java.io.Serializable { private final Sync sync; protected WriteLock(ReentrantReadWriteLock lock) { sync = lock.sync; } //Acquires the write lock. //Acquires the write lock if neither the read nor write lock are held by another thread and returns immediately, //setting the write lock hold count to one. //If the current thread already holds the write lock then the hold count is incremented by one and the method return immediately. //If the lock is held by another thread then the current thread becomes disabled //for thread scheduling purposes and lies dormant until the write lock has been acquired, //at which time the write lock hold count is set to one. public void lock() { sync.acquire(1); } //Acquires the write lock unless the current thread is Thread#interrupt interrupted. public void lockInterruptibly() throws InterruptedException { sync.acquireInterruptibly(1); } //Acquires the write lock only if it is not held by another thread at the time of invocation. public boolean tryLock( ) { return sync.tryWriteLock(); } //Acquires the write lock if it is not held by another thread //within the given waiting time and the current thread has not been Thread#interrupt interrupted. public boolean tryLock(long timeout, TimeUnit unit) throws InterruptedException { return sync.tryAcquireNanos(1, unit.toNanos(timeout)); } //Attempts to release this lock. //If the current thread is the holder of this lock then the hold count is decremented. //If the hold count is now zero then the lock is released. //If the current thread is not the holder of this lock then IllegalMonitorStateException is thrown. public void unlock() { sync.release(1); } ... } ...}(3)ReentractReadWriteLock基于AQS的两对模版方法
独占锁和读写锁的写锁都是基于AQS的acquire/release模版方法实现的，读写锁的读锁是基于AQS的acquireShared/releaseShared模版方法实现的。
public abstract class AbstractQueuedSynchronizer extends AbstractOwnableSynchronizer implements java.io.Serializable { ... //Acquires in exclusive mode, ignoring interrupts. //Implemented by invoking at least once #tryAcquire, returning on success. //Otherwise the thread is queued, possibly repeatedly blocking and unblocking, invoking #tryAcquire until success. //This method can be used to implement method Lock#lock. public final void acquire(int arg) { //tryAcquire()需要子类重写 if (!tryAcquire(arg) && acquireQueued(addWaiter(Node.EXCLUSIVE), arg)) { selfInterrupt(); } } protected boolean tryAcquire(int arg) { throw new UnsupportedOperationException(); } //Releases in exclusive mode. //Implemented by unblocking one or more threads if #tryRelease returns true. //This method can be used to implement method Lock#unlock. public final boolean release(int arg) { //tryRelease()需要子类重写 if (tryRelease(arg)) { Node h = head; if (h != null && h.waitStatus != 0) { unparkSuccessor(h); } return true; } return false; } protected boolean tryRelease(int arg) { throw new UnsupportedOperationException(); } //Acquires in shared mode, ignoring interrupts. //Implemented by first invoking at least once #tryAcquireShared, returning on success. //Otherwise the thread is queued, possibly repeatedly blocking and unblocking, invoking #tryAcquireShared until success. public final void acquireShared(int arg) { //tryAcquireShared()需要子类重写 if (tryAcquireShared(arg) < 0) { doAcquireShared(arg); } } protected int tryAcquireShared(int arg) { throw new UnsupportedOperationException(); } //Releases in shared mode. //Implemented by unblocking one or more threads if #tryReleaseShared returns true. public final boolean releaseShared(int arg) { //tryReleaseShared()需要子类重写 if (tryReleaseShared(arg)) { doReleaseShared(); return true; } return false; } protected boolean tryReleaseShared(int arg) { throw new UnsupportedOperationException(); } ...}
3.ReentractReadWriteLock如何竞争写锁
(1)WriteLock的获取
(2)WriteLock的释放
(1)WriteLock的获取
WriteLock的lock()方法会调用AQS的acquire()模版方法来获取锁，而AQS的acquire()方法又会调用继承自AQS的Sync类的tryAcquire()方法。
在Sync类的tryAcquire()方法中，getState()方法会返回当前state变量的值。exclusiveCount()方法会从state变量中查找当前获得写锁的线程数量，writerShouldBlock()方法会判断当前写线程在抢占锁时是否应该阻塞。
情况一：c != 0 && w == 0
说明此时有线程持有读锁，所以当前线程获取不到写锁，返回false。由此可见，一个线程获取读锁后不能再继续重入获取写锁(不能锁升级)。但从后续可知，一个线程获取写锁后可以再继续重入获取读锁(能锁降级)。
情况二：c != 0 && w != 0
说明此时有线程持有写锁且不可能有线程持有读锁，所以需要判断持有写锁的线程是否是当前线程自己，如果不是则返回false。
情况三：c != 0 && w != 0 && current持有锁
说明此时当前线程正在持有写锁，属于重入写锁的情况，需要判断重入次数，锁重入的次数不能大于65535。
情况四：c == 0
说明此时没有线程持有锁，所以当前线程可以通过CAS操作抢占锁。
public class ReentrantReadWriteLock implements ReadWriteLock, java.io.Serializable { ... //Performs all synchronization mechanics final Sync sync; //The lock returned by method {@link ReentrantReadWriteLock#writeLock}. public static class WriteLock implements Lock, java.io.Serializable { private final Sync sync; protected WriteLock(ReentrantReadWriteLock lock) { sync = lock.sync; } //Acquires the write lock. //Acquires the write lock if neither the read nor write lock are held by another thread and returns immediately, //setting the write lock hold count to one. //If the current thread already holds the write lock then the hold count is incremented by one and the method return immediately. //If the lock is held by another thread then the current thread becomes disabled //for thread scheduling purposes and lies dormant until the write lock has been acquired, //at which time the write lock hold count is set to one. public void lock() { //执行AQS的acquire()模版方法，获取写锁(独占锁) sync.acquire(1); } ... } //Synchronization implementation for ReentrantReadWriteLock. //Subclassed into fair and nonfair versions. abstract static class Sync extends AbstractQueuedSynchronizer { static final int MAX_COUNT = (1 << SHARED_SHIFT) - 1; static final int EXCLUSIVE_MASK = (1 << SHARED_SHIFT) - 1; //Returns the number of exclusive holds represented in count static int exclusiveCount(int c) { //获取写锁状态：c & ((1 << 16) - 1) //也就是从state变量中查找当前获得写锁的线程数量 return c & EXCLUSIVE_MASK; } //获取写锁(独占锁) protected final boolean tryAcquire(int acquires) { //Walkthrough: //1.If read count nonzero or write count nonzero and owner is a different thread, fail. //2.If count would saturate, fail. (This can only happen if count is already nonzero.) //3. Otherwise, this thread is eligible for lock if it is either a reentrant acquire or queue policy allows it. //If so, update state and set owner. Thread current = Thread.currentThread(); int c = getState();//获取锁的状态 int w = exclusiveCount(c);//获取写锁的状态 if (c != 0) { //如果c != 0 && w == 0时，说明有线程持有读锁，所以当前获取写锁的线程会被阻塞，会返回false //如果c != 0 && w != 0 && current没获取锁，说明有其他线程持有写锁(不可能有线程持有读锁)，所以当前获取写锁的线程会被阻塞，会返回false if (w == 0 || current != getExclusiveOwnerThread()) { return false; } //判断重入次数不能大于65535 if (w + exclusiveCount(acquires) > MAX_COUNT) { throw new Error("Maximum lock count exceeded"); } //Reentrant acquire setState(c + acquires); return true; } //此时c == 0，说明没有线程持有锁，可以通过CAS操作抢占锁 if (writerShouldBlock() || !compareAndSetState(c, c + acquires)) { return false; } setExclusiveOwnerThread(current); return true; } ... }}public abstract class AbstractQueuedSynchronizer extends AbstractOwnableSynchronizer implements java.io.Serializable { ... //获取写锁(独占锁) //Acquires in exclusive mode, ignoring interrupts. //Implemented by invoking at least once #tryAcquire, returning on success. //Otherwise the thread is queued, possibly repeatedly blocking and unblocking, invoking #tryAcquire until success. //This method can be used to implement method Lock#lock. public final void acquire(int arg) { //tryAcquire()需要子类重写，此时执行ReentrantReadWriteLock的内部类Sync的tryAcquire()方法 if (!tryAcquire(arg) && acquireQueued(addWaiter(Node.EXCLUSIVE), arg)) { selfInterrupt(); } } ...}(2)WriteLock的释放
WriteLock的unlock()方法会调用AQS的release()模版方法来释放锁，而AQS的release()方法又会调用继承自AQS的Sync类的tryRelease()方法。
在Sync类的tryRelease()方法中，首先通过getState() - releases来递减写锁的次数。由于写锁的重入次数保存在低位，所以直接十进制相减即可。然后通过exclusiveCount()获取写锁的重入次数，如果为0说明锁释放成功。最后通过setState()方法修改state变量的值。由于写锁是独占锁，所以设置state变量的值不需要CAS操作。
public class ReentrantReadWriteLock implements ReadWriteLock, java.io.Serializable { ... //Performs all synchronization mechanics final Sync sync; //The lock returned by method {@link ReentrantReadWriteLock#writeLock}. public static class WriteLock implements Lock, java.io.Serializable { private final Sync sync; protected WriteLock(ReentrantReadWriteLock lock) { sync = lock.sync; } //Attempts to release this lock. //If the current thread is the holder of this lock then the hold count is decremented. //If the hold count is now zero then the lock is released. //If the current thread is not the holder of this lock then IllegalMonitorStateException is thrown. public void unlock() { //执行AQS的release()方法，释放写锁(独占锁) sync.release(1); } ... } //Synchronization implementation for ReentrantReadWriteLock. //Subclassed into fair and nonfair versions. abstract static class Sync extends AbstractQueuedSynchronizer { static final int MAX_COUNT = (1 << SHARED_SHIFT) - 1; static final int EXCLUSIVE_MASK = (1 << SHARED_SHIFT) - 1; //Returns the number of exclusive holds represented in count static int exclusiveCount(int c) { //获取写锁状态：c & ((1 << 16) - 1) //也就是从state变量中查找当前获得写锁的线程数量 return c & EXCLUSIVE_MASK; } //Note that tryRelease and tryAcquire can be called by Conditions. //So it is possible that their arguments contain both read and write holds //that are all released during a condition wait and re-established in tryAcquire. protected final boolean tryRelease(int releases) { if (!isHeldExclusively()) { throw new IllegalMonitorStateException(); } int nextc = getState() - releases;//递减写锁的次数 boolean free = exclusiveCount(nextc) == 0;//计算写锁的重入次数 if (free) { setExclusiveOwnerThread(null); } setState(nextc); return free; } ... } ...}public abstract class AbstractQueuedSynchronizer extends AbstractOwnableSynchronizer implements java.io.Serializable { ... //释放写锁(独占锁) //Releases in exclusive mode. //Implemented by unblocking one or more threads if #tryRelease returns true. //This method can be used to implement method Lock#unlock. public final boolean release(int arg) { //tryRelease()需要子类重写，此时执行ReentrantReadWriteLock的内部类Sync的tryRelease()方法 if (tryRelease(arg)) { Node h = head; if (h != null && h.waitStatus != 0) { unparkSuccessor(h); } return true; } return false; } ...}
4.ReentractReadWriteLock如何竞争读锁
(1)ReadLock的获取
(2)ReadLock的释放
(3)fullTryAcquireShared()方法
(1)ReadLock的获取
ReadLock的lock()方法会调用AQS的acquireShared()模版方法来获取读锁，而AQS的acquireShared()方法又会调用Sync类的tryAcquireShared()方法。
在继承自AQS的Sync类的tryAcquireShared()方法中：首先会判断是否有线程持有写锁 + 持有写锁的线程是否是当前线程。如果有线程持有写锁，但不是当前线程持有写锁，那么会阻塞当前线程。然后判断当前线程获取读锁是否应该阻塞，读锁重入次数是否小于65535，以及通过CAS修改state值来抢占读锁是否成功。
如果当前线程获取读锁不应该被阻塞，读锁重入次数也小于65535，且CAS抢占读锁也成功，那么就使用ThreadLocal记录线程重入读锁的次数。否则，就继续调用fullTryAcquireShared()方法通过自旋尝试获取锁。
如果调用Sync的tryAcquireShared()方法返回-1，则调用AQS的doAcquireShared()方法入队等待队列和阻塞当前线程。
在等待队列中，如果等待获取读锁的线程被唤醒，那么会继续循环把其后连续的所有等待获取读锁的线程都唤醒，直到遇到一个等待获取写锁的线程为止。
public class ReentrantReadWriteLock implements ReadWriteLock, java.io.Serializable { ... //Performs all synchronization mechanics final Sync sync; //The lock returned by method {@link ReentrantReadWriteLock#readLock}. public static class ReadLock implements Lock, java.io.Serializable { private final Sync sync; protected ReadLock(ReentrantReadWriteLock lock) { sync = lock.sync; } //Acquires the read lock. //Acquires the read lock if the write lock is not held by another thread and returns immediately. //If the write lock is held by another thread then the current thread becomes disabled //for thread scheduling purposes and lies dormant until the read lock has been acquired. public void lock() { //执行AQS的acquireShared()方法，获取读锁(共享锁) sync.acquireShared(1); } ... } //Synchronization implementation for ReentrantReadWriteLock. //Subclassed into fair and nonfair versions. abstract static class Sync extends AbstractQueuedSynchronizer { static final int SHARED_SHIFT = 16; static final int SHARED_UNIT = (1 << SHARED_SHIFT); static final int EXCLUSIVE_MASK = (1 << SHARED_SHIFT) - 1; //Returns the number of exclusive holds represented in count static int exclusiveCount(int c) { //获取写锁状态：c & ((1 << 16) - 1) //也就是从state变量中查找当前获得写锁的线程数量 return c & EXCLUSIVE_MASK; } //Returns the number of shared holds represented in count static int sharedCount(int c) { //获取读锁状态：c >>> 16 //也就是从state变量中查找当前获得读锁的线程数量 return c >>> SHARED_SHIFT; } //A counter for per-thread read hold counts. //Maintained as a ThreadLocal; cached in cachedHoldCounter static final class HoldCounter { int count = 0; //Use id, not reference, to avoid garbage retention final long tid = getThreadId(Thread.currentThread()); } //ThreadLocal subclass. Easiest to explicitly define for sake of deserialization mechanics. static final class ThreadLocalHoldCounter extends ThreadLocal<HoldCounter> { public HoldCounter initialValue() { return new HoldCounter(); } } //The number of reentrant read locks held by current thread. //Initialized only in constructor and readObject. //Removed whenever a thread's read hold count drops to 0. private transient ThreadLocalHoldCounter readHolds; //The hold count of the last thread to successfully acquire readLock. //This saves ThreadLocal lookup in the common case where the next thread to release is the last one to acquire. //This is non-volatile since it is just used as a heuristic, and would be great for threads to cache. private transient HoldCounter cachedHoldCounter; //firstReader is the first thread to have acquired the read lock. //firstReaderHoldCount is firstReader's hold count. private transient Thread firstReader = null; private transient int firstReaderHoldCount; Sync() { readHolds = new ThreadLocalHoldCounter(); setState(getState());// ensures visibility of readHolds } //获取读锁(共享锁) protected final int tryAcquireShared(int unused) { //Walkthrough: //1.If write lock held by another thread, fail. //2.Otherwise, this thread is eligible for lock wrt state, //so ask if it should block because of queue policy. //If not, try to grant by CASing state and updating count. //Note that step does not check for reentrant acquires, //which is postponed to full version to avoid having to check hold count in the more typical non-reentrant case. //3.If step 2 fails either because thread apparently not eligible or CAS fails or count saturated, //chain to version with full retry loop. Thread current = Thread.currentThread(); int c = getState(); //如果写锁的持有线程不是当前线程，则直接阻塞 //从而说明，如果一个线程先获取了写锁，那么是可以重入获取读锁的 if (exclusiveCount(c) != 0 && getExclusiveOwnerThread() != current) { return -1; } int r = sharedCount(c);//获取读锁的状态 //首先判断当前线程获取读锁是否应该阻塞，然后判断读锁重入次数是否小于65535，最后通过CAS修改state值抢占读锁 if (!readerShouldBlock() && r < MAX_COUNT && compareAndSetState(c, c + SHARED_UNIT)) { //使用ThreadLocal记录每个线程重入读锁的次数 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; } //如果当前线程获取读锁失败，则调用fullTryAcquireShared()方法通过自旋尝试获取锁 return fullTryAcquireShared(current); } ... } ...}public abstract class AbstractQueuedSynchronizer extends AbstractOwnableSynchronizer implements java.io.Serializable { ... //获取读锁(共享锁) //Acquires in shared mode, ignoring interrupts. //Implemented by first invoking at least once #tryAcquireShared, returning on success. //Otherwise the thread is queued, possibly repeatedly blocking and unblocking, invoking #tryAcquireShared until success. public final void acquireShared(int arg) { //tryAcquireShared()需要子类重写，此时执行ReentrantReadWriteLock的内部类Sync的tryAcquireShared()方法 if (tryAcquireShared(arg) < 0) { //调用AQS的doAcquireShared()方法入队等待队列和阻塞当前线程 doAcquireShared(arg); } } //Acquires in shared uninterruptible mode. 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; } } //执行shouldParkAfterFailedAcquire()方法设置node结点的前驱结点的状态为SIGNAL //执行parkAndCheckInterrupt()方法挂起当前线程 if (shouldParkAfterFailedAcquire(p, node) && parkAndCheckInterrupt()) { interrupted = true; } } } finally { if (failed) { cancelAcquire(node); } } } ...}(2)ReadLock的释放
ReadLock的unlock()方法会调用AQS的releaseShared()模版方法来释放锁，而AQS的releaseShared()方法又会调用Sync类的tryReleaseShared()方法。
在Sync类的tryReleaseShared()方法中：首先会结合ThreadLocal处理当前线程重入读锁的次数，然后再通过自旋 + CAS设置state值来实现释放读锁，最后执行AQS的doReleaseShared()方法唤醒阻塞的线程。
tryRelease()和tryReleaseShared()的区别：读锁是共享锁，由多个线程持有，所以释放读锁需要通过自旋 + CAS完成。写锁是独占锁，由单个线程持有，所以释放写锁时不需要CAS操作。
public class ReentrantReadWriteLock implements ReadWriteLock, java.io.Serializable { ... //Performs all synchronization mechanics final Sync sync; //The lock returned by method {@link ReentrantReadWriteLock#readLock}. public static class ReadLock implements Lock, java.io.Serializable { private final Sync sync; protected ReadLock(ReentrantReadWriteLock lock) { sync = lock.sync; } //Attempts to release this lock. //If the number of readers is now zero then the lock is made available for write lock attempts. public void unlock() { //执行AQS的releaseShared()方法，释放读锁(共享锁) sync.releaseShared(1); } ... } //Synchronization implementation for ReentrantReadWriteLock. //Subclassed into fair and nonfair versions. abstract static class Sync extends AbstractQueuedSynchronizer { ... //释放读锁(共享锁) protected final boolean tryReleaseShared(int unused) { Thread current = Thread.currentThread(); //首先结合ThreadLocal处理当前线程重入读锁的次数 if (firstReader == current) { if (firstReaderHoldCount == 1) { firstReader = null; } else { firstReaderHoldCount--; } } else { HoldCounter rh = cachedHoldCounter; if (rh == null || rh.tid != getThreadId(current)) { rh = readHolds.get(); } int count = rh.count; if (count <= 1) { readHolds.remove(); if (count <= 0) { throw unmatchedUnlockException(); } } --rh.count; } //然后通过自旋 + CAS设置state值来实现释放读锁 for (;;) { int c = getState(); int nextc = c - SHARED_UNIT; if (compareAndSetState(c, nextc)) { //Releasing the read lock has no effect on readers, //but it may allow waiting writers to proceed if both read and write locks are now free. return nextc == 0; } } } ... } ...}public abstract class AbstractQueuedSynchronizer extends AbstractOwnableSynchronizer implements java.io.Serializable { ... //释放读锁(共享锁) //Releases in shared mode. //Implemented by unblocking one or more threads if #tryReleaseShared returns true. public final boolean releaseShared(int arg) { //tryReleaseShared()需要子类重写，此时执行ReentrantReadWriteLock的内部类Sync的tryReleaseShared()方法 if (tryReleaseShared(arg)) { //执行AQS的doReleaseShared()方法唤醒阻塞的线程 doReleaseShared(); return true; } return false; } //Release action for shared mode -- signals successor and ensures propagation. //Note: For exclusive mode, release just amounts to calling unparkSuccessor of head if it needs signal. private void doReleaseShared() { //Ensure that a release propagates, even if there are other in-progress acquires/releases. //This proceeds in the usual way of trying to unparkSuccessor of head if it needs signal. //But if it does not, status is set to PROPAGATE to ensure that upon release, propagation continues. //Additionally, we must loop in case a new node is added while we are doing this. //Also, unlike other uses of unparkSuccessor, we need to know if CAS to reset status fails, if so rechecking. for (;;) { Node h = head; if (h != null && h != tail) { int ws = h.waitStatus; if (ws == Node.SIGNAL) { if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0)) { //loop to recheck cases continue; } //执行AQS的unparkSuccessor()方法 unparkSuccessor(h); } else if (ws == 0 && !compareAndSetWaitStatus(h, 0, Node.PROPAGATE)) { //loop on failed CAS continue; } } //loop if head changed if (h == head) { break; } } } //Wakes up node's successor, if one exists. 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. //Node结点的状态watiStatus可以分为如下几种： //默认(0)、CANCELED(1)、SIGNAL(-1)、CONDITION(-2)、PROPAGATE(-3) //默认情况下，watiStatus应该是0，或者是空 //获得头结点的状态 int ws = node.waitStatus; //需要设置头结点的状态为0 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 = node.next; //如果头结点的后继结点为null或其状态为CANCELED 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); } } ...}(3)fullTryAcquireShared()方法
在如下两种情况下，该方法会返回-1让当前线程加入等待队列进行等待。
情况一：此时有线程获得了写锁但不是当前线程获得写锁后重入读锁。
情况二：readerShouldBlock()方法返回true且不是重入读锁。
public class ReentrantReadWriteLock implements ReadWriteLock, java.io.Serializable { ... //Synchronization implementation for ReentrantReadWriteLock. //Subclassed into fair and nonfair versions. abstract static class Sync extends AbstractQueuedSynchronizer { static final int SHARED_SHIFT = 16; static final int SHARED_UNIT = (1 << SHARED_SHIFT); static final int EXCLUSIVE_MASK = (1 << SHARED_SHIFT) - 1; //Returns the number of exclusive holds represented in count static int exclusiveCount(int c) { //获取写锁状态：c & ((1 << 16) - 1) //也就是从state变量中查找当前获得写锁的线程数量 return c & EXCLUSIVE_MASK; } //Returns the number of shared holds represented in count static int sharedCount(int c) { //获取读锁状态：c >>> 16 //也就是从state变量中查找当前获得读锁的线程数量 return c >>> SHARED_SHIFT; } //A counter for per-thread read hold counts. //Maintained as a ThreadLocal; cached in cachedHoldCounter static final class HoldCounter { int count = 0; //Use id, not reference, to avoid garbage retention final long tid = getThreadId(Thread.currentThread()); } //ThreadLocal subclass. Easiest to explicitly define for sake of deserialization mechanics. static final class ThreadLocalHoldCounter extends ThreadLocal<HoldCounter> { public HoldCounter initialValue() { return new HoldCounter(); } } //The number of reentrant read locks held by current thread. //Initialized only in constructor and readObject. //Removed whenever a thread's read hold count drops to 0. private transient ThreadLocalHoldCounter readHolds; //The hold count of the last thread to successfully acquire readLock. //This saves ThreadLocal lookup in the common case where the next thread to release is the last one to acquire. //This is non-volatile since it is just used as a heuristic, and would be great for threads to cache. private transient HoldCounter cachedHoldCounter; //firstReader is the first thread to have acquired the read lock. //firstReaderHoldCount is firstReader's hold count. private transient Thread firstReader = null; private transient int firstReaderHoldCount; Sync() { readHolds = new ThreadLocalHoldCounter(); setState(getState());// ensures visibility of readHolds } ... //Full version of acquire for reads, //that handles CAS misses and reentrant reads not dealt with in tryAcquireShared. final int fullTryAcquireShared(Thread current) { //This code is in part redundant with that in tryAcquireShared //but is simpler overall by not complicating tryAcquireShared with interactions //between retries and lazily reading hold counts. HoldCounter rh = null; for (;;) { int c = getState(); if (exclusiveCount(c) != 0) { //如果当前有其他线程获取到写锁，那么返回-1，当前线程会挂起阻塞并进入等待队列 if (getExclusiveOwnerThread() != current) { return -1; } //else we hold the exclusive lock; blocking here would cause deadlock. } else if (readerShouldBlock()) { //如果readerShouldBlock()返回true，表示当前线程获取读锁需要被阻塞 //结合ThreadLocal处理当前线程重入读锁的次数 //Make sure we're not acquiring read lock reentrantly if (firstReader == current) { //assert firstReaderHoldCount > 0; } else { if (rh == null) { rh = cachedHoldCounter; if (rh == null || rh.tid != getThreadId(current)) { rh = readHolds.get(); if (rh.count == 0) { readHolds.remove(); } } } //如果不是重入 if (rh.count == 0) { return -1; } } } if (sharedCount(c) == MAX_COUNT) { throw new Error("Maximum lock count exceeded"); } if (compareAndSetState(c, c + SHARED_UNIT)) { //结合ThreadLocal处理当前线程重入读锁的次数 if (sharedCount(c) == 0) { firstReader = current; firstReaderHoldCount = 1; } else if (firstReader == current) { firstReaderHoldCount++; } else { if (rh == null) { rh = cachedHoldCounter; } if (rh == null || rh.tid != getThreadId(current)) { rh = readHolds.get(); } else if (rh.count == 0) { readHolds.set(rh); } rh.count++; cachedHoldCounter = rh; // cache for release } return 1; } } } ... } ...}
5.ReentractReadWriteLock的公平锁和非公平锁
(1)公平锁的实现代码
(2)非公平锁的实现代码
(1)公平锁的实现代码
对公平锁来说，用于判断读线程在抢占锁时是否应该阻塞的readerShouldBlock()方法，以及用于判断写线程在抢占锁时是否应该阻塞的writerShouldBlock()方法，都会通过hasQueuedPredecessors()方法判断当前队列中是否有线程排队。只要队列中有其他线程在排队，写线程和读线程都要排队尾，不能抢占锁。
public class ReentrantReadWriteLock implements ReadWriteLock, java.io.Serializable { ... //Fair version of Sync static final class FairSync extends Sync { final boolean writerShouldBlock() { return hasQueuedPredecessors(); } final boolean readerShouldBlock() { return hasQueuedPredecessors(); } } //Synchronization implementation for ReentrantReadWriteLock. //Subclassed into fair and nonfair versions. abstract static class Sync extends AbstractQueuedSynchronizer { ... //Returns true if the current thread, when trying to acquire the read lock, //and otherwise eligible to do so, should block because of policy for overtaking other waiting threads. abstract boolean readerShouldBlock(); //Returns true if the current thread, when trying to acquire the write lock, //and otherwise eligible to do so, should block because of policy for overtaking other waiting threads. abstract boolean writerShouldBlock(); ... } ...}public abstract class AbstractQueuedSynchronizer extends AbstractOwnableSynchronizer implements java.io.Serializable { ... //Queries whether any threads have been waiting to acquire longer than the current thread. //判断当前队列中是否有线程排队 public final boolean hasQueuedPredecessors() { Node t = tail; // Read fields in reverse initialization order Node h = head; Node s; //所以!hasQueuedPredecessors() 等价于： //h == t || (h.next != null && h.next.thread == Thread.currentThread()) return h != t && ((s = h.next) == null || s.thread != Thread.currentThread()); } ...}(2)非公平锁的实现代码
对非公平锁来说，writerShouldBlock()方法会直接返回false，因为写线程可以去抢非公平锁的情况一定是：没有其他线程持有锁或者是线程自己重入写锁，所以不需要阻塞。而readerShouldBlock()方法会调用一个方法来决定是否阻塞读线程，在一定程度上避免发生写锁无限等待的问题(死锁饥饿问题)。
readerShouldBlock()方法调用的这个方法就是AQS的apparentlyFirstQueuedIsExclusive()方法。如果当前等待队列头结点的后继结点是写锁结点，那么该方法就返回true，表示当前来获取读锁的读线程需要排队。如果当前等待队列头结点的后继结点是读锁结点，那么该方法就返回false，表示当前来获取读锁的读线程可以抢占锁。
由于读线程和读线程是不互斥的，如果当前正在有线程持有读锁，而新来的读线程还非公平地抢读锁，可能导致写线程永远拿不到写锁。
public class ReentrantReadWriteLock implements ReadWriteLock, java.io.Serializable { ... static final class NonfairSync extends Sync { //写线程调用 final boolean writerShouldBlock() { return false; // writers can always barge } //读线程调用 final boolean readerShouldBlock() { //As a heuristic to avoid indefinite writer starvation, //block if the thread that momentarily appears to be head of queue, if one exists, is a waiting writer. //This is only a probabilistic effect since a new reader will not block //if there is a waiting writer behind other enabled readers that have not yet drained from the queue. return apparentlyFirstQueuedIsExclusive(); } } ...}public abstract class AbstractQueuedSynchronizer extends AbstractOwnableSynchronizer implements java.io.Serializable { ... //如果当前等待队列头结点的后继结点是写锁结点，那么该方法就返回true，表示当前来获取读锁的读线程需要排队； //如果当前等待队列头结点的后继结点是读锁结点，那么该方法就返回false，表示当前来获取读锁的读线程可以抢占锁； final boolean apparentlyFirstQueuedIsExclusive() { Node h, s; return (h = head) != null && (s = h.next)!= null && !s.isShared() && s.thread != null; } ...}
6.ReentrantReadWriteLock中的锁降级
(1)什么是ReentrantReadWriteLock中的锁降级
(2)ReentrantReadWriteLock中的锁降级例子
(1)什么是ReentrantReadWriteLock中的锁降级
这里的锁降级，指的是从写锁降级到读锁。也就是如果线程A获取了写锁，在写锁没有释放的情况下允许再去获取读锁。如果线程A获取了写锁，然后释放写锁，之后再获取读锁，这就不是锁降级。当然如果线程A获取了读锁，在读锁没有释放的情况下是不允许再获取写锁。
(2)ReentrantReadWriteLock中的锁降级例子
锁降级是为了提升性能。如果只使用写锁，那么长时间执行use(data)读数据时会阻塞其他读线程。所以通过将写锁降级为读锁，那么执行use(data)时也不会阻塞读线程。
Object data;public void processData() { readLock.lock();//要先获取读锁，因为后面要读数据，比如update为true时，需要 if (!update) {//发现要进行修改 readLock.unlock();//必须先释放读锁，接下来才能获取写锁 writeLock.lock();//锁降级从这里获取到写锁开始 try { if (!update) { //准备修改数据，写数据data data = ...; update = true; } readLock.lock();//在获得写锁的基础上，继续获取读锁 } finally { writeLock.unlock();//释放写锁，写锁降级为读锁，完成锁降级 } } try { //使用数据，读数据data使用时间长 use(data); } finally { readLock.unlock();//释放读锁 }}
7.Condition的说明介绍
(1)Condition接口
(2)Condition说明
(1)Condition接口
public interface Condition { //Causes the current thread to wait until it is signalled or Thread#interrupt interrupted. void await() throws InterruptedException; //Causes the current thread to wait until it is signalled. void awaitUninterruptibly(); //Causes the current thread to wait until it is signalled or interrupted, or the specified waiting time elapses. long awaitNanos(long nanosTimeout) throws InterruptedException; //Causes the current thread to wait until it is signalled or interrupted, or the specified waiting time elapses. boolean await(long time, TimeUnit unit) throws InterruptedException; //Causes the current thread to wait until it is signalled or interrupted, or the specified deadline elapses. boolean awaitUntil(Date deadline) throws InterruptedException; //Wakes up one waiting thread. void signal(); //Wakes up all waiting threads. void signalAll();}(2)Condition说明
一.Condition与wait/notify的对比
Condition的功能和wait/notify类似，可以实现等待/通知模式。wait/notify必须要和synchronized一起使用，Condition也必须要和Lock一起使用。Condition避免了wait/notify的生产者通知生产者，消费者通知消费者的问题。
二.Condition的使用
一般都会将Condition对象作为成员变量。当调用Condition的await()方法后，当前线程会释放锁并挂起等待。当其他线程线程调用Condition的signal()方法通知当前线程(释放锁)后，当前线程才会从Condition的await()方法处返回，并且返回的时候已经获得了锁。
三.两个线程在使用Condition的交互流程
线程1 -> 获取锁 -> 释放锁 + await()阻塞等待 ->
线程2 -> 获取锁 -> signal()唤醒线程1 + 释放锁 ->
线程1 -> 被唤醒 + 尝试获取锁 -> 释放锁
四.读写锁和独占锁是否支持Condition
独占锁和读写锁的写锁都支持Condition，但是读写锁的读锁是不支持Condition的。
ReentrantLock支持Condition。
ReentrantReadWriteLock的WriteLock支持Condition。
五.由AQS的内部类ConditionObject实现
每个Condition对象都有一个Condition队列，这个队列是Condition对象实现等待/通知功能的关键。
六.Condition的应用场景
LinkedBlockingQueue、ArrayBlockQueue、CyclicBarrier都用到了Condition来实现线程等待。
public class ConditionDemo() { static ReentrantLock lock = new ReentrantLock(); static Condition condition = lock.newCondition(); public static void main(String[] args) throws Exception { new Thread() { public void run() { lock.lock(); System.out.println("第一个线程加锁"); try { System.out.println("第一个线程释放锁以及阻塞等待"); condition.await(); System.out.println("第一个线程重新获取锁"); } catch (Exception e) { e.printStackTrace(); } System.out.println("第一个线程释放锁"); lock.unlock(); }; }.start(); Thread.sleep(3000); new Thread() { public void run() { lock.lock(); System.out.println("第二个线程加锁"); System.out.println("第二个线程唤醒第一个线程"); condition.signal(); lock.unlock(); System.out.println("第二个线程释放锁"); }; }.start(); }}
8.Condition的源码实现
(1)创建ConditionObject对象
(2)ConditionObject的Condition队列
(3)ConditionObject的等待方法await()
(4)ConditionObject的通知方法signal()
(1)创建ConditionObject对象
调用ReentrantLock的newCondition()方法可以创建一个Condition对象，ReentrantLock的newCondition()方法会调用Sync的newCondition()方法，而Sync的newCondition()方法会创建ConditionObject对象。
public class ReentrantLock implements Lock, java.io.Serializable { ... //Synchronizer providing all implementation mechanics private final Sync sync; //Returns a {@link Condition} instance for use with this Lock instance. public Condition newCondition() { //执行ReentrantLock内部类Sync的newCondition()方法 return sync.newCondition(); } //Base of synchronization control for this lock. //Subclassed into fair and nonfair versions below. //Uses AQS state to represent the number of holds on the lock. abstract static class Sync extends AbstractQueuedSynchronizer { ... final ConditionObject newCondition() { return new ConditionObject(); } ... } ...} public abstract class AbstractQueuedSynchronizer extends AbstractOwnableSynchronizer implements java.io.Serializable { ... public class ConditionObject implements Condition, java.io.Serializable { //First node of condition queue. private transient Node firstWaiter; //Last node of condition queue. private transient Node lastWaiter; //Creates a new ConditionObject instance. public ConditionObject() { } ... } ...}(2)ConditionObject的Condition队列
在ConditionObject对象中，会有一个Condition队列，ConditionObject对象维护的Condition队列是一个单向链表。
ConditionObject对象的属性firstWaiter和lastWaiter代表队列的头尾结点。当线程调用await()方法时，该线程就会被封装成Node结点。然后该Node结点就会被作为新增结点加入到Condition队列的尾部。
由于Condition对象拥有Condition队列的头尾结点的引用，所以只需将原来尾结点的nextWaiter指向新增结点，并更新尾结点即可。这个更新过程无须使用CAS保证，因为调用await()方法的线程已经获取锁了。
(3)ConditionObject的等待方法await()
ConditionObject的await()方法的主要逻辑：
一.通过addConditionWaiter()方法将当前线程添加到Condition队列
二.通过fullyRelease()方法释放锁
三.通过LockSupport.park()方法挂起当前线程
四.被signal()方法唤醒后，通过acquireQueued()方法尝试获取锁
其实相当于将等待队列的头结点(获取了锁的结点)移动到Condition队列中。但等待队列的头结点不会直接加入Condition队列，而是会把当前线程封装成一个新的Node结点加入到Condition队列尾部。
注意：调用await()方法的线程其实已成功获取锁，该线程对应等待队列的头结点。await()方法会将当前线程封装成结点并加入到Condition队列，然后释放锁，并唤醒等待队列头结点的后继结点对应线程，再挂起当前线程进入等待状态。当该线程被signal()方法唤醒后，便会通过acquireQueued()方法尝试获取锁。所以调用await()方法的线程在阻塞后被唤醒，也有可能获取锁失败继续阻塞。
Condition.await()原理总结：将自己加入Condition队列、释放锁、挂起自己。
public abstract class AbstractQueuedSynchronizer extends AbstractOwnableSynchronizer implements java.io.Serializable { ... public class ConditionObject implements Condition, java.io.Serializable { //First node of condition queue. private transient Node firstWaiter; //Last node of condition queue. private transient Node lastWaiter; ... //Implements interruptible condition wait. //If current thread is interrupted, throw InterruptedException. public final void await() throws InterruptedException { if (Thread.interrupted()) { throw new InterruptedException(); } //1.执行ConditionObject的addConditionWaiter()方法，将当前线程封装成Node结点并加入Condition队列 Node node = addConditionWaiter(); //2.调用AQS的fullyRelease()方法释放锁 int savedState = fullyRelease(node); int interruptMode = 0; while (!isOnSyncQueue(node)) { //3.阻塞当前线程 LockSupport.park(this); if ((interruptMode = checkInterruptWhileWaiting(node)) != 0) { break; } } //4.当前线程被signal()方法唤醒后，执行AQS的acquireQueued()方法尝试获取锁 if (acquireQueued(node, savedState) && interruptMode != THROW_IE) { interruptMode = REINTERRUPT; } if (node.nextWaiter != null) {// clean up if cancelled unlinkCancelledWaiters(); } if (interruptMode != 0) { reportInterruptAfterWait(interruptMode); } } //将当前线程封装成Node结点并加入Condition队列 private Node addConditionWaiter() { Node t = lastWaiter; //If lastWaiter is cancelled, clean out. if (t != null && t.waitStatus != Node.CONDITION) { unlinkCancelledWaiters(); t = lastWaiter; } Node node = new Node(Thread.currentThread(), Node.CONDITION); if (t == null) { firstWaiter = node; } else { t.nextWaiter = node; } lastWaiter = node; return node; } ... } //释放锁 //Invokes release with current state value; returns saved state. //Cancels node and throws exception on failure. final int fullyRelease(Node node) { boolean failed = true; try { int savedState = getState(); if (release(savedState)) { failed = false; return savedState; } else { throw new IllegalMonitorStateException(); } } finally { if (failed) { node.waitStatus = Node.CANCELLED; } } } //Releases in exclusive mode. //Implemented by unblocking one or more threads if #tryRelease returns true. //This method can be used to implement method 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; } //Acquires in exclusive uninterruptible mode for thread already in queue. //Used by condition wait methods as well as acquire. 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; } //执行shouldParkAfterFailedAcquire()方法设置node结点的前驱结点的状态为SIGNAL //执行parkAndCheckInterrupt()方法挂起当前线程 if (shouldParkAfterFailedAcquire(p, node) && parkAndCheckInterrupt()) { interrupted = true; } } } finally { if (failed) { cancelAcquire(node); } } } ...}(4)ConditionObject的通知方法signal()
ConditionObject的signal()方法的主要逻辑：
一.首先从Condition队列中取出等待时间最长的结点，也就是first结点
二.然后将等待时间最长的结点(first结点)转移到AQS的等待队列(CLH队列)中
三.最后唤醒该first结点对应的线程
由于该first结点对应的线程在await()方法中加入Condition队列后被阻塞，所以该first结点对应的线程在被唤醒后，会回到await()方法中继续执行，也就是会执行AQS的acquireQueued()方法去尝试获取锁。
调用signal()方法的前提条件是当前线程必须获得了锁，所以signal()方法首先会检查当前线程是否获取了锁，接着才去获取Condition队列的first结点，然后才将first结点移动到等待队列，并唤醒该first结点对应的线程。
通过调用AQS的enq()方法，Condition队列的first结点会添加到等待队列。当first结点被移动到等待队列后，再唤醒first结点对应的线程尝试获取锁。
被唤醒的first结点对应的线程，将从await()方法中的while循环退出。因为已经在等待队列中，所以isOnSyncQueue()方法会返回true，从而会调用AQS的acquireQueued()方法来竞争获取锁。
Condition.signal()原理总结：把Condition队列中的头结点，转化为等待队列中的尾结点。
public abstract class AbstractQueuedSynchronizer extends AbstractOwnableSynchronizer implements java.io.Serializable { ... public class ConditionObject implements Condition, java.io.Serializable { //First node of condition queue. private transient Node firstWaiter; //Last node of condition queue. private transient Node lastWaiter; ... //Moves the longest-waiting thread, if one exists, //from the wait queue for this condition to the wait queue for the owning lock. public final void signal() { //通过isHeldExclusively()方法检查当前线程是否获取了锁 if (!isHeldExclusively()) { throw new IllegalMonitorStateException(); } Node first = firstWaiter; if (first != null) { doSignal(first); } } //Removes and transfers nodes until hit non-cancelled one or null. //Split out from signal in part to encourage compilers to inline the case of no waiters. private void doSignal(Node first) { do { //firstWaiter后移 if ((firstWaiter = first.nextWaiter) == null) { lastWaiter = null; } //firstWaiter出队 first.nextWaiter = null; } while (!transferForSignal(first) && (first = firstWaiter) != null); } //Transfers a node from a condition queue onto sync queue. //Returns true if successful. final boolean transferForSignal(Node node) { //在addConditionWaiter()方法时，node就被封装为CONDITION类型 //如果CAS失败，说明当前结点已被修改为CANCELED，此时需要继续查找Condition队列的下一个结点 if (!compareAndSetWaitStatus(node, Node.CONDITION, 0)) { return false; } //将node结点转移到等待队列中，返回等待队列的尾结点 Node p = enq(node); int ws = p.waitStatus; if (ws > 0 || !compareAndSetWaitStatus(p, ws, Node.SIGNAL)) { //唤醒node结点 LockSupport.unpark(node.thread); } return true; } //Implements interruptible condition wait. //If current thread is interrupted, throw InterruptedException. public final void await() throws InterruptedException { if (Thread.interrupted()) { throw new InterruptedException(); } //1.执行ConditionObject的addConditionWaiter()方法，将当前线程封装成Node结点并加入Condition队列 Node node = addConditionWaiter(); //2.调用AQS的fullyRelease()方法释放锁 int savedState = fullyRelease(node); int interruptMode = 0; //一开始node结点并不在等待队列中，所以isOnSyncQueue()方法返回false，进行阻塞 //后来其他线程调用signal()方法，该node结点就会被唤醒，然后发现该node结点已经在等待队列中，于是isOnSyncQueue()方法返回true while (!isOnSyncQueue(node)) { //3.阻塞当前线程 LockSupport.park(this); if ((interruptMode = checkInterruptWhileWaiting(node)) != 0) { break; } } //4.当前线程被signal()方法唤醒后，执行AQS的acquireQueued()方法尝试获取锁 if (acquireQueued(node, savedState) && interruptMode != THROW_IE) { interruptMode = REINTERRUPT; } if (node.nextWaiter != null) {// clean up if cancelled unlinkCancelledWaiters(); } if (interruptMode != 0) { reportInterruptAfterWait(interruptMode); } } ... } //Inserts node into queue, initializing if necessary. See picture above. 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; } } } } ...}
页:
[1]