JUC之CountDownLatch的源码和使用场景分析
JUC之CountDownLatch的源码和使用场景分析
最近工作不饱和，写写文章充充电。何以解忧，唯有Coding。后续更新的文章涉及的方向有:ThreadPoolExecutor、Spring、MyBatis、ReentrantLock、CyclicBarrier、Semaphore.
同系列文章：
saoqi.png
开始讲解之前，自定义ThreadPoolExecutor和Task。
public class CustomThreadPoolExecutor extends ThreadPoolExecutor {
public CustomThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue) {
super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue);
}
public static class CustomTask<V> extends FutureTask<V> {
public CustomTask(Callable<V> callable) {
super(callable);
}
public CustomTask(Runnable runnable, V result) {
super(runnable, result);
}
}
}- 为什么要需要线程池 ？ 线程资源必须通过线程池提供，不允许在应用中自行显式的创建线程。使用线程池的好处是减少在创建和销毁线程上所消耗的时间以及系统资源的开销，解决资源不足的问题。如果不使用线程池的话，有可能造成系统创建大量的同类线程而消耗完线程而导致消耗完内存或者过度切换的问题。
- 为什么要不能通过Executors去创建线程池？ 因为FixedThreadPool和SingleThreadPool使用的是无界队列，会堆积大量的请求，造成OOM。还有CachedThreadPool和ScheduledThreadPool会造成Integer.MAX_VALUE，会创建大量的线程，造成OOM
分析CountDownLatch
CountDownLatch用于协调多个线程的同步，能让一个线程在等待其他线程执行完任务后，再继续执行。内部是通过一个计数器去完成实现。
静态内部类Sync继承AQS，通过state变量完成计数器的实现。
/**
* Synchronization control For CountDownLatch.
* Uses AQS state to represent count.
*/
private static final class Sync extends AbstractQueuedSynchronizer {
private static final long serialVersionUID = 4982264981922014374L;
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;
}
}
}CountDownLatch构造方法，count代表需要执行的线程数量。
/**
* Constructs a {@code CountDownLatch} initialized with the given count.
*
* @param count the number of times {@link #countDown} must be invoked
* before threads can pass through {@link #await}
* @throws IllegalArgumentException if {@code count} is negative
*/
public CountDownLatch(int count) {
if (count < 0) throw new IllegalArgumentException("count < 0");
this.sync = new Sync(count);
}在计数值 > 0的情况下，每当一个线程完成任务，计数减去1。
/**
* Decrements the count of the latch, releasing all waiting threads if
* the count reaches zero.
*
* <p>If the current count is greater than zero then it is decremented.
* If the new count is zero then all waiting threads are re-enabled for
* thread scheduling purposes.
*
* <p>If the current count equals zero then nothing happens.
*/
public void countDown() {
sync.releaseShared(1);
}让当前线程等待其他线程，直到计数为0，除非当前线程被中断了。
public void await() throws InterruptedException {
sync.acquireSharedInterruptibly(1);
}让当前线程等待其他线程，如果超过指定的timeout时间范围，那么忽略要等待的线程，
直接执行。
public boolean await(long timeout, TimeUnit unit)
throws InterruptedException {
return sync.tryAcquireSharedNanos(1, unit.toNanos(timeout));
}返回当前计数量。
/**
* Returns the current count.
*
* <p>This method is typically used for debugging and testing purposes.
*
* @return the current count
*/
public long getCount() {
return sync.getCount();
}当计数器应该为0，所有的线程执行完自己的任务。在CountDownLatch等待的线程，可以继续执行的任务。
使用场景
适用于一个任务的执行需要等待其他任务执行完毕，方可执行的场景。
场景1：一群学生在教室考试，学生们都完成了作答，老师才可以进行收卷操作。
场景2：110跨栏比赛中，所有运动员准备好起跑姿势，进入到预备状态，等待裁判一声枪响。裁判开了枪，所有运动员才可以开跑。
CountDownLatch是一次性的，只能通过构造方法设置初始计数量，计数完了无法进行复位，不能达到复用。而CyclicBarrier可以实现复用。
场景1
一个线程的执行要等待其他线程执行完毕后，才能继续执行。
public static void test1() {
final CountDownLatch countDownLatch = new CountDownLatch(2);
ExecutorService executorService = new CustomThreadPoolExecutor(2,
2, 0L,
TimeUnit.MILLISECONDS, new ArrayBlockingQueue<Runnable>(10));
for (int i = 0; i < 2; i++) {
CustomThreadPoolExecutor.CustomTask task = new CustomThreadPoolExecutor.CustomTask(new Runnable() {
@Override
public void run() {
System.out.println("子线程" + Thread.currentThread().getName()
+ "正在执行...");
System.out.println("子线程" + Thread.currentThread().getName()
+ "执行完毕...");
countDownLatch.countDown();
}
}, "success");
executorService.submit(task);
}
try {
System.out.println("等待2个线程...");
countDownLatch.await();
executorService.shutdown();
System.out.println("2个线程执行完毕...");
} catch (InterruptedException ex) {
ex.printStackTrace();
}
}输出结果
等待2个线程...
子线程pool-1-thread-2正在执行...
子线程pool-1-thread-2执行完毕...
子线程pool-1-thread-1正在执行...
子线程pool-1-thread-1执行完毕...
2个线程执行完毕..场景2
多个线程在某一个时刻同时执行。
public static void test2() {
final CountDownLatch start = new CountDownLatch(1);
final CountDownLatch end = new CountDownLatch(10);
ExecutorService executorService = new CustomThreadPoolExecutor(10,
10, 0L,
TimeUnit.MILLISECONDS, new ArrayBlockingQueue<Runnable>(10));
for (int i = 0; i < 10; i++) {
CustomThreadPoolExecutor.CustomTask task = new CustomThreadPoolExecutor.CustomTask(new Runnable() {
@Override
public void run() {
try {
System.out.println("子线程" + Thread.currentThread().getName()
+ "正在执行...");
start.await();
System.out.println("子线程" + Thread.currentThread().getName()
+ "执行完毕...");
} catch (InterruptedException ex) {
ex.printStackTrace();
} finally {
end.countDown();
}
}
}, "success");
executorService.submit(task);
}
start.countDown();
try {
System.out.println("等待10个线程...");
end.await();
executorService.shutdown();
System.out.println("10个线程执行完毕...");
} catch (InterruptedException ex) {
ex.printStackTrace();
}
}输出结果
子线程pool-1-thread-2正在执行...
子线程pool-1-thread-1正在执行...
子线程pool-1-thread-3正在执行...
子线程pool-1-thread-4正在执行...
子线程pool-1-thread-5正在执行...
子线程pool-1-thread-6正在执行...
子线程pool-1-thread-7正在执行...
等待10个线程...
子线程pool-1-thread-2执行完毕...
子线程pool-1-thread-8正在执行...
子线程pool-1-thread-8执行完毕...
子线程pool-1-thread-1执行完毕...
子线程pool-1-thread-3执行完毕...
子线程pool-1-thread-4执行完毕...
子线程pool-1-thread-10正在执行...
子线程pool-1-thread-10执行完毕...
子线程pool-1-thread-5执行完毕...
子线程pool-1-thread-6执行完毕...
子线程pool-1-thread-7执行完毕...
子线程pool-1-thread-9正在执行...
子线程pool-1-thread-9执行完毕...
10个线程执行完毕...CountDownLatch源码分析
CountDownLatch的构造方法初始计数器值，是通过其内部类Sync的构造方法来实现的。
Sync(int count) {
setState(count);
}AQS中的state变量可以表示状态。对于ReentrantLock而言，代表着锁获取的次数。而对于CountDownLatch代表着计数器的值。state变量通过volatile修饰，具有可见性，可以在多个线程中共享变量。
/**
* Sets the value of synchronization state.
* This operation has memory semantics of a {@code volatile} write.
* @param newState the new state value
*/
protected final void setState(int newState) {
state = newState;
}AQS中的CAS操作，使其state变量具有原子性。
protected final boolean compareAndSetState(int expect, int update) {
// See below for intrinsics setup to support this
return unsafe.compareAndSwapInt(this, stateOffset, expect, update);
}
private static final Unsafe unsafe = Unsafe.getUnsafe();
private static final long stateOffset;
private static final long headOffset;
private static final long tailOffset;
private static final long waitStatusOffset;
private static final long nextOffset;
static {
try {
stateOffset = unsafe.objectFieldOffset
(AbstractQueuedSynchronizer.class.getDeclaredField("state"));
headOffset = unsafe.objectFieldOffset
(AbstractQueuedSynchronizer.class.getDeclaredField("head"));
tailOffset = unsafe.objectFieldOffset
(AbstractQueuedSynchronizer.class.getDeclaredField("tail"));
waitStatusOffset = unsafe.objectFieldOffset
(Node.class.getDeclaredField("waitStatus"));
nextOffset = unsafe.objectFieldOffset
(Node.class.getDeclaredField("next"));
} catch (Exception ex) { throw new Error(ex); }
}我们来看一下AQS中的Node节点结构。
- 当waitStatus = CANCELLED时，说明因为超时或者中断，节点会被设置为取消状态。处于取消状态的节点不会参与到竞争中。它会一直保持取消状态，会转变到其他状态。
- 当waitStatus = SIGNAL时，说明当前节点的后继节点处于等待状态。而当前节点的线程如果释放了同步状态或者被取消，将会通知后继节点，使后继节点的线程可以得到运行。
- 当waitStatus = CONDITION，说明该节点在等待队列中，节点线程等待在Condition上。当其他线程对Condition调用了signal()后，该节点会从等待队列中转移到同步队列中，加入到同步状态的获取中。
- 当waitStatus = PROPAGATE，说明下一次共享式获取同步状态，将会无条件的传播下去。
static final class Node {
/** 共享 */
static final Node SHARED = new Node();
/** 独占 */
static final Node EXCLUSIVE = null;
static final int CANCELLED = 1;
static final int SIGNAL = -1;
static final int CONDITION = -2;
static final int PROPAGATE = -3;
/** 等待状态 */
volatile int waitStatus;
/** 前驱节点 */
volatile Node prev;
/** 后继节点 */
volatile Node next;
/** 获取同步状态的线程 */
volatile Thread thread;
Node nextWaiter;
final boolean isShared() {
return nextWaiter == SHARED;
}
final Node predecessor() throws NullPointerException {
Node p = prev;
if (p == null)
throw new NullPointerException();
else
return p;
}
Node() {
}
Node(Thread thread, Node mode) {
this.nextWaiter = mode;
this.thread = thread;
}
Node(Thread thread, int waitStatus) {
this.waitStatus = waitStatus;
this.thread = thread;
}
}countDown()实际是调用AQS中的releaseShared方法中，达到计数减1的目的。
public void countDown() {
sync.releaseShared(1);
}
countDown主要流程(来自于互联网).png
以共享模式去释放锁，如果tryReleaseShared方法释放锁成功，则执行AQS中的doReleaseShared方法去唤醒等待线程，并且返回true；否则返回false，说明锁释放失败。
public final boolean releaseShared(int arg) {
if (tryReleaseShared(arg)) {
doReleaseShared();
return true;
}
return false;
}tryReleaseShared在CountDownLatch中被重写。通过轮询 + CAS方式达到释放锁的目的。第一次循环的时候判断当前state变量，如果等于0，说明计数器值为0或者说锁没有被持有，可以直接返回false。然后进行CAS操作，让获取锁的次数减少1或者说计数器值减少1。如果nextc等于0，说明计数值为0或者持有锁的次数为0，可以让唤醒等待的线程，所以返回true，否则返回false，代表释放锁失败。
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;
}
}首先获得head节点。如果head节点不等于空且head节点不等于tail节点，获得head节点的waitStatus。判断当前head节点状态是否是SINGAL。处于SINGAL状态的节点，说明当前节点的后继节点处于被唤醒的状态。如果CAS操作将head节点的waitStatus重置为0失败，那么跳出当前循环，继续执行下一次循环(重新检查)。如果重置成功，那么调用unparkSuccessor方法唤醒后继节点。 如果当前head节点状态等于0，通过CAS操作将waitStatus设置为PROPAGATE(传播)状态，确保可以向后一个节点传播下去。如果CAS操作失败，那么当前循环，继续执行下一次循环。最后的h == head，是判断head节点是否发生变化。如果没有发生变化，结束循环。如果发生变化，必须再次循环。
/**
* 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))
continue; // loop to recheck cases
unparkSuccessor(h);
}
else if (ws == 0 &&
!compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
continue; // loop on failed CAS
}
if (h == head) // loop if head changed
break;
}
}获取head节点的waitStatus，如果小于0，进行CAS操作重置为0。获取head节点的后继节点，如果后继节点等于null或者后继节点的waitStaus大于0(说明后继节点处于CANCELLED状态),那么从队列从尾部往前进行遍历寻找waitStatus小于等于0的节点。如果这个遍历出来的节点不等于null的话，那么通过LockSupport.unpark()唤醒这个节点中的线程。
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;
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;
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);
}分析完了CountDownLatch的countDown()方法，接着撸await()方法
await主要流程(来自于互联网).png
await方法会使当前线程在计数器值为0之前，一直处于等待状态，除非中断。
public void await() throws InterruptedException {
sync.acquireSharedInterruptibly(1);
}AQS中的acquireSharedInterruptibly方法，会判断线程是否中断。如果中断， 抛出InterruptedException异常。值得注意的是Thread.interrupted()方法，是测试当前线程是否中断。该方法会清除线程的中断状态。换句话说，如果调用这个方法2次，那么第二次会直接返回false，除非当前线程在第一次调用之后再次被中断。如果tryAcquireShared()小于0(说明该计数器值大于0)，继续执行doAcquireSharedInterruptibly。
public final void acquireSharedInterruptibly(int arg)
throws InterruptedException {
if (Thread.interrupted())
throw new InterruptedException();
if (tryAcquireShared(arg) < 0)
doAcquireSharedInterruptibly(arg);
}在CountDownLatch中重写了AQS中的tryAcquireShared()方法，这里只是简单的判断state变量。如果state等于0(说明计数值为0)，返回1，否则返回-1(说明计数器值大于0)。
protected int tryAcquireShared(int acquires) {
return (getState() == 0) ? 1 : -1;
}很明显，是通过轮询的方式去获取共享锁。首先将当前线程包装成类型为SHARED的节点，标志为共享类型的节点。获取当前节点的前驱节点。如果当前节点的前驱节点为head节点的话，说明该节点是在AQS队列中等待获取锁的第一个节点。调用CountDownLatch中的tryAcquireShared()尝试去获取锁。返回的值大于0的话，说明获取锁成功。如果获取共享锁成功，那么把当前节点设置为AQS同步队列中的head节点，同时将p.next置为null(方便GC)。回到头看，如果当前节点的前驱节点不是head节点或者获取锁失败，我们需要调用shouldParkAfterFailedAcquire()方法判断当前线程是否需要挂起，如果需要挂起调用 parkAndCheckInterrupt()
private void doAcquireSharedInterruptibly(int arg)
throws InterruptedException {
final Node node = addWaiter(Node.SHARED);
boolean failed = true;
try {
for (;;) {
final Node p = node.predecessor();
if (p == head) {
int r = tryAcquireShared(arg);
if (r >= 0) {
setHeadAndPropagate(node, r);
p.next = null; // help GC
failed = false;
return;
}
}
if (shouldParkAfterFailedAcquire(p, node) &&
parkAndCheckInterrupt())
throw new InterruptedException();
}
} finally {
if (failed)
cancelAcquire(node);
}
}通过当前线程构造出Node节点，mode用于标志是独占还是共享。如果队列非空，则快速入队。通过CAS将node节点置为tail节点，并返回node节点。如果CAS失败或者队列为空，那么通过enq()入队。
private Node addWaiter(Node mode) {
Node node = new Node(Thread.currentThread(), mode);
// Try the fast path of enq; backup to full enq on failure
Node pred = tail;
if (pred != null) {
node.prev = pred;
if (compareAndSetTail(pred, node)) {
pred.next = node;
return node;
}
}
enq(node);
return node;
}enq中通过死循环的方式保证节点的正确插入。如果队列为空，那么创建一个新的节点。通过CAS将新节点设置为head节点，同时也将head节点设置为tail节点。当队列只有一个元素时，head节点等于tail节点。在循环中，唯一跳出循环的条件是通过CAS将node节点设置为tail节点。这样的话，enq方法将并发插入节点的请求变得串行化了。
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;
}
}
}
}这里将node节点设置为head节点。经过一些条件判断后，获取head节点的后继节点。如果后继节点等于null或者后继节点也是共享节点，那么调用doReleaseShared去唤醒它。
private void setHeadAndPropagate(Node node, int propagate) {
Node h = head; // Record old head for check below
setHead(node);
/*
* Try to signal next queued node if:
* Propagation was indicated by caller,
* or was recorded (as h.waitStatus either before
* or after setHead) by a previous operation
* (note: this uses sign-check of waitStatus because
* PROPAGATE status may transition to SIGNAL.)
* and
* The next node is waiting in shared mode,
* or we don't know, because it appears null
*
* The conservatism in both of these checks may cause
* unnecessary wake-ups, but only when there are multiple
* racing acquires/releases, so most need signals now or soon
* anyway.
*/
if (propagate > 0 || h == null || h.waitStatus < 0 ||
(h = head) == null || h.waitStatus < 0) {
Node s = node.next;
if (s == null || s.isShared())
doReleaseShared();
}
}如果pred节点(node节点的前驱节点)的状态是SIGNAL，说明该pred节点的线程如果释放了同步状态或者被取消，会通知其后继节点(也就是node节点)。所以我们可以安全让node节点的线程挂起。如果pred节点处于取消状态，我们进行死循环， 直到pred节点的状态不是取消状态。通过死循环，我们能确保node节点的前驱节点不处于取消状态。反之，如果一开始pred节点不处于取消状态，那么我们通过CAS将pre节点的状态置为SIGNAL，为后面循环涉及到的park操作进行准备。但是有一点要注意，我们进行park之前，要确保当前节点获取锁失败。
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;
}调用LockSupport.park()挂起当前线程，并且返回中断状态。
private final boolean parkAndCheckInterrupt() {
LockSupport.park(this);
return Thread.interrupted();
}CountDownLatch的await(long timeout, TimeUnit unit)和await()方法实现方式基本一致，只不过加入了超时机制而已。
// 返回false，代表超时
public boolean await(long timeout, TimeUnit unit)
throws InterruptedException {
return sync.tryAcquireSharedNanos(1, unit.toNanos(timeout));
} // 返回false，代表超时。返回true，代表获得共享锁成功
public final boolean tryAcquireSharedNanos(int arg, long nanosTimeout)
throws InterruptedException {
if (Thread.interrupted())
throw new InterruptedException();
return tryAcquireShared(arg) >= 0 ||
doAcquireSharedNanos(arg, nanosTimeout);
}如果在nanosTimeout时间范围内，还没有获取共享锁成功的话，直接返回false。spinForTimeoutThreadshold的值为1000nanoseconds。如果shouldParkAfterFailedAcquire(p, node)返回true且超时时间大于阀值spinForTimeoutThreadshold的话，会通过LockSupport.parkNanos(this, nanosTimeout);让线程挂起nanosTimeout时间。这样的策略体现是：如果超时时间很短的话，就不把当前线程挂起，而是通过自旋，这样线程获取锁很快就释放的情况下，可以减少cpu资源和线程挂起和恢复的性能损耗。
private boolean doAcquireSharedNanos(int arg, long nanosTimeout)
throws InterruptedException {
if (nanosTimeout <= 0L)
return false;
final long deadline = System.nanoTime() + nanosTimeout;
final Node node = addWaiter(Node.SHARED);
boolean failed = true;
try {
for (;;) {
final Node p = node.predecessor();
if (p == head) {
int r = tryAcquireShared(arg);
if (r >= 0) {
setHeadAndPropagate(node, r);
p.next = null; // help GC
failed = false;
return true;
}
}
nanosTimeout = deadline - System.nanoTime();
if (nanosTimeout <= 0L)
return false;
if (shouldParkAfterFailedAcquire(p, node) &&
nanosTimeout > spinForTimeoutThreshold)
LockSupport.parkNanos(this, nanosTimeout);
if (Thread.interrupted())
throw new InterruptedException();
}
} finally {
if (failed)
cancelAcquire(node);
}
}总结
通过重写AQS中的模板方法，可以实现共享锁和独占锁。
如果获取共享锁失败，请求共享锁的线程封装成SHARED类型的Node对象并且加入到AQS同步队列中，并挂起Node对象对应的线程，等待锁的释放。待共享锁的可以被获取后，从head节点开始依次唤醒head节点之后的所有SHARED类型的节点，实现共享状态的传播。而独占锁则是，当锁被头节点获取后，只有头节点获得了锁，其他节点的线程继续沉睡。等待锁被释放了，才会唤醒下一个节点的线程，少了setHeadAndPropagate()这一步。
尾言
大家好，我是cmazxiaoma(寓意是沉梦昂志的小马)，希望和你们一起成长进步，感谢各位阅读本文章。
如果您对这篇文章有什么意见或者错误需要改进的地方,欢迎与我讨论。
如果您觉得还不错的话,希望你们可以点个赞。
希望我的文章对你能有所帮助。
有什么意见、见解或疑惑，欢迎留言讨论。
最后送上：心之所向，素履以往。生如逆旅，一苇以航。
saoqi.png