1. CountDownLatch的介紹
CountDownLatch是一個同步工具,它主要用線程執行之間的協作。CountDownLatch 的作用和 Thread.join() 方法類似,讓一些線程阻塞直到另一些線程完成一系列操作后才被喚醒。在直接創建線程的年代(Java 5.0 之前),我們可以使用 Thread.join()。在線程池出現后,因為線程池中的線程不能直接被引用,所以就必須使用 CountDownLatch 了。
CountDownLatch主要有兩個方法,當一個或多個線程調用await方法時,這些線程會阻塞。其它線程調用countDown方法會將計數器減1(調用countDown方法的線程不會阻塞),當計數器的值變為0時,因await方法阻塞的線程會被喚醒,繼續執行。
實現原理:計數器的值由構造函數傳入,並用它初始化AQS的state值。當線程調用await方法時會檢查state的值是否為0,如果是就直接返回(即不會阻塞);如果不是,將表示該節點的線程入列,然后將自身阻塞。當其它線程調用countDown方法會將計數器減1,然后判斷計數器的值是否為0,當它為0時,會喚醒隊列中的第一個節點,由於CountDownLatch使用了AQS的共享模式,所以第一個節點被喚醒后又會喚醒第二個節點,以此類推,使得所有因await方法阻塞的線程都能被喚醒而繼續執行。
從源代碼和實現原理中可以看出一個CountDownLatch對象,只能使用一次,不能重復使用。
await方法源碼
public void await() throws InterruptedException { sync.acquireSharedInterruptibly(1); } public final void acquireSharedInterruptibly(int 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; }
doAcquireSharedInterruptibly 主要實現線程的入列與阻塞。
countDown方法
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 (;;) { int c = getState(); if (c == 0) return false; int nextc = c-1; if (compareAndSetState(c, nextc)) return nextc == 0; } }
doReleaseShared主要實現喚醒第一個節點,第一個節點有會喚醒第二個節點,……。
2. 使用示例
package demo; import java.util.Random; import java.util.concurrent.CountDownLatch; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; public class CountDownLatchDemo { private CountDownLatch cdl = new CountDownLatch(2); private Random rnd = new Random(); class FirstTask implements Runnable{ private String id; public FirstTask(String id){ this.id = id; } @Override public void run(){ System.out.println("Thread "+ id + " is start"); try { Thread.sleep(rnd.nextInt(1000)); } catch (InterruptedException e) { e.printStackTrace(); } System.out.println("Thread "+ id + " is over"); cdl.countDown(); } } class SecondTask implements Runnable{ private String id; public SecondTask(String id){ this.id = id; } @Override public void run(){ try { cdl.await(); } catch (InterruptedException e) { e.printStackTrace(); } System.out.println("----------Thread "+ id + " is start"); try { Thread.sleep(rnd.nextInt(1000)); } catch (InterruptedException e) { e.printStackTrace(); } System.out.println("----------Thread "+ id + " is over"); } } public static void main(String[] args){ ExecutorService es = Executors.newCachedThreadPool(); CountDownLatchDemo cdld = new CountDownLatchDemo(); es.submit(cdld.new SecondTask("c")); es.submit(cdld.new SecondTask("d")); es.submit(cdld.new FirstTask("a")); es.submit(cdld.new FirstTask("b")); es.shutdown(); } }
在這個示例中,我們創建了四個線程a、b、c、d,這四個線程幾乎同時提交給了線程池。c線程和d線程會在a線程和b線程結束后開始執行。
運行結果
Thread a is start
Thread b is start
Thread b is over
Thread a is over
----------Thread c is start
----------Thread d is start
----------Thread d is over
----------Thread c is over
3. 參考內容
[1] http://developer.51cto.com/art/201403/432095.htm