CountDownLatch源码解析。

ゞ浴缸里的玫瑰 2021-11-04 19:56 445阅读 0赞

/**
     *一个同步辅助类，在完成一组正在其他线程中执行的操作之前，它允许一个或多个线程一直等待。 
     *用给定的计数 初始化 CountDownLatch。由于调用了 countDown() 方法，所以在当前计数到达零之前，await 方法会一直受阻塞。
     *之后，会释放所有等待的线程，await 的所有后续调用都将立即返回。这种现象只出现一次——计数无法被重置。如果需要重置计数，请考虑使用 CyclicBarrier。 
     *CountDownLatch 是一个通用同步工具，它有很多用途。将计数 1 初始化的 CountDownLatch 用作一个简单的开/关锁存器，
     *或入口：在通过调用 countDown() 的线程打开入口前，所有调用 await 的线程都一直在入口处等待。
     *用 N 初始化的 CountDownLatch 可以使一个线程在 N 个线程完成某项操作之前一直等待，或者使其在某项操作完成 N 次之前一直等待。 
     *CountDownLatch 的一个有用特性是，它不要求调用 countDown 方法的线程等到计数到达零时才继续，而在所有线程都能通过之前，它只是阻止任何线程继续通过一个 await。 
     *示例用法： 下面给出了两个类，其中一组 worker 线程使用了两个倒计数锁存器： 
     *	第一个类是一个启动信号，在 driver 为继续执行 worker 做好准备之前，它会阻止所有的 worker 继续执行。 
     *	第二个类是一个完成信号，它允许 driver 在完成所有 worker 之前一直等待。 
     *	class Driver { // ...
     *		void main() throws InterruptedException {
     *			CountDownLatch startSignal = new CountDownLatch(1);
     *			CountDownLatch doneSignal = new CountDownLatch(N);
     *			for (int i = 0; i < N; ++i) // create and start threads
     *				new Thread(new Worker(startSignal, doneSignal)).start();
     *			doSomethingElse();            // don't let run yet
     *			startSignal.countDown();      // let all threads proceed
     *			doSomethingElse();
     *			doneSignal.await();           // wait for all to finish
     *		}
     *	}
     *	class Worker implements Runnable {
     *		private final CountDownLatch startSignal;
     *		private final CountDownLatch doneSignal;
     *		Worker(CountDownLatch startSignal, CountDownLatch doneSignal) {
     *			this.startSignal = startSignal;
     *			this.doneSignal = doneSignal;
     *		}
     *		public void run() {
     *			try {
     *				startSignal.await();
     *				doWork();
     *				doneSignal.countDown();
     *			} catch (InterruptedException ex) {} // return;
     *		}
     *		void doWork() { ... }
     *	}
     *另一种典型用法是，将一个问题分成 N 个部分，用执行每个部分并让锁存器倒计数的 Runnable 来描述每个部分，
     *然后将所有 Runnable 加入到 Executor 队列。当所有的子部分完成后，协调线程就能够通过 await。
     *（当线程必须用这种方法反复倒计数时，可改为使用 CyclicBarrier。） 
     *	class Driver2 { // ...
     *		void main() throws InterruptedException {
     *			CountDownLatch doneSignal = new CountDownLatch(N);
     *			Executor e = ...
     *			for (int i = 0; i < N; ++i) // create and start threads
     *				e.execute(new WorkerRunnable(doneSignal, i));
     *			doneSignal.await();           // wait for all to finish
     *		}
     *	}
     *	class WorkerRunnable implements Runnable {
     *		private final CountDownLatch doneSignal;
     *		private final int i;
     *		WorkerRunnable(CountDownLatch doneSignal, int i) {
     *			this.doneSignal = doneSignal;
     *			this.i = i;
     *		}
     *		public void run() {
     *			try {
     *				doWork(i);
     *				doneSignal.countDown();
     *			} catch (InterruptedException ex) {} // return;
     *		}
     *		void doWork() { ... }
     *	}
     *内存一致性效果：线程中调用 countDown() 之前的操作 happen-before 紧跟在从另一个线程中对应 await() 成功返回的操作。
     */
    public class CountDownLatch {
    
    	/**
    	 * 倒计数锁存器的同步控制。
    	 * 使用AQS状态表示计数。
    	 *
    	 */
    	private static final class Sync extends AbstractQueuedSynchronizer {
            private static final long serialVersionUID = 4982264981922014374L;
    
            Sync(int count) {
                setState(count);// 设置同步状态的值。
            }
            /**
             * 返回同步状态的当前值。此操作具有 volatile 读的内存语义。
             * @return	当前状态值
             */
            int getCount() {
                return getState();	// 返回同步状态的当前值。
            }
            /**
             * 试图在共享模式下获取对象状态。
             * 覆写：AbstractQueuedSynchronizer的tryAcquireShared()方法
             * @param	acquires	acquire 参数。该值总是传递给 acquire 方法的那个值，或者是因某个条件等待而保存在条目上的值。该值是不间断的，并且可以表示任何内容。 
             * @return	在失败时返回负值；如果共享模式下的获取成功并且其后续共享模式下的获取可能够成功，则返回正值，在这种情况下，后续等待线程必须检查可用性。
             */
            protected int tryAcquireShared(int acquires) {
                return (getState() == 0) ? 1 : -1;
            }
    
            /**
             * 试图设置状态来反映共享模式下的一个释放。
             * 覆写：AbstractQueuedSynchronizer的tryReleaseShared()方法
             * @param	release 参数。该值总是传递给 release 方法的那个值，或者是因某个条件等待而保存在条目上的当前状态值。该值是不间断的，并且可以表示任何内容。
             */
            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;
        
        /**
         * 构造一个用给定计数初始化的 CountDownLatch。
         * @param count	在线程能通过 await() 之前，必须调用 countDown() 的次数
         */
        public CountDownLatch(int count) {
        	// 如果 count 为负
            if (count < 0) throw new IllegalArgumentException("count < 0");
            this.sync = new Sync(count);
        }
        
        /**
         * 使当前线程在锁存器倒计数至零之前一直等待，除非线程被中断。 
         * 如果当前计数为零，则此方法立即返回。
         *  
         * 如果当前计数大于零，则出于线程调度目的，将禁用当前线程，且在发生以下两种情况之一前，该线程将一直处于休眠状态： 
         * 由于调用 countDown() 方法，计数到达零；
         * 或者其他某个线程中断当前线程。
         *  
         * 如果当前线程： 
         * 	在进入此方法时已经设置了该线程的中断状态；
         * 	或者在等待时被中断， 
         * 则抛出 InterruptedException，并且清除当前线程的已中断状态。 
         * @throws InterruptedException	如果当前线程在等待时被中断
         */
        public void await() throws InterruptedException {
            sync.acquireSharedInterruptibly(1);
        }
        
        /**
         * 使当前线程在锁存器倒计数至零之前一直等待，除非线程被中断或超出了指定的等待时间。 
         * 如果当前计数为零，则此方法立刻返回 true 值。
         *  
         * 如果当前计数大于零，则出于线程调度目的，将禁用当前线程，且在发生以下三种情况之一前，该线程将一直处于休眠状态： 
         * 	由于调用 countDown() 方法，计数到达零；
         * 	或者其他某个线程中断当前线程；
         * 	或者已超出指定的等待时间。
         *  
         * 如果计数到达零，则该方法返回 true 值。 
         * 
         * 如果当前线程： 
         * 在进入此方法时已经设置了该线程的中断状态；
         * 或者在等待时被中断， 
         * 则抛出 InterruptedException，并且清除当前线程的已中断状态。 
         * 
         * 如果超出了指定的等待时间，则返回值为 false。如果该时间小于等于零，则此方法根本不会等待。
         * @param timeout	要等待的最长时间
         * @param unit	timeout 参数的时间单位。
         * @return	如果计数到达零，则返回 true；如果在计数到达零之前超过了等待时间，则返回 false 
         * @throws InterruptedException	如果当前线程在等待时被中断
         */
        public boolean await(long timeout, TimeUnit unit)
                throws InterruptedException {
                return sync.tryAcquireSharedNanos(1, unit.toNanos(timeout));
        }
        
        /**
         * 递减锁存器的计数，如果计数到达零，则释放所有等待的线程。 
         * 如果当前计数大于零，则将计数减少。如果新的计数为零，出于线程调度目的，将重新启用所有的等待线程。 
         * 如果当前计数等于零，则不发生任何操作。
         */
        public void countDown() {
            sync.releaseShared(1);
        }
        
        /**
         * 返回当前计数。 
         * 此方法通常用于调试和测试。
         * @return	当前计数
         */
        public long getCount() {
            return sync.getCount();
        }
        
        /**
         * 返回标识此锁存器及其状态的字符串。状态用括号括起来，包括字符串 "Count ="，后跟当前计数。
         * 覆写：Object中的toString方法
         * @return	标识此锁存器及其状态的字符串
         */
        public String toString() {
            return super.toString() + "[Count = " + sync.getCount() + "]";
        }
    }