JDK5多線程框架java.util.concurrent

時間  2019-11-13
標籤 jdk5 jdk 多線程 框架 java.util.concurrent java util concurrent 欄目 Java 简体版
原文   原文鏈接

JDK5中的一個亮點就是將Doug Lea的併發庫引入到Java標準庫中。Doug Lea確實是一個牛人，能教書，能出書，能編碼，不過這在國外仍是比較廣泛的，而國內的教授們就相差太遠了。

通常的服務器都須要線程池，好比Web、FTP等服務器，不過它們通常都本身實現了線程池，好比之前介紹過的Tomcat、Resin和Jetty等，如今有了JDK5，咱們就沒有必要重複造車輪了，直接使用就能夠，況且使用也很方便，性能也很是高。java

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能夠輕鬆的完成這些任務，並且很是的簡單。web

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

有時候在實際應用中，某些操做很耗時，但又不是不可或缺的步驟。好比用網頁瀏覽器瀏覽新聞時，最重要的是要顯示文字內容，至於與新聞相匹配的圖片 就沒有那麼重要的，因此此時首先保證文字信息先顯示，而圖片信息會後顯示，但又不能不顯示，因爲下載圖片是一個耗時的操做，因此必須一開始就得下載。


Java的併發庫的Future類就能夠知足這個要求。Future的重要方法包括get()和cancel()，get()獲取數據對象，若是 數據沒有加載，就會阻塞直到取到數據，而 cancel()是取消數據加載。另一個get(timeout)操做，表示若是在timeout時間內沒有取到就失敗返回，而再也不阻塞。

下面的Demo簡單的說明了Future的使用方法：一個很是耗時的操做必須一開始啓動，但又不能一直等待；其餘重要的事情又必須作，等完成後，就能夠作不重要的事情。服務器

Java代碼 複製代碼
  1. package  concurrent;  
  2.   
  3. import  java.util.concurrent.Callable;  
  4. import  java.util.concurrent.ExecutionException;  
  5. import  java.util.concurrent.ExecutorService;  
  6. import  java.util.concurrent.Executors;  
  7. import  java.util.concurrent.Future;  
  8.   
  9. public   class  TestFutureTask {  
  10. public   static   void  main(String[] args) throws  InterruptedException,  
  11. ExecutionException {  
  12. final  ExecutorService exec = Executors.newFixedThreadPool( 5 );  
  13. Callable call = new  Callable() {  
  14. public  String call()  throws  Exception {  
  15. Thread.sleep(1000  *  5 );  
  16. return  「Other less important but longtime things.」;  
  17. }  
  18. };  
  19. Future task = exec.submit(call);  
  20. // 重要的事情   
  21. Thread.sleep(1000  *  3 );  
  22. System.out.println(「Let’s do  important things.」);  
  23. // 其餘不重要的事情   
  24. String obj = task.get();  
  25. System.out.println(obj);  
  26. // 關閉線程池   
  27. exec.shutdown();  
  28. }  
  29. }  


運行結果：
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對象。網絡

Java代碼 複製代碼
  1. package  concurrent;  
  2.   
  3. import  java.util.concurrent.Callable;  
  4. import  java.util.concurrent.CompletionService;  
  5. import  java.util.concurrent.ExecutionException;  
  6. import  java.util.concurrent.ExecutorCompletionService;  
  7. import  java.util.concurrent.ExecutorService;  
  8. import  java.util.concurrent.Executors;  
  9. import  java.util.concurrent.Future;  
  10.   
  11. public   class  TestCompletionService {  
  12. public   static   void  main(String[] args)  throws  InterruptedException,  
  13. ExecutionException {  
  14. ExecutorService exec = Executors.newFixedThreadPool(10 );  
  15. CompletionService serv =  
  16. new  ExecutorCompletionService(exec);  
  17.   
  18. for  ( int  index =  0 ; index <  5 ; index++) {  
  19. final   int  NO = index;  
  20. Callable downImg = new  Callable() {  
  21. public  String call()  throws  Exception {  
  22. Thread.sleep((long ) (Math.random() *  10000 ));  
  23. return  「Downloaded Image 」 + NO;  
  24. }  
  25. };  
  26. serv.submit(downImg);  
  27. }  
  28.   
  29. Thread.sleep(1000  *  2 );  
  30. System.out.println(「Show web content」);  
  31. for  ( int  index =  0 ; index <  5 ; index++) {  
  32. Future task = serv.take();  
  33. String img = task.get();  
  34. System.out.println(img);  
  35. }  
  36. System.out.println(「End」);  
  37. // 關閉線程池   
  38. exec.shutdown();  
  39. }  
  40. }  


運行結果:
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()獲取和釋放訪問許可。數據結構

Java代碼 複製代碼
  1. package  concurrent;  
  2.   
  3. import  java.util.concurrent.ExecutorService;  
  4. import  java.util.concurrent.Executors;  
  5. import  java.util.concurrent.Semaphore;  
  6.   
  7. public   class  TestSemaphore {  
  8. public   static   void  main(String[] args) {  
  9. // 線程池   
  10. ExecutorService exec = Executors.newCachedThreadPool();  
  11. // 只能5個線程同時訪問   
  12. final  Semaphore semp =  new  Semaphore( 5 );  
  13. // 模擬20個客戶端訪問   
  14. for  ( int  index =  0 ; index <  20 ; index++) {  
  15. final   int  NO = index;  
  16. Runnable run = new  Runnable() {  
  17. public   void  run() {  
  18. try  {  
  19. // 獲取許可   
  20. semp.acquire();  
  21. System.out.println(「Accessing: 」 + NO);  
  22. Thread.sleep((long ) (Math.random() *  10000 ));  
  23. // 訪問完後，釋放   
  24. semp.release();  
  25. } catch  (InterruptedException e) {  
  26. }  
  27. }  
  28. };  
  29. exec.execute(run);  
  30. }  
  31. // 退出線程池   
  32. exec.shutdown();  
  33. }  
  34. }  


運行結果：
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 併發



已有 0 人發表留言，猛擊->>這裏<<-參與討論


JavaEye推薦
相關文章
相關標籤/搜索
每日一句
    每一个你不满意的现在，都有一个你没有努力的曾经。
本站公眾號
   歡迎關注本站公眾號,獲取更多信息
聯繫我們 最近搜索 最新文章 沪ICP备13005482号-10 MyBatis教程 SQL 教程 MySQL教程 Java 教程 Thymeleaf 教程 Hibernate教程 Spring教程 Redis教程