CountDownLatch使用简介与源码解析

CountDownLatch试用场景

来自于Javadoc的解释

    A synchronization aid that allows one or more threads to wait until
    a set of operations being performed in other threads completes.

表示一个或者多个线程等待一系列的操作完成。

CountDownLatch是一个同步工具类，用来协调多个线程之间的同步，或者说起到线程之间的通信作用，简单的说CountDownLatch就是一个计数器，能够使一个或者多个线程等待另外一些线程操作完成之后，再继续执行，计数器的数量就是线程的数量，当每个线程完成自己的任务之后，计数器减一
当计数器的数值变为0的时候，表示所有线程都完成了自己的任务，等待在CountDownLatch上的线程可以继续执行自己的任务。

例如当需要对两个表进行查询，然后将查询结果合并进行下一步操作，由于对两个表的查询时IO密集型操作，我们可以考虑使用多线程来提高性能，但是我们需要在两个查询操作完成之后通知主线程进行合并操作等，我们可以用CountDownLatch来完成，代码如下，通过Thread.sleep(1000)来模拟一个IO耗时的操作

public class CountDownLatchDemo {

    static ThreadPoolExecutor poolExecutor = new ThreadPoolExecutor(4, 8, 60,
            TimeUnit.SECONDS, new ArrayBlockingQueue<Runnable>(100), new ThreadPoolExecutor.CallerRunsPolicy());

    public static void main(String[] args){

        CountDownLatch countDownLatch = new CountDownLatch(2);

        poolExecutor.execute(() -> {
            System.out.println("query from table A");
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            countDownLatch.countDown();
        });

        poolExecutor.execute(() -> {
            System.out.println("query from table B");
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            countDownLatch.countDown();
        });
        try {
            countDownLatch.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println("merge result");
    }
}

执行结果为

query from table A
query from table B
merge result

CountDownLatch源码解析

CountDownLatch通过内部类Sync来完成线程间的同步和通信，

    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;
            }
        }
    }

Sync实现了AQS的两个模板方法tryAcquireShared和tryReleaseShared，分别表示获取同步锁和释放同步锁的操作。CountDownLatch中最重要的两个方法分别为countDown()和await()

countDown方法如下所示

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

直接调用AQS的releaseShared方法来修改同步状态，将state的值减一。

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();
        if (tryAcquireShared(arg) < 0)
            doAcquireSharedInterruptibly(arg);
    }

首先调用tryAcquireShared方法，该方法若state为0则返回1，反之返回-1，即若state为0时，该方法直接返回，若state不为0，该方法会执行doAcquireSharedInterruptibly方法，

    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);
        }
    }

doAcquireSharedInterruptibly方法会自旋的调用Sync实现的tryAcquireShared方法，直至state为0，该方法才能返回，以此来达到阻塞线程的目的。