JUC并发—7.AQS源码分析三 - ToB企服应用市场:ToB评测及商务社交产业平台

public class CountDownLatchDemo {
public static void main(String[] args) throws Exception {
final CountDownLatch latch = new CountDownLatch(2);
new Thread() {
public void run() {
try {
Thread.sleep(1000);
System.out.println("线程1开始执行，休眠2秒...");
Thread.sleep(1000);
System.out.println("线程1准备执行countDown操作...");
latch.countDown();
System.out.println("线程1完成执行countDown操作...");
} catch (Exception e) {
e.printStackTrace();
}
}
}.start();
new Thread() {
public void run() {
try {
Thread.sleep(1000);
System.out.println("线程2开始执行，休眠2秒...");
Thread.sleep(1000);
System.out.println("线程2准备执行countDown操作...");
latch.countDown();
System.out.println("线程2完成执行countDown操作...");
} catch (Exception e) {
e.printStackTrace();
}
}
}.start();
System.out.println("main线程准备执行countDownLatch的await操作，将会同步阻塞等待...");
latch.await();
System.out.println("所有线程都完成countDown操作，结束同步阻塞等待...");
}
}

复制代码

//A synchronization aid that allows one or more threads to wait until a set of operations being performed in other threads completes.
public class CountDownLatch {
private final Sync sync;
public CountDownLatch(int count) {
if (count < 0) {
throw new IllegalArgumentException("count < 0");
}
this.sync = new Sync(count);
}
//Synchronization control For CountDownLatch.
//Uses AQS state to represent count.
private static final class Sync extends AbstractQueuedSynchronizer {
Sync(int count) {
setState(count);
}
int getCount() {
return getState();
}
protected int tryAcquireShared(int acquires) {
return (getState() == 0) ? 1 : -1;
}
protected boolean tryReleaseShared(int releases) {
//Decrement count; signal when transition to zero
for (;;) {
int c = getState();
if (c == 0) {
return false;
}
int nextc = c-1;
if (compareAndSetState(c, nextc)) {
return nextc == 0;
}
}
}
}
//Causes the current thread to wait until the latch has counted down to zero,
//unless the thread is Thread#interrupt interrupted.
public void await() throws InterruptedException {
//执行AQS的acquireSharedInterruptibly()方法
sync.acquireSharedInterruptibly(1);
}
...
}
public abstract class AbstractQueuedSynchronizer extends AbstractOwnableSynchronizer implements java.io.Serializable {
...
//Acquires in shared mode, aborting if interrupted.
//Implemented by first checking interrupt status, then invoking at least once #tryAcquireShared, returning on success.
//Otherwise the thread is queued, possibly repeatedly blocking and unblocking,
//invoking #tryAcquireShared until success or the thread is interrupted.
public final void acquireSharedInterruptibly(int arg) throws InterruptedException {
if (Thread.interrupted()) {
throw new InterruptedException();
}
//执行CountDownLatch的内部类Sync实现的tryAcquireShared()方法，抢占共享锁
if (tryAcquireShared(arg) < 0) {
//执行AQS的doAcquireSharedInterruptibly()方法
doAcquireSharedInterruptibly(arg);
}
}
//Acquires in shared interruptible mode.
private void doAcquireSharedInterruptibly(int arg) throws InterruptedException {
final Node node = addWaiter(Node.SHARED);//封装当前线程为Shared类型的Node结点
boolean failed = true;
try {
//第一次循环r = -1，所以会执行AQS的shouldParkAfterFailedAcquire()方法
//将node结点的有效前驱结点的状态设置为SIGNAL
for (;;) {
final Node p = node.predecessor();//node结点的前驱结点
if (p == head) {
int r = tryAcquireShared(arg);
if (r >= 0) {
setHeadAndPropagate(node, r);
p.next = null; // help GC
failed = false;
return;
}
}
//执行shouldParkAfterFailedAcquire()方法设置node结点的前驱结点的状态为SIGNAL
//执行parkAndCheckInterrupt()方法挂起当前线程
if (shouldParkAfterFailedAcquire(p, node) && parkAndCheckInterrupt()) {
throw new InterruptedException();
}
}
} 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.
private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
int ws = pred.waitStatus;
if (ws == Node.SIGNAL) {
//This node has already set status asking a release to signal it, so it can safely park.
return true;
}
if (ws > 0) {
//Predecessor was cancelled. Skip over predecessors and indicate retry.
do {
node.prev = pred = pred.prev;
} while (pred.waitStatus > 0);
pred.next = node;
} else {
//waitStatus must be 0 or PROPAGATE.
//Indicate that we need a signal, but don't park yet.
//Caller will need to retry to make sure it cannot acquire before parking.
compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
}
return false;
}
//设置头结点和唤醒后续线程
//Sets head of queue, and checks if successor may be waiting in shared mode,
//if so propagating if either propagate > 0 or PROPAGATE status was set.
private void setHeadAndPropagate(Node node, int propagate) {
Node h = head;
setHead(node);//将node结点设置为头结点
if (propagate > 0 || h == null || h.waitStatus < 0 || (h = head) == null || h.waitStatus < 0) {
Node s = node.next;
if (s == null || s.isShared()) {
doReleaseShared();
}
}
}
private void setHead(Node node) {
head = node;
node.thread = null;
node.prev = null;
}
...
}

复制代码

//A synchronization aid that allows one or more threads to wait until a set of operations being performed in other threads completes.
public class CountDownLatch {
private final Sync sync;
public CountDownLatch(int count) {
if (count < 0) {
throw new IllegalArgumentException("count < 0");
}
this.sync = new Sync(count);
}
//Synchronization control For CountDownLatch.
//Uses AQS state to represent count.
private static final class Sync extends AbstractQueuedSynchronizer {
Sync(int count) {
setState(count);
}
int getCount() {
return getState();
}
protected int tryAcquireShared(int acquires) {
return (getState() == 0) ? 1 : -1;
}
protected boolean tryReleaseShared(int releases) {
//Decrement count; signal when transition to zero
for (;;) {
int c = getState();
if (c == 0) {
return false;
}
int nextc = c-1;
if (compareAndSetState(c, nextc)) {
return nextc == 0;
}
}
}
}
//Decrements the count of the latch, releasing all waiting threads if the count reaches zero.
public void countDown() {
//执行AQS的releaseShared()方法
sync.releaseShared(1);
}
...
}
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) {
//执行CountDownLatch的内部类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() {
for (;;) {
//每次循环时头结点都会发生变化
//因为调用unparkSuccessor()方法会唤醒doAcquireSharedInterruptibly()方法中阻塞的线程
//然后阻塞的线程会在执行setHeadAndPropagate()方法时通过setHead()修改头结点
Node h = head;//获取最新的头结点
if (h != null && h != tail) {//等待队列中存在挂起线程的结点
int ws = h.waitStatus;
if (ws == Node.SIGNAL) {//头结点的状态正常，表示对应的线程可以被唤醒
if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0)) {
continue;//loop to recheck cases
}
//唤醒头结点的后继结点
//唤醒的线程会在doAcquireSharedInterruptibly()方法中执行setHeadAndPropagate()方法修改头结点
unparkSuccessor(h);
} else if (ws == 0 && !compareAndSetWaitStatus(h, 0, Node.PROPAGATE)) {
//如果ws = 0表示初始状态，则修改结点为PROPAGATE状态
continue;//loop on failed CAS
}
}
if (h == head) {//判断头结点是否有变化
break;//loop if head changed
}
}
}
//Wakes up node's successor, if one exists.
private void unparkSuccessor(Node node) {
int ws = node.waitStatus;
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);
}
}
...
}

复制代码