經過實例對JAVA 多線程同步工具類進行總結

 

CountDownLatch 閉鎖：能夠延遲線程的進度，直到鎖到達終止狀態。閉鎖的做用至關於一扇門，在鎖到達終止狀態以前這扇門一直是關閉的。當鎖到達終止狀態時，容許全部線程經過。CountDownLatch 有一個初始值，經過調用 countDown 能夠減小該值，一直到 0 時到達終止狀態。

　　FutureTask 用於執行一個可返回結果的長任務，任務在單獨的線程中執行，其餘線程能夠用 get 方法取任務結果，若是任務還沒有完成，線程在 get 上阻塞。

　　Semaphore 用於控制同時訪問某資源，或同時執行某操做的線程數目。信號量有一個初始值便可以分配的信號量總數目。線程任務開始前先調用 acquire 取得信號量，任務結束後調用 release 釋放信號量。在 acquire 是若是沒有可用信號量，線程將阻塞在 acquire 上，直到其餘線程釋放一個信號量。

　　CyclicBarrier 柵欄用於多個線程屢次迭代時進行同步，在一輪任務中，任何線程完成任務後都在 barrier 上等待，直到全部其餘線程也完成任務，而後一塊兒釋放，同時進入下一輪迭代。

　　CountDownLatch 的例子：

　　[java]

　　import java.util.concurrent.CountDownLatch;

　　import java.util.concurrent.atomic.AtomicInteger;

　　public class DemoOfLatch {

　　// 利用閉鎖 CountDownLatch 控制主線程和子線程的同步

　　public static void main（String[] args） {

　　int numberOfThread = 5;

　　final CountDownLatch startLatch = new CountDownLatch（1）；                // 用於控制子線程開始

　　final CountDownLatch stopLatch = new CountDownLatch（numberOfThread）；    // 用於子線程計數

　　final AtomicInteger count = new AtomicInteger（0）；                       // 用於分配子線程惟一標識

　　System.out.println（"Main thread start…"）；

　　for （ int i=0; i<numberOfThread; i++ ） {

　　Thread thread = new Thread（ new Runnable（） {

　　@Override

　　public void run（） {

　　int tid = count.getAndIncrement（）；

　　try {

　　// 等代主線程打開啓動信號

　　startLatch.await（）；

　　System.out.printf（"Thread %d started…%n", tid）；

　　int duration = （int）（Math.random（） * 5000）；

　　Thread.sleep（duration）；

　　} catch （InterruptedException e） {

　　e.printStackTrace（）；

　　Thread.currentThread（）。interrupt（）；

　　} finally {

　　System.out.printf（"Thread %d stoped…%n", tid）；

　　// 線程終止前減小線程計數

　　stopLatch.countDown（）；

　　}

　　}

　　}）；

　　thread.start（）；

　　}

　　// 在放行子線程以前作點什麼別的事情

　　System.out.println（"Main thread do preparation work for child threads…"）；

　　try {

　　Thread.sleep（2000）；

　　} catch （InterruptedException e） {

　　e.printStackTrace（）；

　　}

　　// 打開閉鎖放行全部子線程

　　System.out.println（"Main thread let child threads go…"）；

　　startLatch.countDown（）；

　　try {

　　// 等待子線程計數降爲 0 即全部子線程執行完畢

　　System.out.println（"Main thread wait for all child threads…"）；

　　stopLatch.await（）；

　　} catch （InterruptedException e） {

　　e.printStackTrace（）；

　　}

　　System.out.println（"Main thread exit…"）；

　　}

　　}

FutureTask 的例子：

　　[java]

　　import java.util.concurrent.Callable;

　　import java.util.concurrent.ExecutionException;

　　import java.util.concurrent.FutureTask;

