Java 多线程之闭锁-CountDownLatch

    实现并发的最直接方式是在操作系统级别使用进程。进程是运行在自己的地址空间内的自包容程序。多任务操作系统可以通过周期性地将CPU从一个进程切换到另一个进程，来实现同时运行多个进程的。操作系统将进程相互隔离开，因此他们不会相互干扰，这使得通过进程实现并发编程相对容易一些。而JAVA的并发时通过多线程机制实现的。

   一个线程就是在进程中的一个单一的顺序控制流，因此，单个进程可以拥有多个并发执行的任务，使得程序看起来好像都有其自己的CPU一样。其底层的机制是切分CPU的时间片。

    java.util.concurrency包提供了很多并发编程的工具类，可以极大提高并发编程的效率。

 

   闭锁是一种同步工具类，可以延迟线程的进度直到闭锁到达终止状态。Latch在英语中就是门栓的意思，所以形象地说闭锁就相当于一扇门，在日常生 活中我们都遇到过类似的场景，进入一个场馆前，必须达到一定的条件，比如活动开始前半小时可以入场；如果来早了的话，对不起，以便等着。 CountDownLatch 在多线程中也是这样的作用，在闭锁到达结束状态前，这扇门是一直关闭的，不允许任何线程通过，当到达结束状态时，这扇门就 保持打开，并且是永久的处于打开状态；也就是说这个门是一次性的。 如同柏林墙一样，推到了就建不起来了。

     闭锁可以用来确保某些活动直到其他活动都完成后才继续执行。CountDownLatch是一种灵活的闭锁实现，它可以使一个或多个线程 等待一组事件的发生。闭锁包括一个计数器，该计数器被初始化为一个正整数，表示等待的事件数量。有一等待时间发生时，使用countDown方法递减计数 器，而await方法就是等待所有事件的发生，也就是计数器的值为0.如果计数器的值不为0 那么await 方法需要一直等待。

 

public class CountDownLatch {
    /**
     * 闭锁的同步控制类
     */
    private static final class Sync extends AbstractQueuedSynchronizer {
        private static final long serialVersionUID = 4982264981922014374L;
		// 
        Sync(int count) {
            setState(count);
        }

        int getCount() {
            return getState();
        }
		//获取共享锁，如果当前线程数量为0则表示获取到锁，否则无法获取
        protected int tryAcquireShared(int acquires) {
            return (getState() == 0) ? 1 : -1;
        }
		//释放共享锁
        protected boolean tryReleaseShared(int releases) {
            for (;;) {
                int c = getState();
                if (c == 0)
                    return false;
                int nextc = c-1;
                if (compareAndSetState(c, nextc))
                    return nextc == 0;
            }
        }
    }

    private final Sync sync;

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

    /**
     *  当前线程一直等到闭锁的数量为0才可以执行
     */
    public void await() throws InterruptedException {
        sync.acquireSharedInterruptibly(1);
    }

    /**
     * 当前线程等到闭锁的数量为0或者超时
     */
    public boolean await(long timeout, TimeUnit unit)
        throws InterruptedException {
        return sync.tryAcquireSharedNanos(1, unit.toNanos(timeout));
    }

    /**
     * 闭锁减去1
     */
    public void countDown() {
        sync.releaseShared(1);
    }

    /**
     * 获取当前闭锁的数量
     */
    public long getCount() {
        return sync.getCount();
    }

   
    public String toString() {
        return super.toString() + "[Count = " + sync.getCount() + "]";
    }
}

 AbstractQueuedSynchronizer#acquireSharedInterruptibly()方法，首先调用子类的实现方法判断是否可以获取到锁，如果无法获取到锁，则调用doAcquireAsharedInterruptibly()方法

   public final void acquireSharedInterruptibly(int arg)
            throws InterruptedException {
        if (Thread.interrupted())
            throw new InterruptedException();
        if (tryAcquireShared(arg) < 0)
            doAcquireSharedInterruptibly(arg);
    }

 

 

 /**
     *.
     *在中断模式中获取共享锁
     */
    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);
        }
    }

 

 

/**
 *
 * @author zhangwei_david
 * @version $Id: TestHarness.java, v 0.1 2014年11月10日 下午5:08:17 zhangwei_david Exp $
 */
public class TestHarness {

    public long timeTasks(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() {
                @Override
                public void run() {
                    try {
                        // 等待所有线程都准备好
                        startGate.await();
                        try {
                            task.run();
                        } finally {
                            // 线程执行结束，闭锁减一
                            endGate.countDown();
                        }
                    } catch (InterruptedException e1) {
                    }
                }
            };
            t.start();
        }
        long start = System.nanoTime();
        //// 所有的线程都准备好，可以执行
        startGate.countDown();
        // 等待所有线程执行结束
        endGate.await();

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

 

/**
 *
 * @author zhangwei_david
 * @version $Id: Test.java, v 0.1 2014年11月10日 下午5:13:24 zhangwei_david Exp $
 */
public class Test {
    public static void main(String[] args) throws InterruptedException {

        System.out.println(new TestHarness().timeTasks(10, new Runnable() {

            public void run() {
                System.out.println(Thread.currentThread().getName() + "开始执行时间："
                                   + System.currentTimeMillis());
            }
        }));
    }
}

 

   结果是：

Thread-2开始执行时间：1435904413836
Thread-9开始执行时间：1435904413836
Thread-5开始执行时间：1435904413836
Thread-7开始执行时间：1435904413836
Thread-6开始执行时间：1435904413836
Thread-3开始执行时间：1435904413836
Thread-8开始执行时间：1435904413836
Thread-1开始执行时间：1435904413836
Thread-0开始执行时间：1435904413836
Thread-4开始执行时间：1435904413836
1192112

 通过console控制台输出的结果可以看出：所有的线程几乎是在同一时刻开始执行的，这个就是闭锁的作用！

评论

1 楼 toknowme 2015-07-03  

闭锁还可以用于所有线程一起结束

http://toknowme.iteye.com/blog/2210399