?? JDK5中的一個亮點就是將Doug Lea的并發庫引入到Java標準庫中。Doug
Lea確實是一個牛人，能教書，能出書，能編碼，不過這在國外還是比較普遍的，而國內的教授們就相差太遠了。
??
一般的服務器都需要線程池，比如Web、FTP等服務器，不過它們一般都自己實現了線程池，比如以前介紹過的Tomcat、Resin和Jetty等，現在有了JDK5，我們就沒有必要重復造車輪了，直接使用就可以，何況使用也很方便，性能也非常高。
package?concurrent;??
import?java.util.concurrent.ExecutorService;??
import?java.util.concurrent.Executors;??
public?class?TestThreadPool?{??
public?static?void?main(String?args[])?throws?InterruptedException?{??
//?only?two?threads??
ExecutorService?exec?=?Executors.newFixedThreadPool(2);??
for(int?index?=?0;?index?<?100;?index++)?{??
?Runnable?run?=?new?Runnable()?{??
?public?void?run()?{??
long?time?=?(long)?(Math.random()?*?1000);??
System.out.println(“Sleeping?”?+?time?+?“ms”);??
?try?{??
Thread.sleep(time);??
}?catch?(InterruptedException?e)?{??
}??
}??
?};??
?exec.execute(run);??
}??
//?must?shutdown??
exec.shutdown();??
}??
}?
上面是一個簡單的例子，使用了2個大小的線程池來處理100個線程。但有一個問題：在for循環的過程中，會等待線程池有空閑的線程，所以主線程會阻塞
的。為了解決這個問題，一般啟動一個線程來做for循環，就是為了避免由于線程池滿了造成主線程阻塞。不過在這里我沒有這樣處理。[重要修正：經過測試，
即使線程池大小小于實際線程數大小，線程池也不會阻塞的，這與Tomcat的線程池不同，它將Runnable實例放到一個“無限”的
BlockingQueue中，所以就不用一個線程啟動for循環，Doug Lea果然厲害]
另外它使用了Executors的靜態函數生成一個固定的線程池，顧名思義，線程池的線程是不會釋放的，即使它是Idle。這就會產生性能問題，
比如如果線程池的大小為200，當全部使用完畢后，所有的線程會繼續留在池中，相應的內存和線程切換（while(true)+sleep循環）都會增
加。如果要避免這個問題，就必須直接使用ThreadPoolExecutor()來構造。可以像Tomcat的線程池一樣設置“最大線程數”、“最小線
程數”和“空閑線程keepAlive的時間”。通過這些可以基本上替換Tomcat的線程池實現方案。
需要注意的是線程池必須使用shutdown來顯式關閉，否則主線程就無法退出。shutdown也不會阻塞主線程。
許多長時間運行的應用有時候需要定時運行任務完成一些諸如統計、優化等工作，比如在電信行業中處理用戶話單時，需要每隔1分鐘處理話單；網站每天
凌晨統計用戶訪問量、用戶數；大型超時凌晨3點統計當天銷售額、以及最熱賣的商品；每周日進行數據庫備份；公司每個月的10號計算工資并進行轉帳等，這些
都是定時任務。通過 java的并發庫concurrent可以輕松的完成這些任務，而且非常的簡單。
package?concurrent;
import?static?java.util.concurrent.TimeUnit.SECONDS;
import?java.util.Date;
import?java.util.concurrent.Executors;
import?java.util.concurrent.ScheduledExecutorService;
import?java.util.concurrent.ScheduledFuture;
public?class?TestScheduledThread?{
public?static?void?main(String[]?args)?{
final?ScheduledExecutorService?scheduler?=?Executors
.newScheduledThreadPool(2);
final?Runnable?beeper?=?new?Runnable()?{
int?count?=?0;
public?void?run()?{
System.out.println(new?Date()?+?”?beep?”?+?(++count));
}
};
//?1秒鐘后運行，并每隔2秒運行一次
final?ScheduledFuture?beeperHandle?=?scheduler.scheduleAtFixedRate(
beeper,?1,?2,?SECONDS);
//?2秒鐘后運行，并每次在上次任務運行完后等待5秒后重新運行
final?ScheduledFuture?beeperHandle2?=?scheduler
.scheduleWithFixedDelay(beeper,?2,?5,?SECONDS);
//?30秒后結束關閉任務，并且關閉Scheduler
scheduler.schedule(new?Runnable()?{
public?void?run()?{
beeperHandle.cancel(true);
beeperHandle2.cancel(true);
scheduler.shutdown();
}
},?30,?SECONDS);
}
}
為了退出進程，上面的代碼中加入了關閉Scheduler的操作。而對于24小時運行的應用而言，是沒有必要關閉Scheduler的。
在實際應用中，有時候需要多個線程同時工作以完成同一件事情，而且在完成過程中，往往會等待其他線程都完成某一階段后再執行，等所有線程都到達某一個階段后再統一執行。
比如有幾個旅行團需要途經深圳、廣州、韶關、長沙最后到達武漢。旅行團中有自駕游的，有徒步的，有乘坐旅游大巴的；這些旅行團同時出發，并且每到一個目的地，都要等待其他旅行團到達此地后再同時出發，直到都到達終點站武漢。
這時候CyclicBarrier就可以派上用場。CyclicBarrier最重要的屬性就是參與者個數，另外最要方法是await()。當所有線程都調用了await()后，就表示這些線程都可以繼續執行，否則就會等待。
package?concurrent;
import?java.text.SimpleDateFormat;
import?java.util.Date;
import?java.util.concurrent.BrokenBarrierException;
import?java.util.concurrent.CyclicBarrier;
import?java.util.concurrent.ExecutorService;
import?java.util.concurrent.Executors;
public?class?TestCyclicBarrier?{
//?徒步需要的時間:?Shenzhen,?Guangzhou,?Shaoguan,?Changsha,?Wuhan
private?static?int[]?timeWalk?=?{?5,?8,?15,?15,?10?};
//?自駕游
private?static?int[]?timeSelf?=?{?1,?3,?4,?4,?5?};
//?旅游大巴
private?static?int[]?timeBus?=?{?2,?4,?6,?6,?7?};
static?String?now()?{
SimpleDateFormat?sdf?=?new?SimpleDateFormat(“HH:mm:ss”);
return?sdf.format(new?Date())?+?“:?“;
}
static?class?Tour?implements?Runnable?{
private?int[]?times;
private?CyclicBarrier?barrier;
private?String?tourName;
public?Tour(CyclicBarrier?barrier,?String?tourName,?int[]?times)?{
this.times?=?times;
this.tourName?=?tourName;
this.barrier?=?barrier;
}
public?void?run()?{
try?{
Thread.sleep(times[0]?*?1000);
System.out.println(now()?+?tourName?+?”?Reached?Shenzhen”);
barrier.await();
Thread.sleep(times[1]?*?1000);
System.out.println(now()?+?tourName?+?”?Reached?Guangzhou”);
barrier.await();
Thread.sleep(times[2]?*?1000);
System.out.println(now()?+?tourName?+?”?Reached?Shaoguan”);
barrier.await();
Thread.sleep(times[3]?*?1000);
System.out.println(now()?+?tourName?+?”?Reached?Changsha”);
barrier.await();
Thread.sleep(times[4]?*?1000);
System.out.println(now()?+?tourName?+?”?Reached?Wuhan”);
barrier.await();
}?catch?(InterruptedException?e)?{
}?catch?(BrokenBarrierException?e)?{
}
}
}
public?static?void?main(String[]?args)?{
//?三個旅行團
CyclicBarrier?barrier?=?new?CyclicBarrier(3);
ExecutorService?exec?=?Executors.newFixedThreadPool(3);
exec.submit(new?Tour(barrier,?“WalkTour”,?timeWalk));
exec.submit(new?Tour(barrier,?“SelfTour”,?timeSelf));
exec.submit(new?Tour(barrier,?“BusTour”,?timeBus));
exec.shutdown();
}
}
運行結果：
00:02:25: SelfTour Reached Shenzhen
00:02:25: BusTour Reached Shenzhen
00:02:27: WalkTour Reached Shenzhen
00:02:30: SelfTour Reached Guangzhou
00:02:31: BusTour Reached Guangzhou
00:02:35: WalkTour Reached Guangzhou
00:02:39: SelfTour Reached Shaoguan
00:02:41: BusTour Reached Shaoguan
并發庫中的BlockingQueue是一個比較好玩的類，顧名思義，就是阻塞隊列。該類主要提供了兩個方法put()和take()，前者將一
個對象放到隊列中，如果隊列已經滿了，就等待直到有空閑節點；后者從head取一個對象，如果沒有對象，就等待直到有可取的對象。
下面的例子比較簡單，一個讀線程，用于將要處理的文件對象添加到阻塞隊列中，另外四個寫線程用于取出文件對象，為了模擬寫操作耗時長的特點，特讓
線程睡眠一段隨機長度的時間。另外，該Demo也使用到了線程池和原子整型（AtomicInteger），AtomicInteger可以在并發情況下
達到原子化更新，避免使用了synchronized，而且性能非常高。由于阻塞隊列的put和take操作會阻塞，為了使線程退出，特在隊列中添加了一
個“標識”，算法中也叫“哨兵”，當發現這個哨兵后，寫線程就退出。
當然線程池也要顯式退出了。
package?concurrent;
import?java.io.File;
import?java.io.FileFilter;
import?java.util.concurrent.BlockingQueue;
import?java.util.concurrent.ExecutorService;
import?java.util.concurrent.Executors;
import?java.util.concurrent.LinkedBlockingQueue;
import?java.util.concurrent.atomic.AtomicInteger;
public?class?TestBlockingQueue?{
static?long?randomTime()?{
return?(long)?(Math.random()?*?1000);
}
public?static?void?main(String[]?args)?{
//?能容納100個文件
final?BlockingQueue?queue?=?new?LinkedBlockingQueue(100);
//?線程池
final?ExecutorService?exec?=?Executors.newFixedThreadPool(5);
final?File?root?=?new?File(“F:\\JavaLib”);
//?完成標志
final?File?exitFile?=?new?File(“”);
//?讀個數
final?AtomicInteger?rc?=?new?AtomicInteger();
//?寫個數
final?AtomicInteger?wc?=?new?AtomicInteger();
//?讀線程
Runnable?read?=?new?Runnable()?{
public?void?run()?{
scanFile(root);
scanFile(exitFile);
}
public?void?scanFile(File?file)?{
if?(file.isDirectory())?{
File[]?files?=?file.listFiles(new?FileFilter()?{
public?boolean?accept(File?pathname)?{
return?pathname.isDirectory()
||?pathname.getPath().endsWith(“.java”);
}
});
for?(File?one?:?files)
scanFile(one);
}?else?{
try?{
int?index?=?rc.incrementAndGet();
System.out.println(“Read0:?”?+?index?+?”?“
+?file.getPath());
queue.put(file);
}?catch?(InterruptedException?e)?{
}
}
}
};
exec.submit(read);
//?四個寫線程
for?(int?index?=?0;?index?<?4;?index++)?{
//?write?thread
final?int?NO?=?index;
Runnable?write?=?new?Runnable()?{
String?threadName?=?“Write”?+?NO;
public?void?run()?{
while?(true)?{
try?{
Thread.sleep(randomTime());
int?index?=?wc.incrementAndGet();
File?file?=?queue.take();
//?隊列已經無對象
if?(file?==?exitFile)?{
//?再次添加”標志”，以讓其他線程正常退出
queue.put(exitFile);
break;
}
System.out.println(threadName?+?“:?”?+?index?+?”?“
+?file.getPath());
}?catch?(InterruptedException?e)?{
}
}
}
};
exec.submit(write);
}
exec.shutdown();
}
}
從名字可以看出，CountDownLatch是一個倒數計數的鎖，當倒數到0時觸發事件，也就是開鎖，其他人就可以進入了。在一些應用場合中，需要等待某個條件達到要求后才能做后面的事情；同時當線程都完成后也會觸發事件，以便進行后面的操作。
CountDownLatch最重要的方法是countDown()和await()，前者主要是倒數一次，后者是等待倒數到0，如果沒有到達0，就只有阻塞等待了。
一個CountDouwnLatch實例是不能重復使用的，也就是說它是一次性的，鎖一經被打開就不能再關閉使用了，如果想重復使用，請考慮使用CyclicBarrier。
下面的例子簡單的說明了CountDownLatch的使用方法，模擬了100米賽跑，10名選手已經準備就緒，只等裁判一聲令下。當所有人都到達終點時，比賽結束。
同樣，線程池需要顯式shutdown。
package?concurrent;
import?java.util.concurrent.CountDownLatch;
import?java.util.concurrent.ExecutorService;
import?java.util.concurrent.Executors;
public?class?TestCountDownLatch?{
public?static?void?main(String[]?args)?throws?InterruptedException?{
//?開始的倒數鎖
final?CountDownLatch?begin?=?new?CountDownLatch(1);
//?結束的倒數鎖
final?CountDownLatch?end?=?new?CountDownLatch(10);
//?十名選手
final?ExecutorService?exec?=?Executors.newFixedThreadPool(10);
for(int?index?=?0;?index?<?10;?index++)?{
final?int?NO?=?index?+?1;
Runnable?run?=?new?Runnable(){
public?void?run()?{
try?{
begin.await();
Thread.sleep((long)?(Math.random()?*?10000));
System.out.println(“No.”?+?NO?+?”?arrived”);
}?catch?(InterruptedException?e)?{
}?finally?{
end.countDown();
}
}
};
exec.submit(run);
}
System.out.println(“Game?Start”);
begin.countDown();
end.await();
System.out.println(“Game?Over”);
exec.shutdown();
}
}
運行結果:
Game Start
No.4 arrived
No.1 arrived
No.7 arrived
No.9 arrived
No.3 arrived
No.2 arrived
No.8 arrived
No.10 arrived
No.6 arrived
No.5 arrived
Game Over
有時候在實際應用中，某些操作很耗時，但又不是不可或缺的步驟。比如用網頁瀏覽器瀏覽新聞時，最重要的是要顯示文字內容，至于與新聞相匹配的圖片
就沒有那么重要的，所以此時首先保證文字信息先顯示，而圖片信息會后顯示，但又不能不顯示，由于下載圖片是一個耗時的操作，所以必須一開始就得下載。
Java的并發庫的Future類就可以滿足這個要求。Future的重要方法包括get()和cancel()，get()獲取數據對象，如果
數據沒有加載，就會阻塞直到取到數據，而
cancel()是取消數據加載。另外一個get(timeout)操作，表示如果在timeout時間內沒有取到就失敗返回，而不再阻塞。
下面的Demo簡單的說明了Future的使用方法：一個非常耗時的操作必須一開始啟動，但又不能一直等待；其他重要的事情又必須做，等完成后，就可以做不重要的事情。
package?concurrent;
import?java.util.concurrent.Callable;
import?java.util.concurrent.ExecutionException;
import?java.util.concurrent.ExecutorService;
import?java.util.concurrent.Executors;
import?java.util.concurrent.Future;
public?class?TestFutureTask?{
public?static?void?main(String[]?args)throws?InterruptedException,
ExecutionException?{
final?ExecutorService?exec?=?Executors.newFixedThreadPool(5);
Callable?call?=?new?Callable()?{
public?String?call()?throws?Exception?{
Thread.sleep(1000?*?5);
return?“Other?less?important?but?longtime?things.”;
}
};
Future?task?=?exec.submit(call);
//?重要的事情
Thread.sleep(1000?*?3);
System.out.println(“Let’s?do?important?things.”);
//?其他不重要的事情
String?obj?=?task.get();
System.out.println(obj);
//?關閉線程池
exec.shutdown();
}
}
運行結果：
Let’s do important things.
Other less important but longtime things.
考慮以下場景：瀏覽網頁時，瀏覽器了5個線程下載網頁中的圖片文件，由于圖片大小、網站訪問速度等諸多因素的影響，完成圖片下載的時間就會有很大的不同。如果先下載完成的圖片就會被先顯示到界面上，反之，后下載的圖片就后顯示。
Java的并發庫的CompletionService可以滿足這種場景要求。該接口有兩個重要方法：submit()和take()。
submit用于提交一個runnable或者callable，一般會提交給一個線程池處理；而take就是取出已經執行完畢runnable或者
callable實例的Future對象，如果沒有滿足要求的，就等待了。
CompletionService還有一個對應的方法poll，該方法與take類似，只是不會等待，如果沒有滿足要求，就返回null對象。
package?concurrent;
import?java.util.concurrent.Callable;
import?java.util.concurrent.CompletionService;
import?java.util.concurrent.ExecutionException;
import?java.util.concurrent.ExecutorCompletionService;
import?java.util.concurrent.ExecutorService;
import?java.util.concurrent.Executors;
import?java.util.concurrent.Future;
public?class?TestCompletionService?{
public?static?void?main(String[]?args)?throws?InterruptedException,
ExecutionException?{
ExecutorService?exec?=?Executors.newFixedThreadPool(10);
CompletionService?serv?=
new?ExecutorCompletionService(exec);
for?(int?index?=?0;?index?<?5;?index++)?{
final?int?NO?=?index;
Callable?downImg?=?new?Callable()?{
public?String?call()?throws?Exception?{
Thread.sleep((long)?(Math.random()?*?10000));
return?“Downloaded?Image?”?+?NO;
}
};
serv.submit(downImg);
}
Thread.sleep(1000?*?2);
System.out.println(“Show?web?content”);
for?(int?index?=?0;?index?<?5;?index++)?{
Future?task?=?serv.take();
String?img?=?task.get();
System.out.println(img);
}
System.out.println(“End”);
//?關閉線程池
exec.shutdown();
}
}
運行結果:
Show web content
Downloaded Image 1
Downloaded Image 2
Downloaded Image 4
Downloaded Image 0
Downloaded Image 3
End
操作系統的信號量是個很重要的概念，在進程控制方面都有應用。Java并發庫的Semaphore可以很輕松完成信號量控制，Semaphore
可以控制某個資源可被同時訪問的個數，acquire()獲取一個許可，如果沒有就等待，而release()釋放一個許可。比如在Windows下可以
設置共享文件的最大客戶端訪問個數。
Semaphore維護了當前訪問的個數，提供同步機制，控制同時訪問的個數。在數據結構中鏈表可以保存“無限”的節點，用Semaphore可以實現有限大小的鏈表。另外重入鎖ReentrantLock也可以實現該功能，但實現上要負責些，代碼也要復雜些。
下面的Demo中申明了一個只有5個許可的Semaphore，而有20個線程要訪問這個資源，通過acquire()和release()獲取和釋放訪問許可。
package?concurrent;
import?java.util.concurrent.ExecutorService;
import?java.util.concurrent.Executors;
import?java.util.concurrent.Semaphore;
public?class?TestSemaphore?{
public?static?void?main(String[]?args)?{
//?線程池
ExecutorService?exec?=?Executors.newCachedThreadPool();
//?只能5個線程同時訪問
final?Semaphore?semp?=?new?Semaphore(5);
//?模擬20個客戶端訪問
for?(int?index?=?0;?index?<?20;?index++)?{
final?int?NO?=?index;
Runnable?run?=?new?Runnable()?{
public?void?run()?{
try?{
//?獲取許可
semp.acquire();
System.out.println(“Accessing:?”?+?NO);
Thread.sleep((long)?(Math.random()?*?10000));
//?訪問完后，釋放
semp.release();
}?catch?(InterruptedException?e)?{
}
}
};
exec.execute(run);
}
//?退出線程池
exec.shutdown();
}
}
運行結果：
Accessing: 0
Accessing: 1
Accessing: 2
Accessing: 3
Accessing: 4
Accessing: 5
Accessing: 6
Accessing: 7
Accessing: 8
Accessing: 9
Accessing: 10
Accessing: 11
Accessing: 12
Accessing: 13
Accessing: 14
Accessing: 15
Accessing: 16
Accessing: 17
Accessing: 18
Accessing: 19