Multithread framework 1/2

http://www.zhuaxia.com/item/590227619/
http://daoger.javaeye.com/blog/142485

java.util.concurrent 多線程框架 (上)

JDK5中的一個亮點就是將Doug Lea的並發庫引入到Java標準庫中。
一般的服務器都需要線程池，比如Web、FTP等服務器，不過它們一般都自己實現了線程池，比如以前介紹過的Tomcat、Resin和Jetty等，現在有了JDK5，我們就沒有必要重複造車輪了，直接使用就可以，何況使用也很方便，性能也非常高。

Java代碼 

1. package concurrent;  
2. import java.util.concurrent.ExecutorService;  
3. import java.util.concurrent.Executors;  
4. public class TestThreadPool {  
5. public static void main(String args[]) throws InterruptedException {  
6. // only two threads  
7. ExecutorService exec = Executors.newFixedThreadPool(2);  
8. for(int index = 0; index < 100; index++) {  
9. Runnable run = new Runnable() {  
10. public void run() {  
11. long time = (long) (Math.random() * 1000);  
12. System.out.println(“Sleeping ” + time + “ms”);  
13. try {  
14. Thread.sleep(time);  
15. } catch (InterruptedException e) {  
16. }  
17. }  
18. };  
19. exec.execute(run);  
20. }  
21. // must shutdown  
22. exec.shutdown();  
23. }  
24. }  

上面是一個簡單的例子，使用了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可以輕鬆的完成這些任務，而且非常的簡單。

Java代碼 

1. package concurrent;  
2. import static java.util.concurrent.TimeUnit.SECONDS;  
3. import java.util.Date;  
4. import java.util.concurrent.Executors;  
5. import java.util.concurrent.ScheduledExecutorService;  
6. import java.util.concurrent.ScheduledFuture;  
7. public class TestScheduledThread {  
8. public static void main(String[] args) {  
9. final ScheduledExecutorService scheduler = Executors  
10. .newScheduledThreadPool(2);  
11. final Runnable beeper = new Runnable() {  
12. int count = 0;  
13. public void run() {  
14. System.out.println(new Date() + ” beep ” + (++count));  
15. }  
16. };  
17. // 1秒鐘後運行，並每隔2秒運行一次  
18. final ScheduledFuture beeperHandle = scheduler.scheduleAtFixedRate(  
19. beeper, 1, 2, SECONDS);  
20. // 2秒鐘後運行，並每次在上次任務運行完後等待5秒後重新運行  
21. final ScheduledFuture beeperHandle2 = scheduler  
22. .scheduleWithFixedDelay(beeper, 2, 5, SECONDS);  
23. // 30秒後結束關閉任務，並且關閉Scheduler  
24. scheduler.schedule(new Runnable() {  
25. public void run() {  
26. beeperHandle.cancel(true);  
27. beeperHandle2.cancel(true);  
28. scheduler.shutdown();  
29. }  
30. }, 30, SECONDS);  
31. }  
32. }  

為了退出進程，上面的代碼中加入了關閉Scheduler的操作。 而對於24小時運行的應用而言，是沒有必要關閉Scheduler的。

在實際應用中，有時候需要多個線程同時工作以完成同一件事情，而且在完成過程中，往往會等待其他線程都完成某一階段後再執行，等所有線程都到達某一個階段後再統一執行。

比如有幾個旅行團需要途經深圳、廣州、韶關、長沙最後到達武漢。 旅行團中有自駕游的，有徒步的，有乘坐旅遊大巴的；這些旅行團同時出發，並且每到一個目的地，都要等待其他旅行團到達此地後再同時出發，直到都到達終點站武漢。

這時候CyclicBarrier就可以派上用場。 CyclicBarrier最重要的屬性就是參與者個數，另外最要方法是await()。 當所有線程都調用了await()後，就表示這些線程都可以繼續執行，否則就會等待。

Java代碼 

