Java concurrency之CountDownLatch原理和示例_動(dòng)力節(jié)點(diǎn)Java學(xué)院整理

// 使當(dāng)前線程在鎖存器倒計(jì)數(shù)至零之前一直等待，除非線程被中斷。
void await()
// 使當(dāng)前線程在鎖存器倒計(jì)數(shù)至零之前一直等待，除非線程被中斷或超出了指定的等待時(shí)間。
boolean await(long timeout, TimeUnit unit)
// 遞減鎖存器的計(jì)數(shù)，如果計(jì)數(shù)到達(dá)零，則釋放所有等待的線程。
void countDown()
// 返回當(dāng)前計(jì)數(shù)。
long getCount()
// 返回標(biāo)識(shí)此鎖存器及其狀態(tài)的字符串。
String toString()

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

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

protected final void setState(long newState) {
  state = newState;
}

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

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

protected int tryAcquireShared(int acquires) {
  return (getState() == 0) ? 1 : -1;
}

private void doAcquireSharedInterruptibly(long arg)
  throws InterruptedException {
  // 創(chuàng)建"當(dāng)前線程"的Node節(jié)點(diǎn)，且Node中記錄的鎖是"共享鎖"類型；并將該節(jié)點(diǎn)添加到CLH隊(duì)列末尾。
  final Node node = addWaiter(Node.SHARED);
  boolean failed = true;
  try {
    for (;;) {
      // 獲取上一個(gè)節(jié)點(diǎn)。
      // 如果上一節(jié)點(diǎn)是CLH隊(duì)列的表頭，則"嘗試獲取共享鎖"。
      final Node p = node.predecessor();
      if (p == head) {
        long r = tryAcquireShared(arg);
        if (r >= 0) {
          setHeadAndPropagate(node, r);
          p.next = null; // help GC
          failed = false;
          return;
        }
      }
      // (上一節(jié)點(diǎn)不是CLH隊(duì)列的表頭) 當(dāng)前線程一直等待，直到獲取到共享鎖。
      // 如果線程在等待過程中被中斷過，則再次中斷該線程(還原之前的中斷狀態(tài))。
      if (shouldParkAfterFailedAcquire(p, node) &&
        parkAndCheckInterrupt())
        throw new InterruptedException();
    }
  } finally {
    if (failed)
      cancelAcquire(node);
  }
}

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

public final boolean releaseShared(int arg) {
  if (tryReleaseShared(arg)) {
    doReleaseShared();
    return true;
  }
  return false;
}

protected boolean tryReleaseShared(int releases) {
  // Decrement count; signal when transition to zero
  for (;;) {
    // 獲取“鎖計(jì)數(shù)器”的狀態(tài)
    int c = getState();
    if (c == 0)
      return false;
    // “鎖計(jì)數(shù)器”-1
    int nextc = c-1;
    // 通過CAS函數(shù)進(jìn)行賦值。
    if (compareAndSetState(c, nextc))
      return nextc == 0;
  }
}

 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.CyclicBarrier;
 public class CountDownLatchTest1 {
   private static int LATCH_SIZE = 5;
   private static CountDownLatch doneSignal;
   public static void main(String[] args) {
     try {
       doneSignal = new CountDownLatch(LATCH_SIZE);
       // 新建5個(gè)任務(wù)
       for(int i=0; i<LATCH_SIZE; i++)
         new InnerThread().start();
       System.out.println("main await begin.");
       // "主線程"等待線程池中5個(gè)任務(wù)的完成
       doneSignal.await();
       System.out.println("main await finished.");
     } catch (InterruptedException e) {
       e.printStackTrace();
     }
   }
   static class InnerThread extends Thread{
     public void run() {
       try {
         Thread.sleep(1000);
         System.out.println(Thread.currentThread().getName() + " sleep 1000ms.");
         // 將CountDownLatch的數(shù)值減1
         doneSignal.countDown();
       } catch (InterruptedException e) {
         e.printStackTrace();
       }
     }
   }
 }

main await begin.
Thread-0 sleep 1000ms.
Thread-2 sleep 1000ms.
Thread-1 sleep 1000ms.
Thread-4 sleep 1000ms.
Thread-3 sleep 1000ms.
main await finished.