java并发学习之一：CountDownLatch

看了几个月的《Java Concurrency in Practice》到了今天终于算可以收尾了，之前留下的看不懂的代码，现在也基本明晰了一些 

全书介绍了很多细节问题，很多注意的点，很多原则性问题，个人感觉，无论看几遍，都是值得的。但很多都是一些需要去记忆的东西，这个是需要经验的积累的。 
真正想在思考上，在设计上得到更大的提高，看来必然是要落在了concurrent包的数个同步器的实现的分析和对AQS的理解上了 

同步器是针对一些通用的场景设计的，由Doug Lea实现的，换句话来说，其实从同步器可以看出常用的一些需求，既然有了需求，又学会了工具（AQS框架）的使用，就可以按照自己的思维，试着重新去实现一下了 


CountDownLatch 

CountDownLatch是针对这样一个需求设计的：n个参与者需要进行一件事情（比如开会），只有当他们都到了，才能正式开始，所以先到的需要等，并且同时开始 

首先，先用点“拙劣”的工具(1.5之前的实现方法)试着写一下实现 

public class TestOfCountDownLatches {

    int count;

    Object lock = new Object();

    public TestOfCountDownLatches(int count)
    {
        this.count = count;
    }

    public void countDown()
    {
        synchronized(lock)
        {
            count--;
            if(count == 0)
            {
                lock.notifyAll();
            }
        }
    }

        public void await() throws InterruptedException
    {
        synchronized(lock)
        {
//中间还出现了点小问题：没写if(count == 0)，这就导致了如果主线程先执行了countdown，其他线程再执行await，就会无限制的等待着了，从这个小错误也学会了并发框架的一个原则：阻塞还是不阻塞，应该是严格依赖于状态（state）的，而与先后次序没关系
            if(count != 0)
                lock.wait();
        }
    }
}

测试方法： 

public long timeTasks1(int nThreads,final Runnable task)
        {
            final TestOfCountDownLatches endLatch = new TestOfCountDownLatches(nThreads);
            final TestOfCountDownLatches startLatch = new TestOfCountDownLatches(1);
            for(int i = 0;i<nThreads;i++)
            {
                Thread t = new Thread(){
                    public void run() {
                        try {
                            startLatch.await();
                            try {
                                task.run();
                            } finally {
                                endLatch.countDown();
                            }
                        } catch (InterruptedException ignored) { }
                    }
                };
                t.start();
            }
            long start = System.nanoTime();
            try {
                startLatch.countDown();
                endLatch.await();
            } catch (InterruptedException e) {

            }
            long end = System.nanoTime();
            return end-start;
        }

跑了一下，没啥问题，与CountDownLatches的功能基本一致的 

再看看CountDownLatches的实现（主要是看它的同步器） 

private static final class Sync extends AbstractQueuedSynchronizer {
        private static final long serialVersionUID = 4982264981922014374L;

        Sync(int count) {
            setState(count);
        }

        int getCount() {
            return getState();
        }
//被await()调用
        public int tryAcquireShared(int acquires) {
            return getState() == 0? 1 : -1;
        }
//被countDown()调用
        public 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;
            }
        }
    }

在实现原理上是一致的，await时比较一下state（相当于第一个版本的count），如果满足了，则不阻塞（返回1），如果不满足，阻塞（返回-1）（关于阻塞与否的代码都在AQS里面） 
在countDown时则进行一下--操作，如果满足0“return nextc == 0;”了，则释放阻塞 

既然主要目的是学习AQS，就debug了一下，尽量去了解AQS的实现原理 
主要了解到以下几个方面：（下面的描述只针对AQS的shared模式） 
1.AQS没用正常的wait，notifyAll，lock的阻塞等方法，而是用了一个LockSupport对象来支持类似操作，该对象的方法全是本地方法，相当于对synchronized和wait等操作用了一个统一的形式 
2.AQS中维持了一个等待队列，很奇特的是，该等待队列“完全没用锁”，比如说一个acquice操作，当没成功的时候，是需要将该线程进入等待队列的，多线程的访问，所以这个入队的过程应该要保证它的原子性 
但AQS没有这样做，它只用了一些非常高效的改变int型的原子方法（由Unsafe提供），针对每一个需要保证原子性的操作（如入队），它都用了一个while(true)这样的形式，一旦失败(比如发现队尾已经被别的线程改变了)，则从新获取当前状态，重新入队，很神奇，很高效，同时也需要考虑得面面俱到的一个实现，在文章中称之为“非阻塞算法”，在后文会对该算法写一篇详细介绍 
虽然用1.5之前的wait等方法可以实现，但无疑用AQS这套框架会更高效，响应更快 

再贴一个使用CountDownLatches的例子做为结尾 

public long timeTasks2(int nThreads, final Runnable task)
                throws InterruptedException {
            final CountDownLatch startGate = new CountDownLatch(1);
            final CountDownLatch endGate = new CountDownLatch(nThreads);

            for (int i = 0; i < nThreads; i++) {
                Thread t = new Thread() {
                    public void run() {
                        try {
                            startGate.await();
                            try {
                                task.run();
                            } finally {
                                endGate.countDown();
                            }
                        } catch (InterruptedException ignored) { }
                    }
                };
                t.start();
            }

            long start = System.nanoTime();
            startGate.countDown();
            endGate.await();
            long end = System.nanoTime();
            return end-start;
        }