JUC-CountDownLatch基础篇
JUC-CountDownLatch源码分析
JUC-Semaphore基础篇
JUC-Semaphore源码分析
JUC-ReentrantReadWriteLock锁基础篇
JUC-ReentrantReadWriteLock锁源码分析
JUC-ReentrantLock锁基础篇
JUC-ReentrantLock锁源码分析
JUC-CyclicBarrier基础篇
JUC-CyclicBarrier源码分析
文章目录
- 1.Condition的概述及其使用
- 2.await()方法源码
-
- 2.1 addConditionWaiter()方法源码
-
- 2.1.1 unlinkCancelledWaiters()方法源码
- 2.2 fullyRelease(Node node)方法源码
- 2.3 isOnSyncQueue(Node node)方法源码
-
- 2.3.1 findNodeFromTail(Node node)方法源码
- 3.signal()方法源码
-
- 3.1 doSignal(Node first)方法源码
-
- 3.1.1 transferForSignal(Node node)方法源码
1.Condition的概述及其使用
我们知道在AQS源码中用lock和unlock可以用来解决并发中的互斥问题。那么AQS的Condition是干啥的呢？Condition主要是用来解决线程之间的同步问题，类似于实现了线程之前的通信。当某个条件满足时，执行某个线程自己的操作。当条件不满足时，调用await方法将当前线程放入到Condition的条件队列中挂起。当这个条件满足了，其他的线程调用signal方法，将挂起的线程从条件队列中转移到同步队列中，让这个线程唤醒。Condition的await方法、signal方法和signalAll方法主要对应的是内置锁synchronized中配套的wait方法、notify方法和notifAll方法，不过Condition可以更加细化的控制锁的唤醒条件。
它们之间的对比：
- 同步：内置锁synchronized的wait()必须要在synchronize代码块中才能调用，也就是必须获取到监视器锁之后才能执行。而Condition的await()也是必须获取到锁之后才可以执行。
- 等待: 调用内置锁synchronized的wait()的线程会释放已经获取到的锁，进入到当前监视器锁的等待队列中。而Condition的await()也会释放当前线程获取到的锁进入到条件队列中
- 唤醒: 调用notify()会唤醒等待在该监视器锁的线程，并参与监视器锁的竞争，并在获取锁之后从wait()处恢复执行。调用signal()会将当前线程从条件队列转移到同步队列中，并重新参与锁的竞争并在获取锁之后继续执行await()方法里面剩余的代码。
注意：在使用Condition的await和signal方法之前必须先获取锁，调用完之后要释放锁，这里的锁是独占锁，也就是说Condition只能与独占锁搭配使用。
lock.lock();
try {
//使用await()或者signal()时必须先获取到ReentrantLock的锁
await();
//signal()
//signalAll()
} finally {
lock.unlock();
}
具体await和signal的举例可以参考条件变量 Condition详解
2.await()方法源码
public final void await() throws InterruptedException {
//若当前线程已经中断则抛出中断异常
if (Thread.interrupted())
throw new InterruptedException();
//将当前线程加入到条件队列中去
Node node = addConditionWaiter();
//释放当前线程所占用的锁，并保存当前锁的状态
int savedState = fullyRelease(node);
int interruptMode = 0;
while (!isOnSyncQueue(node)) {
//只有调用signal方法才能将条件队列中的节点放到同步队列中去
//如果当前node节点不在同步队列中，说明没有调用signal方法那就park阻塞它
LockSupport.park(this);//这一行代码执行，当前线程就阻塞住了
//执行到这一步说明当前线程被唤醒了有可能是线程中断被唤醒
//也有可能是调用了signal方法被唤醒
if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
//这是线程被中断的情况，下面进行详解
break;
}
//走到这一步说明节点node此时一定在同步队列当中了
//当线程被唤醒后重新进行锁的竞争，竞争到就接着执行，竞争不到就在同步队列中阻塞
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加入到条件队列中去
- 释放当前线程所占用的锁，并保存当前锁的状态
- 循环阻塞线程
- 被唤醒后接着acquireQueued方法(此方法源码可以看JUC-AQS源码篇)来竞争锁
举个场景：
现在我们知道在await方法内部先是释放锁，再获取锁。这就与使用await方法之前必须先获取锁，使用完之后必须释放锁对应起来了。
整个流程就是先调用lock.lock()获取锁->然后await方法内部释放这把锁->然后阻塞住->然后被唤醒->然后去获取锁，获取不到接着阻塞直到获取到->然后调用lock.unlock释放这把获取到的锁。
问题一：为什么要调用fullyRelease方法释放锁？为什么叫fullyRelease名字？
前面我们知道调用await方法之前必须先获取独占锁lock.lock()，调用signal方法也要先获取独占锁，如果你调用await阻塞之前，不先把独占锁释放掉了，那么signal方法永远也无法获取锁，那么你自己岂不是永远无法被唤醒了。那为啥函数名却叫fullyRelease，因为我们知道，对于可重入锁来说(比如ReentrantLock)，可以重入多次，所以我们得把当前线程获取到的独占锁一次性释放完，所以才叫fullyRelease。同时fullyRelease方法返回了当前线程之前调用lock.lock()方法获取独占锁时所竞争到的锁资源个数。为什么要返回呢？因为下面的acquireQueued方法是需要重新竞争独占锁的，那么我当然要知道我之前竞争独占锁时需要竞争多少个资源。不然，我这个acquireQueued方法竞争独占锁时，该要竞争多少个资源呢。
问题二：isOnSyncQueue为什么要在阻塞之前判断一下node节点在不在同步队列上呢？
这是因为可能其它线程获取到了锁，导致并发调用了signal方法或者signalAll方法导致已经把node节点从条件队列转移到同步队列上面了，那么既然已经发出signal信号了，那么node节点就不需要执行 LockSupport.park(this)代码了。
问题三：当执行了LockSupport.park(this)让当前线程阻塞了，有那么情况会让当前线程被唤醒呢？
情况一：当前线程被中断了。
情况二：当其他线程执行signal时把node节点放入到同步队列中发现它的前驱节点被取消了或者通过CAS的方式将它的前驱节点的waitStatus设置成-1失败了，会调用LockSupport.unpark(node.thread)把它唤醒。
情况三：当其他线程执行signal时把node节点成功放入到同步队列。此node节点在同步队列中被其他的线程正常的唤醒。
再来讲讲await方法中中断的逻辑
首先我们知道当前线程被唤醒大体上可以分为两类一类是中断唤醒一类是其他线程调用signal方法唤醒的。如果是signal唤醒那么node节点就已经在同步队列上面了，如果是中断唤醒，那么后面的checkInterruptWhileWaiting也保证了会将node节点放入到同步队列当中去。所有无论是哪种方式的唤醒，最终线程都会从条件队列到同步队列中去，并且利用acquireQueued()方法进行阻塞式的竞争锁，抢不到就挂起。所以当await()方法返回时，必然是保证当前线程是已经获取到锁了的。
那么如果从线程被唤醒到利用acquireQueued()方法来竞争锁的这段时间发生了中断该如何处理？await是调用checkInterruptWhileWaiting(node)方法来判断的
checkInterruptWhileWaiting(node)方法
private int checkInterruptWhileWaiting(Node node) {
return Thread.interrupted() ?
(transferAfterCancelledWait(node) ? THROW_IE : REINTERRUPT) :
0;
}
final boolean transferAfterCancelledWait(Node node) {
if (compareAndSetWaitStatus(node, Node.CONDITION, 0)) {
enq(node);
return true;
}
while (!isOnSyncQueue(node))
Thread.yield();
return false;
}
首先调用Thread.interrupted()发现没中断那么就直接返回0，如果有中断就调用transferAfterCancelledWait方法。
transferAfterCancelledWait方法中，首选判断compareAndSetWaitStatus(node, Node.CONDITION, 0)的CAS操作是否成功。如果此CAS操作成功了说明node节点还没有被single(因为如果node节点被single了，那么它的waitStatus的值就不是Node.CONDITION，此CAS操作会失败)，那么调用enq(node)并返回true，所以checkInterruptWhileWaiting方法返回的是THROW_IE。如果此CAS失败了，说明此node节点被single了，checkInterruptWhileWaiting方法返回REINTERRUPT。
所以await方法中如果线程发生了中断，可以分两种情况
- 中断发生时，线程还未被signal过
- 中断发生时，线程已经被signal过
对应的await针对这两种情况的处理方式：
首先如果发生中断了
while (!isOnSyncQueue(node)) {
LockSupport.park(this);
if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
break;
}
直接break退出while循环。
private void reportInterruptAfterWait(int interruptMode)
throws InterruptedException {
if (interruptMode == THROW_IE)
throw new InterruptedException();
else if (interruptMode == REINTERRUPT)
selfInterrupt();
}
方式一：如果中断发生时，线程还未被signal过。那么interruptMode值是THROW_IE，那么reportInterruptAfterWait方法执行的结果是抛出一个InterruptedException，因为线程是非正常唤醒，而是被中断的，需要抛出InterruptedException。
在这里需要注意一点是，如果中断发生时，线程还未被signal过，transferAfterCancelledWait方法执行代码的是
if (compareAndSetWaitStatus(node, Node.CONDITION, 0)) {
enq(node);
return true;
}
这三行代码，这里面的enq(node)方法只是将node节点放到同步队列当中去的，并没有将node节点的nextWaiter的指针断开的，并不像调用single方法那样会将nextWaiter指针赋值为null。所以此时node节点很尴尬，因为它既在同步队列上面找的到它，又可以条件队列中找的到它。所以await()方法中的
if (node.nextWaiter != null) // clean up if cancelled
unlinkCancelledWaiters();
这一部分代码就来解决这个问题。如果发现node.nextWaiter不等于null就说明是中断唤醒导致此node节点到同步队列上面的，那么我们需要调用unlinkCancelledWaiters方法将此node节点从条件队列中移出。如果等于null就说明是调用single导致此node节点到同步队列上面的，那么就不用干啥了。
方式二：中断发生时，线程已经被signal过
如果是方式二，interruptMode值是REINTERRUPT，那么reportInterruptAfterWait方法执行的结果是调用selfInterrupt()方法，理由是此时线程已经被signal过了，说明这个中断来的太晚了，我已经被唤醒了你的中断指令才到，我没必要理你，直接忽略。我只需要在await()方法结束后自行中断下，补下这中断状态即可。
static void selfInterrupt() {
Thread.currentThread().interrupt();
}
注意一下： 如果是中断发生时，线程已经被signal过，那么transferAfterCancelledWait方法执行代码的是
while (!isOnSyncQueue(node))
Thread.yield();
return false;
这个while循环是几个意思？我们知道能执行这个三行代码一定是前面的compareAndSetWaitStatus(node, Node.CONDITION, 0)CAS操作失败了。而这个CAS操作失败了，只能说明已经有其他的线程发出了single信号了。此node节点正在加入同步队列当中。也就是说当执行这个while循环时，当前node节点也可能是在进入同步队列的路上。
- 假设有线程A、线程B是并发执行的
- 线程A被唤醒后检测到发生了中断，所以走到了transferAfterCancelledWait
- 而线程B在这之前已经调用了signal方法，该方法会调用transferForSignal将当前线程加入到同步队列队尾
- 这里我们分析的是中断在signal之后，所以此时线程B的compareAndSetWaitStatus(node, Node.CONDITION, 0)会优先于线程A执行
- 所以这里可能会出现B已经修改了node的waitStatus状态，但还未来得及调用enq方法，线程A就执行了transferAfterCancelledWait， 此时发现waitStatus已经不是Condition了，但其实自己还没有添加到同步队列中去。所以就会执行while循环通过isOnSyncQueue方法判断node是不是已经在同步队列上了，如果不在就调用Thread.yield() 一下等待线程B执行完transferForSignal方法将node节点加入到同步队列当中去。
注意一下： interruptMode值是REINTERRUPT还可能是调用acquireQueued方法导致的，也就是说调用acquireQueued方法之前没有发生中断，执行acquireQueued方法内部却发生了中断。此时await()处理这个中断的行为与中断发生时，线程已经被signal过这个中断的行为一致。都是调一下selfInterrupt方法。
if (acquireQueued(node, savedState) && interruptMode != THROW_IE)
interruptMode = REINTERRUPT;
我们知道调用acquireQueued方法时，如果执行acquireQueued方法内部代码时发生了线程中断，那么acquireQueued方法会返回true。当acquireQueued方法返回true时，只要之前的interruptMode 不是THROW_IE，也就是说之前的代码如果没有发生中断或者如果发生了中断，但是中断发生时，线程已经被signal过。那么interruptMode 值统统设置成REINTERRUPT，因为REINTERRUPT代表的await()处理中断的行为是调用一下selfInterrupt()，补下当前线程的中断状态即可。因为执行acquireQueued方法内部代码时发生中断，只需要补下当前线程的中断状态即可，执行acquireQueued方法之前，中断发生时，线程已经被signal过，也只需要补下当前线程的中断状态即可。只有中断发生时，线程还未被signal过，才需要抛出中断异常。
2.1 addConditionWaiter()方法源码
private Node addConditionWaiter() {
Node t = lastWaiter;
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;
}
此方法主要是将当前线程封装成Node节点加入到条件队列中去，加入的过程中如果发现lastWaiter指针指向的节点被取消掉了，就调用unlinkCancelledWaiters方法剔除队列中状态不为CONDITION的节点。
注意：在条件队列中Node节点的waitStatus 取值只有-2代表此节点是条件队列的节点和1代表此节点被取消掉了。
2.1.1 unlinkCancelledWaiters()方法源码
private void unlinkCancelledWaiters() {
Node t = firstWaiter;
Node trail = null;
while (t != null) {
Node next = t.nextWaiter;
if (t.waitStatus != Node.CONDITION) {
t.nextWaiter = null;
if (trail == null)
firstWaiter = next;
else
trail.nextWaiter = next;
if (next == null)
lastWaiter = trail;
}
else
trail = t;
t = next;
}
}
此方法主要是把条件队列中取消掉的节点清除掉，只要了解链表的话可以非常轻松的读懂这段代码，小编就不过多讲解了。
2.2 fullyRelease(Node node)方法源码
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;
}
}
此方法把当前线程所持有的独占锁一次性全都释放掉，因为可能有锁重入的情况，所以要全部释放锁。同时把之前获取到的锁资源个数返回，以便在await方法中执行acquireQueued重新去竞争锁时用到。
注意：如果在释放锁的过程中出现错误了，就将该节点设置成取消节点状态了。failed=true的情况有执行release方法抛出异常或者release释放锁没有成功的情况，此时改node节点就会被取消掉。
这个地方还有一点需要注意下，就是调用release方法是可能会抛出IllegalMonitorStateException异常，这是因为当前线程可能并不是持有锁的线程。
可之前不是说调用await方法的线程一定是持有锁的嘛，虽然理论上是这样的，但其实await方法是并没有校验的，而是放到了fullyRelease()的tryRelease()来校验Thread.currentThread() == getExclusiveOwnerThread()。release(int arg)方法可以参考JUC-AQS源码篇
2.3 isOnSyncQueue(Node node)方法源码
final boolean isOnSyncQueue(Node node) {
if (node.waitStatus == Node.CONDITION || node.prev == null)
return false;
if (node.next != null) // If has successor, it must be on queue
return true;
return findNodeFromTail(node);
}
此方法主要是判断node节点在不在同步队列上面。
我们知道single方法将node节点从条件队列转移到同步队列上面时，会先将node的waitStatus 从Node.CONDITION变成0，再执行enq(node)方法将node节点加入到同步队列中。而enq(node)方法中是先执行node.prev = t将node节点的前驱指针指向同步队列的原来尾节点，再执行compareAndSetTail(t, node)将node设置成同步队列新的尾结点，然后执行t.next = node将前尾结点的next指针指向node节点。通过这个过程我们知道
1、 如果if(node.waitStatus==Node.CONDITION||node.prev==null)成立那么此node节点一定不在同步队列上面；
2、 因为条件队列中的node节点的next指针是没有值的，如果有值，那么也是node节点已经在同步队列上面了并且它已经有了后继节点如果if(node.next!=null)成立，那么此node一定在同步节点上面；
3、 如果前面的两个if都不成立的话，也不能说此node节点一定不在同步队列上面，因为可能执行enq(node)方法当中，已经执行了node.prev=t，但是compareAndSetTail(t,node)这个CAS操作失败了，需要再重新来过才能成功那么可能导致前面的两个if判断都不成功了那么我们需要调用findNodeFromTail方法从tail尾部向前遍历来判断此node在不在同步队列当中；
2.3.1 findNodeFromTail(Node node)方法源码
private boolean findNodeFromTail(Node node) {
Node t = tail;
for (;;) {
if (t == node)
return true;
if (t == null)
return false;
t = t.prev;
}
}
从tail尾部向前遍历的方法判断node节点在不在同步队列当中。
3.signal()方法源码
public final void signal() {
if (!isHeldExclusively())
throw new IllegalMonitorStateException();
Node first = firstWaiter;
if (first != null)
doSignal(first);
}
首先，先调用isHeldExclusively()方法来判断调用signal方法的线程是不是以及获取了锁。然后取条件队列中第一个节点firstWaiter，将它从条件队列转移到同步队列当中。
3.1 doSignal(Node first)方法源码
private void doSignal(Node first) {
do {
if ( (firstWaiter = first.nextWaiter) == null)
lastWaiter = null;
first.nextWaiter = null;
} while (!transferForSignal(first) &&
(first = firstWaiter) != null);
}
通过transferForSignal方法将first节点从条件队列转移到同步队列当中，然后将first节点从条件队列当中踢出去。如果失败了那么就将first节点的nextWaiter当做新的first节点重复以上的操作。
3.1.1 transferForSignal(Node node)方法源码
final boolean transferForSignal(Node node) {
if (!compareAndSetWaitStatus(node, Node.CONDITION, 0))
return false;
Node p = enq(node);
int ws = p.waitStatus;
if (ws > 0 || !compareAndSetWaitStatus(p, ws, Node.SIGNAL))
LockSupport.unpark(node.thread);
return true;
}
此方法是将node节点加入到同步队列当中的。
如果compareAndSetWaitStatus(node, Node.CONDITION, 0)CAS操作失败了，说明这个node节点已经被取消了。因为条件队列中node节点的WaitStatus值只有-2和1.
然后执行enq(node)将node节点加入到同步队列当中，并且返回它的前驱节点p。如果p的waitStatus值大于0，说明这个前驱节点p已经被取消了，所有我们必须调用 LockSupport.unpark(node.thread)然node这个节点的线程重新唤醒，然后接着执行await()中的acquireQueued(node, savedState)让同步队列把这个取消掉的前驱节点踢出同步队列中，好让node节点有一个新的前驱节点。不然这个node节点岂不是在同步队列中永远不会被唤醒了。还有执行compareAndSetWaitStatus(p, ws, Node.SIGNAL)CAS操作失败了，这CAS操作失败了，说明node的前驱节点的WaitStatus值在执行CAS操作期间被改变了，那么我们不知道是什么原因导致改变的，会不会影响到node节点正常的唤醒操作，那么我们就调用一下LockSupport.unpark(node.thread)让node节点唤醒，以重新在同步队列中同步一下，即使这个唤醒操作是多余的也无妨。