　　public class DemoOfFutureTask {

　　public static void main（String[] args） {

　　// 建立一個 Future Task 用於併發執行長任務

　　final FutureTask<Movie> future = new FutureTask<Movie>（ new Callable<Movie>（） {

　　@Override

　　public Movie call（） throws Exception {

　　System.out.println（"Future task started…"）；

　　Thread.sleep（5000）；

　　System.out.println（"Future task finished…"）；

　　return new Movie（"2012","Unknown"）；

　　}

　　}）；

　　// 在子線程中啓動任務

　　Thread thread = new Thread（future）；

　　thread.start（）；

　　// 主線程乾點別的事情

　　System.out.println（"Now let's do sth eles…"）；

　　try {

　　Thread.sleep（1000）；

　　} catch （InterruptedException e1） {

　　e1.printStackTrace（）；

　　}

　　// 主線程開始取結果

　　System.out.println（"Now wait for result of future task…"）；

　　try {

　　Movie res = future.get（）；

　　System.out.printf（"Result from task is name=%s, actor=%s", res.name, res.actor）；

　　} catch （InterruptedException e） {

　　e.printStackTrace（）；

　　} catch （ExecutionException e） {

　　e.printStackTrace（）；

　　}

　　}

　　public static class Movie {

　　final public String name;

　　final public String actor;

　　public Movie（String name, String actor） {

　　this.name = name;

　　this.actor = actor;

　　}

　　}

　　}

　　Semaphore 的例子：

　　[java]

　　import java.util.concurrent.Semaphore;

　　import java.util.concurrent.atomic.AtomicInteger;

　　public class DemoOfSemaphore {

　　/**

　　* @param args

　　*/

　　public static void main（String[] args） {

　　final int numOfThread = 5;

　　final AtomicInteger count = new AtomicInteger（0）；       // 用於分配惟一線程標識

　　final Semaphore semaphore = new Semaphore（numOfThread）； // 用於控制併發線程數目

　　for （int i = 0; i < 10; i++） {

　　Thread thread = new Thread（new Runnable（） {

　　@Override

　　public void run（） {

　　int tid = count.getAndIncrement（）；

　　try {

// 等待直到取得信號量

　　System.out.printf（"Thread %d wait on semaphore…%n", tid）；

　　semaphore.acquire（）；

　　// 取得信號量以後作點事情

　　System.out.printf（"Thread %d get semaphore…%n", tid）；

　　int duration = （int）（Math.random（） * 5000）；

　　Thread.sleep（duration）；

　　} catch （InterruptedException e） {

　　e.printStackTrace（）；

　　} finally {

　　// 作完後釋放信號量

　　System.out.printf（"Thread %d release semaphore…%n", tid）；

　　semaphore.release（）；

　　}

　　}

　　}）；

　　thread.start（）；

　　}

　　}

　　}

　　CyclicBarrier 的例子：

　　[java] view plaincopy

　　import java.util.concurrent.BrokenBarrierException;

　　import java.util.concurrent.CyclicBarrier;

　　public class DemoOfBarrier {

　　public static void main（String[] args） {

　　final int numOfThread = 2;

　　final int numOfIteration = 2;

　　// 建立一個用於線程同步的 Barrier 對象

　　final CyclicBarrier barrier = new CyclicBarrier（numOfThread,

　　new Runnable（） {

　　// 當全部線程到達 Barrier 後會執行這個任務

　　// 任務在第一個 到達 Barrier 的線程中執行

　　@Override

　　public void run（） {

　　long tid = Thread.currentThread（）。getId（）；

　　// 當全部線程完成一輪迭代以後作點清除/準備/提交工做

　　System.out.printf（"[%d] - All threads arrived barrier…%n", tid）；

　　try {

　　Thread.sleep（2000）；

　　} catch （InterruptedException e） {

　　e.printStackTrace（）；

　　}

　　System.out.printf（"[%d] - Clear work done…%n", tid）；

　　}

　　}）；

　　// 建立並啓動多個線程，他們在 Barrier 上同步

　　for （int i = 0; i < numOfThread; i++） {

　　Thread thread = new Thread（new Runnable（） {

　　@Override

　　public void run（） {

　　long tid = Thread.currentThread（）。getId（）；

　　for （ int k=0; k<numOfIteration; k++ ） {

　　try {

　　// 線程進行一輪迭代，作點事情

　　System.out.printf（"Thread %d start its work…%n", tid）；

　　long duration = （int）（Math.random（）*5000）；

　　Thread.sleep（duration）；

　　// 作完迭代後等待其餘線程完成迭代

　　System.out.printf（"Thread %d wait on barrier…%n", tid）；

　　int num = barrier.await（）；

　　// 顯示完成的順序

　　System.out.printf（"Thread %d pass barrier with order=%d…%n", tid, num）；

　　} catch （InterruptedException e） {

　　e.printStackTrace（）；

　　Thread.currentThread（）。interrupt（）；

　　} catch （BrokenBarrierException e） {

　　e.printStackTrace（）；

　　}

　　}

　　}

　　}）；

　　thread.start（）；

　　}

　　}

　　}