1. package concurrent;  
2. import java.text.SimpleDateFormat;  
3. import java.util.Date;  
4. import java.util.concurrent.BrokenBarrierException;  
5. import java.util.concurrent.CyclicBarrier;  
6. import java.util.concurrent.ExecutorService;  
7. import java.util.concurrent.Executors;  
8. public class TestCyclicBarrier {  
9. // 徒步需要的時間: Shenzhen, Guangzhou, Shaoguan, Changsha, Wuhan  
10. private static int[] timeWalk = { 5, 8, 15, 15, 10 };  
11. // 自駕游  
12. private static int[] timeSelf = { 1, 3, 4, 4, 5 };  
13. // 旅遊大巴  
14. private static int[] timeBus = { 2, 4, 6, 6, 7 };  
15.   
16. static String now() {  
17. SimpleDateFormat sdf = new SimpleDateFormat(“HH:mm:ss”);  
18. return sdf.format(new Date()) + “: “;  
19. }  
20.   
21. static class Tour implements Runnable {  
22. private int[] times;  
23. private CyclicBarrier barrier;  
24. private String tourName;  
25. public Tour(CyclicBarrier barrier, String tourName, int[] times) {  
26. this.times = times;  
27. this.tourName = tourName;  
28. this.barrier = barrier;  
29. }  
30. public void run() {  
31. try {  
32. Thread.sleep(times[0] * 1000);  
33. System.out.println(now() + tourName + ” Reached Shenzhen”);  
34. barrier.await();  
35. Thread.sleep(times[1] * 1000);  
36. System.out.println(now() + tourName + ” Reached Guangzhou”);  
37. barrier.await();  
38. Thread.sleep(times[2] * 1000);  
39. System.out.println(now() + tourName + ” Reached Shaoguan”);  
40. barrier.await();  
41. Thread.sleep(times[3] * 1000);  
42. System.out.println(now() + tourName + ” Reached Changsha”);  
43. barrier.await();  
44. Thread.sleep(times[4] * 1000);  
45. System.out.println(now() + tourName + ” Reached Wuhan”);  
46. barrier.await();  
47. } catch (InterruptedException e) {  
48. } catch (BrokenBarrierException e) {  
49. }  
50. }  
51. }  
52.   
53. public static void main(String[] args) {  
54. // 三個旅行團  
55. CyclicBarrier barrier = new CyclicBarrier(3);  
56. ExecutorService exec = Executors.newFixedThreadPool(3);  
57. exec.submit(new Tour(barrier, “WalkTour”, timeWalk));  
58. exec.submit(new Tour(barrier, “SelfTour”, timeSelf));  
59. exec.submit(new Tour(barrier, “BusTour”, timeBus));  
60. exec.shutdown();  
61. }  
62. }  

運行結果：
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操作會阻塞，為了使線程退出，特在隊列中添加了一個“標識”，算法中也叫“哨兵”，當發現這個哨兵後，寫線程就退出。

當然線程池也要顯式退出了。

Java代碼 

