Java并发工具类--CountDownLatch详解

CountDownLatch是一个同步工具类,用来协调多个线程之间的同步。

CountDownLatch能够使一个线程在等待另外一些线程完成各自工作之后,再继续执行。使用一个计数器进行实现。计数器初始值为线程的数量。当每一个线程完成自己任务后,计数器的值就会减一。当计数器的值为0时,表示所有的线程都已经完成一些任务,然后在CountDownLatch上等待的线程就可以恢复执行接下来的任务。

举个例子来说明CountDownLatch的使用:

百米赛跑,10名运动员选手到达场地等待裁判口令,裁判一声口令,选手听到后同时起跑,当所有选手到达终点,裁判进行汇总排名。

public class CountDownLatchTest {

    public static void main(String[] args) {
        ExecutorService service = Executors.newCachedThreadPool();
        final CountDownLatch cdOrder = new CountDownLatch(1);
        final CountDownLatch cdAnswer = new CountDownLatch(10);
        for(int i=0; i<10; i++){
            Runnable runnable = new Runnable() {
                @Override
                public void run() {
                    try{
                        System.out.println("选手" + Thread.currentThread().getName() + "正在等待裁判发布口令");
                        cdOrder.await();
                        System.out.println("选手" + Thread.currentThread().getName() + "已接受裁判口令");
                        Thread.sleep((long) (Math.random() * 10000));
                        System.out.println("选手" + Thread.currentThread().getName() + "到达终点");
                        cdAnswer.countDown();
                    }catch (InterruptedException ie){
                        ie.printStackTrace();
                    }
                }
            };
            service.execute(runnable);
        }

        try {
            Thread.sleep((long) (Math.random() * 10000));
            System.out.println("裁判"+Thread.currentThread().getName()+"即将发布口令");
            cdOrder.countDown();
            System.out.println("裁判"+Thread.currentThread().getName()+"已发送口令,正在等待所有选手到达终点");
            cdAnswer.await();
            System.out.println("所有选手都到达终点");
            System.out.println("裁判"+Thread.currentThread().getName()+"汇总成绩排名");
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        service.shutdown();
    }
}

运行结果如下:

选手pool-1-thread-1正在等待裁判发布口令
选手pool-1-thread-3正在等待裁判发布口令
选手pool-1-thread-2正在等待裁判发布口令
选手pool-1-thread-4正在等待裁判发布口令
选手pool-1-thread-5正在等待裁判发布口令
选手pool-1-thread-6正在等待裁判发布口令
选手pool-1-thread-7正在等待裁判发布口令
选手pool-1-thread-8正在等待裁判发布口令
选手pool-1-thread-9正在等待裁判发布口令
选手pool-1-thread-10正在等待裁判发布口令
裁判main即将发布口令
裁判main已发送口令,正在等待所有选手到达终点
选手pool-1-thread-3已接受裁判口令
选手pool-1-thread-2已接受裁判口令
选手pool-1-thread-1已接受裁判口令
选手pool-1-thread-9已接受裁判口令
选手pool-1-thread-8已接受裁判口令
选手pool-1-thread-7已接受裁判口令
选手pool-1-thread-6已接受裁判口令
选手pool-1-thread-5已接受裁判口令
选手pool-1-thread-4已接受裁判口令
选手pool-1-thread-10已接受裁判口令
选手pool-1-thread-7到达终点
选手pool-1-thread-3到达终点
选手pool-1-thread-10到达终点
选手pool-1-thread-9到达终点
选手pool-1-thread-1到达终点
选手pool-1-thread-5到达终点
选手pool-1-thread-4到达终点
选手pool-1-thread-2到达终点
选手pool-1-thread-8到达终点
选手pool-1-thread-6到达终点
所有选手都到达终点
裁判main汇总成绩排名

具体看一下CountDownLatch是如何实现线程调度的。

首先看一下其构造方法:

public CountDownLatch(int count) {
	if (count < 0) throw new IllegalArgumentException("count < 0");
	this.sync = new Sync(count);
}

继续看一下Sync。

// Sync继承自AQS
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) {
		// 当状态为0,则该线程获取到该共享锁
		return (getState() == 0) ? 1 : -1;
	}

	protected boolean tryReleaseShared(int releases) {
		// 减少count,当检测到状态值为0时,通知同步队列中被挂起的线程
		for (;;) {
			int c = getState();
			if (c == 0)
				return false;
			int nextc = c-1;
			if (compareAndSetState(c, nextc))
				return nextc == 0;
		}
	}
}