1. package concurrent;  
2. import java.io.File;  
3. import java.io.FileFilter;  
4. import java.util.concurrent.BlockingQueue;  
5. import java.util.concurrent.ExecutorService;  
6. import java.util.concurrent.Executors;  
7. import java.util.concurrent.LinkedBlockingQueue;  
8. import java.util.concurrent.atomic.AtomicInteger;  
9.   
10. public class TestBlockingQueue {  
11. static long randomTime() {  
12. return (long) (Math.random() * 1000);  
13. }  
14.   
15. public static void main(String[] args) {  
16. // 能容納100個文件  
17. final BlockingQueue queue = new LinkedBlockingQueue(100);  
18. // 線程池  
19. final ExecutorService exec = Executors.newFixedThreadPool(5);  
20. final File root = new File(“F:\\JavaLib”);  
21. // 完成標誌  
22. final File exitFile = new File(“”);  
23. // 讀個數  
24. final AtomicInteger rc = new AtomicInteger();  
25. // 寫個數  
26. final AtomicInteger wc = new AtomicInteger();  
27. // 讀線程  
28. Runnable read = new Runnable() {  
29. public void run() {  
30. scanFile(root);  
31. scanFile(exitFile);  
32. }  
33.   
34. public void scanFile(File file) {  
35. if (file.isDirectory()) {  
36. File[] files = file.listFiles(new FileFilter() {  
37. public boolean accept(File pathname) {  
38. return pathname.isDirectory()  
39. || pathname.getPath().endsWith(“.java”);  
40. }  
41. });  
42. for (File one : files)  
43. scanFile(one);  
44. } else {  
45. try {  
46. int index = rc.incrementAndGet();  
47. System.out.println(“Read0: ” + index + ” “  
48. + file.getPath());  
49. queue.put(file);  
50. } catch (InterruptedException e) {  
51. }  
52. }  
53. }  
54. };  
55. exec.submit(read);  
56. // 四個寫線程  
57. for (int index = 0; index < 4; index++) {  
58. // write thread  
59. final int NO = index;  
60. Runnable write = new Runnable() {  
61. String threadName = “Write” + NO;  
62. public void run() {  
63. while (true) {  
64. try {  
65. Thread.sleep(randomTime());  
66. int index = wc.incrementAndGet();  
67. File file = queue.take();  
68. // 隊列已經無對象  
69. if (file == exitFile) {  
70. // 再次添加”標誌”，以讓其他線程正常退出  
71. queue.put(exitFile);  
72. break;  
73. }  
74. System.out.println(threadName + “: ” + index + ” “  
75. + file.getPath());  
76. } catch (InterruptedException e) {  
77. }  
78. }  
79. }  
80. };  
81. exec.submit(write);  
82. }  
83. exec.shutdown();  
84. }  
85. }  

從名字可以看出，CountDownLatch是一個倒數計數的鎖，當倒數到0時觸發事件，也就是開鎖，其他人就可以進入了。 在一些應用場合中，需要等待某個條件達到要求後才能做後面的事情；同時當線程都完成後也會觸發事件，以便進行後面的操作。


CountDownLatch最重要的方法是countDown()和await()，前者主要是倒數一次，後者是等待倒數到0，如果沒有到達0，就只有阻塞等待了。

一個CountDouwnLatch實例是不能重複使用的，也就是說它是一次性的，鎖一經被打開就不能再關閉使用了，如果想重複使用，請考慮使用CyclicBarrier。

下面的例子簡單的說明了CountDownLatch的使用方法，模擬了100米賽跑，10名選手已經準備就緒，只等裁判一聲令下。 當所有人都到達終點時，比賽結束。

同樣，線程池需要顯式shutdown。

Java代碼 

1. package concurrent;  
2.   
3. import java.util.concurrent.CountDownLatch;  
4. import java.util.concurrent.ExecutorService;  
5. import java.util.concurrent.Executors;  
6.   
7. public class TestCountDownLatch {  
8. public static void main(String[] args) throws InterruptedException {  
9. // 開始的倒數鎖  
10. final CountDownLatch begin = new CountDownLatch(1);  
11. // 結束的倒數鎖  
12. final CountDownLatch end = new CountDownLatch(10);  
13. // 十名選手  
14. final ExecutorService exec = Executors.newFixedThreadPool(10);  
15. for(int index = 0; index < 10; index++) {  
16. final int NO = index + 1;  
17. Runnable run = new Runnable(){  
18. public void run() {  
19. try {  
20. begin.await();  
21. Thread.sleep((long) (Math.random() * 10000));  
22. System.out.println(“No.” + NO + ” arrived”);  
23. } catch (InterruptedException e) {  
24. } finally {  
25. end.countDown();  
26. }  
27. }  
28. };  
29. exec.submit(run);  
30. }  
31. System.out.println(“Game Start”);  
32. begin.countDown();  
33. end.await();  
34. System.out.println(“Game Over”);  
35. exec.shutdown();  
36. }  
37. }  

運行結果:
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