可以发现,CountDownLatch是基于AQS共享锁来实现的,,只要共享锁状态值不为0,则请求共享锁的线程均会添加到同步队列中,阻塞挂起,等待被通知。

接着看一下await方法:

public void await() throws InterruptedException {
	sync.acquireSharedInterruptibly(1);
}

直接调用的是AQS的acquireSharedInterruptibly方法:

public final void acquireSharedInterruptibly(int arg)
		throws InterruptedException {
	if (Thread.interrupted())
		throw new InterruptedException();
	// tryAcquireShared由Sync实现,即只要状态不为0,则返回-1
	if (tryAcquireShared(arg) < 0)
		doAcquireSharedInterruptibly(arg);
}

CountDownLatch初始化后状态值肯定不为0,所以当前线程tryAcquireShared必然返回-1,继续执行doAcquireSharedInterruptibly方法。

private void doAcquireSharedInterruptibly(int arg)
	throws InterruptedException {
	// 将当前线程添加到同步队列
	final Node node = addWaiter(Node.SHARED);
	boolean failed = true;
	try {
		for (;;) {
			final Node p = node.predecessor();
			// 若当前线程所在节点的前继节点为头节点,则执行tryAcquireShared尝试获取共享锁
			if (p == head) {
				// 因为状态值不为0,tryAcquireShared必然返回-1
				int r = tryAcquireShared(arg);
				if (r >= 0) {
					setHeadAndPropagate(node, r);
					p.next = null; // help GC
					failed = false;
					return;
				}
			}
			// 将前继节点状态CAS更改为SIGNAL后,然后线程阻塞挂起
			if (shouldParkAfterFailedAcquire(p, node) &&
				parkAndCheckInterrupt())
				throw new InterruptedException();
		}
	} finally {
		if (failed)
			cancelAcquire(node);
	}
}

综上,持有CountDownLatch的线程只要调用await方法,就会被添加进AQS的同步队列,并被阻塞挂起。

那这些被阻塞挂起的线程啥时候会被唤醒继续执行呢?

答案在CountDownLatch的countDown方法中:

public void countDown() {
	sync.releaseShared(1);
}

底层调用的是AQS的releaseShared方法:

public final boolean releaseShared(int arg) {
	// 若状态执行-1操作后,状态值未归零,tryReleaseShared返回false
	// 若状态执行-1操作后,状态值归零,tryReleaseShared返回true
	if (tryReleaseShared(arg)) {
		// 若状态值归零,继续执行doReleaseShared方法
		doReleaseShared();
		return true;
	}
	return false;
}

当状态值归零后,当前线程会执行AQS的doReleaseShared方法,doReleaseShared方法我们在之前AQS详解的系列文章里详细介绍过,该方法是一个"唤醒风暴",其会唤醒同步队列中阻塞挂起的线程。

唤醒后的线程会进行获取锁的操作,当状态值归零后,由于tryReleaseShared恒返回1,代表任何线程均可以获取共享锁成功,当线程获取到锁之后,会执行setHeadAndPropagate方法将当前节点置为头节点并唤醒后继节点,后继节点被唤醒后执行获取锁操作,如此反复,直到同步队列中的所有阻塞线程被唤醒。

需要注意的是:CountDownLatch是一次性的,计算器的值只能在构造方法中初始化一次,之后没有任何机制再次对其设置值,当CountDownLatch使用完毕后,它不能再次被使用。


版权声明:本文为WenWu_Both原创文章,遵循CC 4.0 BY-SA版权协议,转载请附上原文出处链接和本声明。