日韩无码专区无码一级三级片|91人人爱网站中日韩无码电影|厨房大战丰满熟妇|AV高清无码在线免费观看|另类AV日韩少妇熟女|中文日本大黄一级黄色片|色情在线视频免费|亚洲成人特黄a片|黄片wwwav色图欧美|欧亚乱色一区二区三区

RELATEED CONSULTING
相關(guān)咨詢
選擇下列產(chǎn)品馬上在線溝通
服務(wù)時間：8:30-17:00
你可能遇到了下面的問題
關(guān)閉右側(cè)工具欄

新聞中心

這里有您想知道的互聯(lián)網(wǎng)營銷解決方案
JDK7中的Fork/Join模式

介  紹

十載的保靖網(wǎng)站建設(shè)經(jīng)驗，針對設(shè)計、前端、開發(fā)、售后、文案、推廣等六對一服務(wù)，響應(yīng)快，48小時及時工作處理。全網(wǎng)營銷推廣的優(yōu)勢是能夠根據(jù)用戶設(shè)備顯示端的尺寸不同，自動調(diào)整保靖建站的顯示方式，使網(wǎng)站能夠適用不同顯示終端，在瀏覽器中調(diào)整網(wǎng)站的寬度，無論在任何一種瀏覽器上瀏覽網(wǎng)站，都能展現(xiàn)優(yōu)雅布局與設(shè)計，從而大程度地提升瀏覽體驗。成都創(chuàng)新互聯(lián)從事“保靖網(wǎng)站設(shè)計”,“保靖網(wǎng)站推廣”以來，每個客戶項目都認真落實執(zhí)行。

隨著多核芯片逐漸成為主流，大多數(shù)軟件開發(fā)人員不可避免地需要了解并行編程的知識。而同時，主流程序語言正在將越來越多的并行特性合并到標準庫或者語言本身之中。我們可以看到，JDK 在這方面同樣走在潮流的前方。在 JDK 標準版 5 中，由 Doug Lea 提供的并行框架成為了標準庫的一部分（JSR-166）。隨后，在 JDK 6 中，一些新的并行特性，例如并行 collection 框架，合并到了標準庫中（JSR-166x）。直到今天，盡管 Java SE 7 還沒有正式發(fā)布，一些并行相關(guān)的新特性已經(jīng)出現(xiàn)在 JSR-166y 中：

1.Fork/Join 模式；

2.TransferQueue，它繼承自 BlockingQueue 并能在隊列滿時阻塞“生產(chǎn)者”；

3.ArrayTasks/ListTasks，用于并行執(zhí)行某些數(shù)組/列表相關(guān)任務(wù)的類；

4.IntTasks/LongTasks/DoubleTasks，用于并行處理數(shù)字類型數(shù)組的工具類，提供了排序、查找、求和、求最小值、求最大值等功能；

其中，對 Fork/Join 模式的支持可能是對開發(fā)并行軟件來說最通用的新特性。在 JSR-166y 中，Doug Lea 實現(xiàn)ArrayTasks/ListTasks/IntTasks/LongTasks/DoubleTasks 時就大量的用到了 Fork/Join 模式。讀者還需要注意一點，因為 JDK 7 還沒有正式發(fā)布，因此本文涉及到的功能和發(fā)布版本有可能不一樣。

Fork/Join 模式有自己的適用范圍。如果一個應(yīng)用能被分解成多個子任務(wù)，并且組合多個子任務(wù)的結(jié)果就能夠獲得最終的答案，那么這個應(yīng)用就適合用 Fork/Join 模式來解決。圖 1 給出了一個 Fork/Join 模式的示意圖，位于圖上部的 Task 依賴于位于其下的 Task 的執(zhí)行，只有當所有的子任務(wù)都完成之后，調(diào)用者才能獲得 Task 0 的返回結(jié)果。

圖 1. Fork/Join 模式示意圖

可以說，F(xiàn)ork/Join 模式能夠解決很多種類的并行問題。通過使用 Doug Lea 提供的 Fork/Join 框架，軟件開發(fā)人員只需要關(guān)注任務(wù)的劃分和中間結(jié)果的組合就能充分利用并行平臺的優(yōu)良性能。其他和并行相關(guān)的諸多難于處理的問題，例如負載平衡、同步等，都可以由框架采用統(tǒng)一的方式解決。這樣，我們就能夠輕松地獲得并行的好處而避免了并行編程的困難且容易出錯的缺點。

使用 Fork/Join 模式

在開始嘗試 Fork/Join 模式之前，我們需要從 Doug Lea 主持的 Concurrency JSR-166 Interest Site 上下載 JSR-166y 的源代碼，并且我們還需要安裝最新版本的 JDK 6（下載網(wǎng)址請參閱 參考資源）。Fork/Join 模式的使用方式非常直觀。首先，我們需要編寫一個 ForkJoinTask 來完成子任務(wù)的分割、中間結(jié)果的合并等工作。隨后，我們將這個 ForkJoinTask 交給 ForkJoinPool 來完成應(yīng)用的執(zhí)行。

通常我們并不直接繼承 ForkJoinTask，它包含了太多的抽象方法。針對特定的問題，我們可以選擇 ForkJoinTask 的不同子類來完成任務(wù)。RecursiveAction 是 ForkJoinTask 的一個子類，它代表了一類最簡單的 ForkJoinTask：不需要返回值，當子任務(wù)都執(zhí)行完畢之后，不需要進行中間結(jié)果的組合。如果我們從 RecursiveAction 開始繼承，那么我們只需要重載 protected void compute() 方法。下面，我們來看看怎么為快速排序算法建立一個 ForkJoinTask 的子類：

清單 1. ForkJoinTask 的子類

 
 
 
 
    
  
  
  
  
  1. classSortTaskextendsRecursiveAction{  
    2. 
  
  
  
  
  2. finallong[]array;  
    3. 
  
  
  
  
  3. finalintlo;  
    4. 
  
  
  
  
  4. finalinthi;  
    5. 
  
  
  
  
  5. privateintTHRESHOLD=30;  
    6. 
  
  
  
  
  6.  
    7. 
  
  
  
  
  7. publicSortTask(long[]array){  
    8. 
  
  
  
  
  8. this.array=array;  
    9. 
  
  
  
  
  9. this.lo=0;  
    10. 
  
  
  
  
  10. this.hi=array.length-1;  
    11. 
  
  
  
  
  11. }  
    12. 
  
  
  
  
  12.  
    13. 
  
  
  
  
  13. publicSortTask(long[]array,intlo,inthi){  
    14. 
  
  
  
  
  14. this.array=array;  
    15. 
  
  
  
  
  15. this.lo=lo;  
    16. 
  
  
  
  
  16. this.hi=hi;  
    17. 
  
  
  
  
  17. }  
    18. 
  
  
  
  
  18.  
    19. 
  
  
  
  
  19. protectedvoidcompute(){  
    20. 
  
  
  
  
  20. if(hi-lo
  
  
  
  
  21. sequentiallySort(array,lo,hi);  
    22. 
  
  
  
  
  22. else{  
    23. 
  
  
  
  
  23. intpivot=partition(array,lo,hi);  
    24. 
  
  
  
  
  24. coInvoke(newSortTask(array,lo,pivot-1),newSortTask(array,  
    25. 
  
  
  
  
  25. pivot+1,hi));  
    26. 
  
  
  
  
  26. }  
    27. 
  
  
  
  
  27. }  
    28. 
  
  
  
  
  28.  
    29. 
  
  
  
  
  29. privateintpartition(long[]array,intlo,inthi){  
    30. 
  
  
  
  
  30. longx=array[hi];  
    31. 
  
  
  
  
  31. inti=lo-1;  
    32. 
  
  
  
  
  32. for(intj=lo;j
  
  
  
  
  33. if(array[j]<=x){  
    34. 
  
  
  
  
  34. i++;  
    35. 
  
  
  
  
  35. swap(array,i,j);  
    36. 
  
  
  
  
  36. }  
    37. 
  
  
  
  
  37. }  
    38. 
  
  
  
  
  38. swap(array,i+1,hi);  
    39. 
  
  
  
  
  39. returni+1;  
    40. 
  
  
  
  
  40. }  
    41. 
  
  
  
  
  41.  
    42. 
  
  
  
  
  42. privatevoidswap(long[]array,inti,intj){  
    43. 
  
  
  
  
  43. if(i!=j){  
    44. 
  
  
  
  
  44. longtemp=array[i];  
    45. 
  
  
  
  
  45. array[i]=array[j];  
    46. 
  
  
  
  
  46. array[j]=temp;  
    47. 
  
  
  
  
  47. }  
    48. 
  
  
  
  
  48. }  
    49. 
  
  
  
  
  49.  
    50. 
  
  
  
  
  50. privatevoidsequentiallySort(long[]array,intlo,inthi){  
    51. 
  
  
  
  
  51. Arrays.sort(array,lo,hi+1);  
    52. 
  
  
  
  
  52. }  
    53. 
  
  
  
  
  53. } 
    54.

在清單1中，SortTask 首先通過 partition() 方法將數(shù)組分成兩個部分。隨后，兩個子任務(wù)將被生成并分別排序數(shù)組的兩個部分。當子任務(wù)足夠小時，再將其分割為更小的任務(wù)反而引起性能的降低。因此，這里我們使用一個 THRESHOLD，限定在子任務(wù)規(guī)模較小時，使用直接排序，而不是再將其分割成為更小的任務(wù)。其中，我們用到了 RecursiveAction 提供的方法 coInvoke()。它表示：啟動所有的任務(wù)，并在所有任務(wù)都正常結(jié)束后返回。如果其中一個任務(wù)出現(xiàn)異常，則其它所有的任務(wù)都取消。coInvoke() 的參數(shù)還可以是任務(wù)的數(shù)組。

現(xiàn)在剩下的工作就是將 SortTask 提交到 ForkJoinPool 了。ForkJoinPool() 默認建立具有與 CPU 可使用線程數(shù)相等線程個數(shù)的線程池。我們在一個 JUnit 的 test 方法中將 SortTask 提交給一個新建的 ForkJoinPool：

清單 2. 新建的 ForkJoinPool

 
 
 
 
    
  
  
  
  
  1. @Test 
    2. 
  
  
  
  
  2. publicvoidtestSort()throwsException{  
    3. 
  
  
  
  
  3. ForkJoinTasksort=newSortTask(array);  
    4. 
  
  
  
  
  4. ForkJoinPoolfjpool=newForkJoinPool();  
    5. 
  
  
  
  
  5. fjpool.submit(sort);  
    6. 
  
  
  
  
  6. fjpool.shutdown();  
    7. 
  
  
  
  
  7.  
    8. 
  
  
  
  
  8. fjpool.awaitTermination(30,TimeUnit.SECONDS);  
    9. 
  
  
  
  
  9.  
    10. 
  
  
  
  
  10. assertTrue(checkSorted(array));  
    11. 
  
  
  
  
  11. }  
    12.

在上面的代碼中，我們用到了 ForkJoinPool 提供的如下函數(shù)：

1. submit()：將 ForkJoinTask 類的對象提交給 ForkJoinPool，F(xiàn)orkJoinPool 將立刻開始執(zhí)行 ForkJoinTask。

2. shutdown()：執(zhí)行此方法之后，F(xiàn)orkJoinPool 不再接受新的任務(wù)，但是已經(jīng)提交的任務(wù)可以繼續(xù)執(zhí)行。如果希望立刻停止所有的任務(wù)，可以嘗試 shutdownNow() 方法。

3. awaitTermination()：阻塞當前線程直到 ForkJoinPool 中所有的任務(wù)都執(zhí)行結(jié)束。

并行快速排序的完整代碼如下所示：

清單 3. 并行快速排序的完整代碼

 
 
 
 
    
  
  
  
  
  1. packagetests;  
    2. 
  
  
  
  
  2. importstaticorg.junit.Assert.*;  
    3. 
  
  
  
  
  3. importjava.util.Arrays;  
    4. 
  
  
  
  
  4. importjava.util.Random;  
    5. 
  
  
  
  
  5. importjava.util.concurrent.TimeUnit;  
    6. 
  
  
  
  
  6. importjsr166y.forkjoin.ForkJoinPool;  
    7. 
  
  
  
  
  7. importjsr166y.forkjoin.ForkJoinTask;  
    8. 
  
  
  
  
  8. importjsr166y.forkjoin.RecursiveAction;  
    9. 
  
  
  
  
  9. importorg.junit.Before;  
    10. 
  
  
  
  
  10. importorg.junit.Test;  
    11. 
  
  
  
  
  11. classSortTaskextendsRecursiveAction{  
    12. 
  
  
  
  
  12. finallong[]array;  
    13. 
  
  
  
  
  13. finalintlo;  
    14. 
  
  
  
  
  14. finalinthi;  
    15. 
  
  
  
  
  15. privateintTHRESHOLD=0;//Fordemoonly  
    16. 
  
  
  
  
  16. publicSortTask(long[]array){  
    17. 
  
  
  
  
  17. this.array=array;  
    18. 
  
  
  
  
  18. this.lo=0;  
    19. 
  
  
  
  
  19. this.hi=array.length-1;  
    20. 
  
  
  
  
  20. }  
    21. 
  
  
  
  
  21. publicSortTask(long[]array,intlo,inthi){  
    22. 
  
  
  
  
  22. this.array=array;  
    23. 
  
  
  
  
  23. this.lo=lo;  
    24. 
  
  
  
  
  24. this.hi=hi;  
    25. 
  
  
  
  
  25. }  
    26. 
  
  
  
  
  26. protectedvoidcompute(){  
    27. 
  
  
  
  
  27. if(hi-lo
  
  
  
  
  28. sequentiallySort(array,lo,hi);  
    29. 
  
  
  
  
  29. else{  
    30. 
  
  
  
  
  30. intpivot=partition(array,lo,hi);  
    31. 
  
  
  
  
  31. System.out.println(" pivot="+pivot+",low="+lo+",high="+hi);  
    32. 
  
  
  
  
  32. System.out.println("array"+Arrays.toString(array));  
    33. 
  
  
  
  
  33. coInvoke(newSortTask(array,lo,pivot-1),newSortTask(array,  
    34. 
  
  
  
  
  34. pivot+1,hi));  
    35. 
  
  
  
  
  35. }  
    36. 
  
  
  
  
  36. }  
    37. 
  
  
  
  
  37. privateintpartition(long[]array,intlo,inthi){  
    38. 
  
  
  
  
  38. longx=array[hi];  
    39. 
  
  
  
  
  39. inti=lo-1;  
    40. 
  
  
  
  
  40. for(intj=lo;j
  
  
  
  
  41. if(array[j]<=x){  
    42. 
  
  
  
  
  42. i++;  
    43. 
  
  
  
  
  43. swap(array,i,j);  
    44. 
  
  
  
  
  44. }  
    45. 
  
  
  
  
  45. }  
    46. 
  
  
  
  
  46. swap(array,i+1,hi);  
    47. 
  
  
  
  
  47. returni+1;  
    48. 
  
  
  
  
  48. }  
    49. 
  
  
  
  
  49. privatevoidswap(long[]array,inti,intj){  
    50. 
  
  
  
  
  50. if(i!=j){  
    51. 
  
  
  
  
  51. longtemp=array[i];  
    52. 
  
  
  
  
  52. array[i]=array[j];  
    53. 
  
  
  
  
  53. array[j]=temp;  
    54. 
  
  
  
  
  54. }  
    55. 
  
  
  
  
  55. }  
    56. 
  
  
  
  
  56. privatevoidsequentiallySort(long[]array,intlo,inthi){  
    57. 
  
  
  
  
  57. Arrays.sort(array,lo,hi+1);  
    58. 
  
  
  
  
  58. }  
    59. 
  
  
  
  
  59. }  
    60. 
  
  
  
  
  60. publicclassTestForkJoinSimple{  
    61. 
  
  
  
  
  61. privatestaticfinalintNARRAY=16;//Fordemoonly  
    62. 
  
  
  
  
  62. long[]array=newlong[NARRAY];  
    63. 
  
  
  
  
  63. Randomrand=newRandom();  
    64. 
  
  
  
  
  64. @Before 
    65. 
  
  
  
  
  65. publicvoidsetUp(){  
    66. 
  
  
  
  
  66. for(inti=0;i
  
  
  
  
  67. array[i]=rand.nextLong()%100;//Fordemoonly  
    68. 
  
  
  
  
  68. }  
    69. 
  
  
  
  
  69. System.out.println("InitialArray:"+Arrays.toString(array));  
    70. 
  
  
  
  
  70. }  
    71. 
  
  
  
  
  71. @Test 
    72. 
  
  
  
  
  72. publicvoidtestSort()throwsException{  
    73. 
  
  
  
  
  73. ForkJoinTasksort=newSortTask(array);  
    74. 
  
  
  
  
  74. ForkJoinPoolfjpool=newForkJoinPool();  
    75. 
  
  
  
  
  75. fjpool.submit(sort);  
    76. 
  
  
  
  
  76. fjpool.shutdown();  
    77. 
  
  
  
  
  77. fjpool.awaitTermination(30,TimeUnit.SECONDS);  
    78. 
  
  
  
  
  78. assertTrue(checkSorted(array));  
    79. 
  
  
  
  
  79. }  
    80. 
  
  
  
  
  80. booleancheckSorted(long[]a){  
    81. 
  
  
  
  
  81. for(inti=0;i
  
  
  
  
  82. if(a[i]>(a[i+1])){  
    83. 
  
  
  
  
  83. returnfalse;  
    84. 
  
  
  
  
  84. }  
    85. 
  
  
  
  
  85. }  
    86. 
  
  
  
  
  86. returntrue;  
    87. 
  
  
  
  
  87. }  
    88. 
  
  
  
  
  88. }  
    89.

運行以上代碼，我們可以得到以下結(jié)果：

 
 
 
 
    
  
  
  
  
  1. InitialArray:[46,-12,74,-67,76,-13,-91,-96]  
    2. 
  
  
  
  
  2.  
    3. 
  
  
  
  
  3. pivot=0,low=0,high=7 
    4. 
  
  
  
  
  4. array[-96,-12,74,-67,76,-13,-91,46]  
    5. 
  
  
  
  
  5.  
    6. 
  
  
  
  
  6. pivot=5,low=1,high=7 
    7. 
  
  
  
  
  7. array[-96,-12,-67,-13,-91,46,76,74]  
    8. 
  
  
  
  
  8.  
    9. 
  
  
  
  
  9. pivot=1,low=1,high=4 
    10. 
  
  
  
  
  10. array[-96,-91,-67,-13,-12,46,74,76]  
    11. 
  
  
  
  
  11.  
    12. 
  
  
  
  
  12. pivot=4,low=2,high=4 
    13. 
  
  
  
  
  13. array[-96,-91,-67,-13,-12,46,74,76]  
    14. 
  
  
  
  
  14.  
    15. 
  
  
  
  
  15. pivot=3,low=2,high=3 
    16. 
  
  
  
  
  16. array[-96,-91,-67,-13,-12,46,74,76]  
    17. 
  
  
  
  
  17.  
    18. 
  
  
  
  
  18. pivot=2,low=2,high=2 
    19. 
  
  
  
  
  19. array[-96,-91,-67,-13,-12,46,74,76]  
    20. 
  
  
  
  
  20.  
    21. 
  
  
  
  
  21. pivot=6,low=6,high=7 
    22. 
  
  
  
  
  22. array[-96,-91,-67,-13,-12,46,74,76]  
    23. 
  
  
  
  
  23.  
    24. 
  
  
  
  
  24. pivot=7,low=7,high=7 
    25. 
  
  
  
  
  25. array[-96,-91,-67,-13,-12,46,74,76]  
    26.

#p#

Fork/Join 模式高級特性

使用 RecursiveTask

除了 RecursiveAction，F(xiàn)ork/Join 框架還提供了其他 ForkJoinTask 子類：帶有返回值的 RecursiveTask，使用 finish() 方法顯式中止的 AsyncAction 和 LinkedAsyncAction，以及可使用 TaskBarrier 為每個任務(wù)設(shè)置不同中止條件的 CyclicAction。

從 RecursiveTask 繼承的子類同樣需要重載 protected void compute() 方法。與 RecursiveAction 稍有不同的是，它可使用泛型指定一個返回值的類型。下面，我們來看看如何使用 RecursiveTask 的子類。

清單 4. RecursiveTask 的子類

 
 
 
 
    
  
  
  
  
  1. classFibonacciextendsRecursiveTask{  
    2. 
  
  
  
  
  2. finalintn;  
    3. 
  
  
  
  
  3.  
    4. 
  
  
  
  
  4. Fibonacci(intn){  
    5. 
  
  
  
  
  5. this.n=n;  
    6. 
  
  
  
  
  6. }  
    7. 
  
  
  
  
  7.  
    8. 
  
  
  
  
  8. privateintcompute(intsmall){  
    9. 
  
  
  
  
  9. finalint[]results={1,1,2,3,5,8,13,21,34,55,89};  
    10. 
  
  
  
  
  10. returnresults[small];  
    11. 
  
  
  
  
  11. }  
    12. 
  
  
  
  
  12.  
    13. 
  
  
  
  
  13. publicIntegercompute(){  
    14. 
  
  
  
  
  14. if(n<=10){  
    15. 
  
  
  
  
  15. returncompute(n);  
    16. 
  
  
  
  
  16. }  
    17. 
  
  
  
  
  17. Fibonaccif1=newFibonacci(n-1);  
    18. 
  
  
  
  
  18. Fibonaccif2=newFibonacci(n-2);  
    19. 
  
  
  
  
  19. f1.fork();  
    20. 
  
  
  
  
  20. f2.fork();  
    21. 
  
  
  
  
  21. returnf1.join()+f2.join();  
    22. 
  
  
  
  
  22. }  
    23. 
  
  
  
  
  23. } 
    24.

在清單4 中，F(xiàn)ibonacci 的返回值為 Integer 類型。其 compute() 函數(shù)首先建立兩個子任務(wù)，啟動子任務(wù)執(zhí)行，阻塞以等待子任務(wù)的結(jié)果返回，相加后得到最終結(jié)果。同樣，當子任務(wù)足夠小時，通過查表得到其結(jié)果，以減小因過多地分割任務(wù)引起的性能降低。其中，我們用到了 RecursiveTask 提供的方法 fork() 和 join()。它們分別表示：子任務(wù)的異步執(zhí)行和阻塞等待結(jié)果完成。

現(xiàn)在剩下的工作就是將 Fibonacci 提交到 ForkJoinPool 了，我們在一個 JUnit 的 test 方法中作了如下處理：

清單 5. 將 Fibonacci 提交到 ForkJoinPool

 
 
 
 
    
  
  
  
  
  1. @Test 
    2. 
  
  
  
  
  2. publicvoidtestFibonacci()throwsInterruptedException,ExecutionException{  
    3. 
  
  
  
  
  3. ForkJoinTaskfjt=newFibonacci(45);  
    4. 
  
  
  
  
  4. ForkJoinPoolfjpool=newForkJoinPool();  
    5. 
  
  
  
  
  5. Futureresult=fjpool.submit(fjt);  
    6. 
  
  
  
  
  6.  
    7. 
  
  
  
  
  7. //dosomething  
    8. 
  
  
  
  
  8. System.out.println(result.get());  
    9. 
  
  
  
  
  9. }  
    10.

使用 CyclicAction 來處理循環(huán)任務(wù)

CyclicAction 的用法稍微復雜一些。如果一個復雜任務(wù)需要幾個線程協(xié)作完成，并且線程之間需要在某個點等待所有其他線程到達，那么我們就能方便的用 CyclicAction 和 TaskBarrier 來完成。圖 2 描述了使用 CyclicAction 和 TaskBarrier 的一個典型場景。

圖 2. 使用 CyclicAction 和 TaskBarrier 執(zhí)行多線程任務(wù)

繼承自 CyclicAction 的子類需要 TaskBarrier 為每個任務(wù)設(shè)置不同的中止條件。從 CyclicAction 繼承的子類需要重載 protected void compute() 方法，定義在 barrier 的每個步驟需要執(zhí)行的動作。compute() 方法將被反復執(zhí)行直到 barrier 的 isTerminated() 方法返回 True。TaskBarrier 的行為類似于 CyclicBarrier。下面，我們來看看如何使用 CyclicAction 的子類。

清單 6. 使用 CyclicAction 的子類

 
 
 
 
    
  
  
  
  
  1. classConcurrentPrintextendsRecursiveAction{  
    2. 
  
  
  
  
  2. protectedvoidcompute(){  
    3. 
  
  
  
  
  3. TaskBarrierb=newTaskBarrier(){  
    4. 
  
  
  
  
  4. protectedbooleanterminate(intcycle,intregisteredParties){  
    5. 
  
  
  
  
  5. System.out.println("Cycleis"+cycle+";" 
    6. 
  
  
  
  
  6. +registeredParties+"parties");  
    7. 
  
  
  
  
  7. returncycle>=10;  
    8. 
  
  
  
  
  8. }  
    9. 
  
  
  
  
  9. };  
    10. 
  
  
  
  
  10. intn=3;  
    11. 
  
  
  
  
  11. CyclicAction[]actions=newCyclicAction[n];  
    12. 
  
  
  
  
  12. for(inti=0;i
  
  
  
  
  13. finalintindex=i;  
    14. 
  
  
  
  
  14. actions[i]=newCyclicAction(b){  
    15. 
  
  
  
  
  15. protectedvoidcompute(){  
    16. 
  
  
  
  
  16. System.out.println("I'mworking"+getCycle()+"" 
    17. 
  
  
  
  
  17. +index);  
    18. 
  
  
  
  
  18. try{  
    19. 
  
  
  
  
  19. Thread.sleep(500);  
    20. 
  
  
  
  
  20. }catch(InterruptedExceptione){  
    21. 
  
  
  
  
  21. e.printStackTrace();  
    22. 
  
  
  
  
  22. }  
    23. 
  
  
  
  
  23. }  
    24. 
  
  
  
  
  24. };  
    25. 
  
  
  
  
  25. }  
    26. 
  
  
  
  
  26. for(inti=0;i
  
  
  
  
  27. actions[i].fork();  
    28. 
  
  
  
  
  28. for(inti=0;i
  
  
  
  
  29. actions[i].join();  
    30. 
  
  
  
  
  30. }  
    31. 
  
  
  
  
  31. }  
    32.

在清單6中，CyclicAction[] 數(shù)組建立了三個任務(wù)，打印各自的工作次數(shù)和序號。而在 b.terminate() 方法中，我們設(shè)置的中止條件表示重復 10 次計算后中止?，F(xiàn)在剩下的工作就是將 ConcurrentPrint 提交到 ForkJoinPool 了。我們可以在 ForkJoinPool 的構(gòu)造函數(shù)中指定需要的線程數(shù)目，例如 ForkJoinPool(4) 就表明線程池包含 4 個線程。我們在一個 JUnit 的 test 方法中運行 ConcurrentPrint 的這個循環(huán)任務(wù)：

清單 7. 運行 ConcurrentPrint 循環(huán)任務(wù)

 
 
 
 
    
  
  
  
  
  1. @Test 
    2. 
  
  
  
  
  2. publicvoidtestBarrier()throwsInterruptedException,ExecutionException{  
    3. 
  
  
  
  
  3. ForkJoinTaskfjt=newConcurrentPrint();  
    4. 
  
  
  
  
  4. ForkJoinPoolfjpool=newForkJoinPool(4);  
    5. 
  
  
  
  
  5. fjpool.submit(fjt);  
    6. 
  
  
  
  
  6. fjpool.shutdown();  
    7. 
  
  
  
  
  7. } 
    8.

RecursiveTask 和 CyclicAction 兩個例子的完整代碼如下所示：

清單 8. RecursiveTask 和 CyclicAction 兩個例子的完整代碼

 
 
 
 
    
  
  
  
  
  1. packagetests;  
    2. 
  
  
  
  
  2.  
    3. 
  
  
  
  
  3. importjava.util.concurrent.ExecutionException;  
    4. 
  
  
  
  
  4. importjava.util.concurrent.Future;  
    5. 
  
  
  
  
  5.  
    6. 
  
  
  
  
  6. importjsr166y.forkjoin.CyclicAction;  
    7. 
  
  
  
  
  7. importjsr166y.forkjoin.ForkJoinPool;  
    8. 
  
  
  
  
  8. importjsr166y.forkjoin.ForkJoinTask;  
    9. 
  
  
  
  
  9. importjsr166y.forkjoin.RecursiveAction;  
    10. 
  
  
  
  
  10. importjsr166y.forkjoin.RecursiveTask;  
    11. 
  
  
  
  
  11. importjsr166y.forkjoin.TaskBarrier;  
    12. 
  
  
  
  
  12.  
    13. 
  
  
  
  
  13. importorg.junit.Test;  
    14. 
  
  
  
  
  14.  
    15. 
  
  
  
  
  15. classFibonacciextendsRecursiveTask{  
    16. 
  
  
  
  
  16. finalintn;  
    17. 
  
  
  
  
  17.  
    18. 
  
  
  
  
  18. Fibonacci(intn){  
    19. 
  
  
  
  
  19. this.n=n;  
    20. 
  
  
  
  
  20. }  
    21. 
  
  
  
  
  21.  
    22. 
  
  
  
  
  22. privateintcompute(intsmall){  
    23. 
  
  
  
  
  23. finalint[]results={1,1,2,3,5,8,13,21,34,55,89};  
    24. 
  
  
  
  
  24. returnresults[small];  
    25. 
  
  
  
  
  25. }  
    26. 
  
  
  
  
  26.  
    27. 
  
  
  
  
  27. publicIntegercompute(){  
    28. 
  
  
  
  
  28. if(n<=10){  
    29. 
  
  
  
  
  29. returncompute(n);  
    30. 
  
  
  
  
  30. }  
    31. 
  
  
  
  
  31. Fibonaccif1=newFibonacci(n-1);  
    32. 
  
  
  
  
  32. Fibonaccif2=newFibonacci(n-2);  
    33. 
  
  
  
  
  33. System.out.println("forknewthreadfor"+(n-1));  
    34. 
  
  
  
  
  34. f1.fork();  
    35. 
  
  
  
  
  35. System.out.println("forknewthreadfor"+(n-2));  
    36. 
  
  
  
  
  36. f2.fork();  
    37. 
  
  
  
  
  37. returnf1.join()+f2.join();  
    38. 
  
  
  
  
  38. }  
    39. 
  
  
  
  
  39. }  
    40. 
  
  
  
  
  40.  
    41. 
  
  
  
  
  41. classConcurrentPrintextendsRecursiveAction{  
    42. 
  
  
  
  
  42. protectedvoidcompute(){  
    43. 
  
  
  
  
  43. TaskBarrierb=newTaskBarrier(){  
    44. 
  
  
  
  
  44. protectedbooleanterminate(intcycle,intregisteredParties){  
    45. 
  
  
  
  
  45. System.out.println("Cycleis"+cycle+";" 
    46. 
  
  
  
  
  46. +registeredParties+"parties");  
    47. 
  
  
  
  
  47. returncycle>=10;  
    48. 
  
  
  
  
  48. }  
    49. 
  
  
  
  
  49. };  
    50. 
  
  
  
  
  50. intn=3;  
    51. 
  
  
  
  
  51. CyclicAction[]actions=newCyclicAction[n];  
    52. 
  
  
  
  
  52. for(inti=0;i
  
  
  
  
  53. finalintindex=i;  
    54. 
  
  
  
  
  54. actions[i]=newCyclicAction(b){  
    55. 
  
  
  
  
  55. protectedvoidcompute(){  
    56. 
  
  
  
  
  56. System.out.println("I'mworking"+getCycle()+"" 
    57. 
  
  
  
  
  57. +index);  
    58. 
  
  
  
  
  58. try{  
    59. 
  
  
  
  
  59. Thread.sleep(500);  
    60. 
  
  
  
  
  60. }catch(InterruptedExceptione){  
    61. 
  
  
  
  
  61. e.printStackTrace();  
    62. 
  
  
  
  
  62. }  
    63. 
  
  
  
  
  63. }  
    64. 
  
  
  
  
  64. };  
    65. 
  
  
  
  
  65. }  
    66. 
  
  
  
  
  66. for(inti=0;i
  
  
  
  
  67. actions[i].fork();  
    68. 
  
  
  
  
  68. for(inti=0;i
  
  
  
  
  69. actions[i].join();  
    70. 
  
  
  
  
  70. }  
    71. 
  
  
  
  
  71. }  
    72. 
  
  
  
  
  72.  
    73. 
  
  
  
  
  73. publicclassTestForkJoin{  
    74. 
  
  
  
  
  74. @Test 
    75. 
  
  
  
  
  75. publicvoidtestBarrier()throwsInterruptedException,ExecutionException{  
    76. 
  
  
  
  
  76. System.out.println(" testingTaskBarrier...");  
    77. 
  
  
  
  
  77. ForkJoinTaskfjt=newConcurrentPrint();  
    78. 
  
  
  
  
  78. ForkJoinPoolfjpool=newForkJoinPool(4);  
    79. 
  
  
  
  
  79. fjpool.submit(fjt);  
    80. 
  
  
  
  
  80. fjpool.shutdown();  
    81. 
  
  
  
  
  81. }  
    82. 
  
  
  
  
  82.  
    83. 
  
  
  
  
  83. @Test 
    84. 
  
  
  
  
  84. publicvoidtestFibonacci()throwsInterruptedException,ExecutionException{  
    85. 
  
  
  
  
  85. System.out.println(" testingFibonacci...");  
    86. 
  
  
  
  
  86. finalintnum=14;//Fordemoonly  
    87. 
  
  
  
  
  87. ForkJoinTaskfjt=newFibonacci(num);  
    88. 
  
  
  
  
  88. ForkJoinPoolfjpool=newForkJoinPool();  
    89. 
  
  
  
  
  89. Futureresult=fjpool.submit(fjt);  
    90. 
  
  
  
  
  90.  
    91. 
  
  
  
  
  91. //dosomething  
    92. 
  
  
  
  
  92. System.out.println("Fibonacci("+num+")="+result.get());  
    93. 
  
  
  
  
  93. }  
    94. 
  
  
  
  
  94. }  
    95.

運行以上代碼，我們可以得到以下結(jié)果：

 
 
 
 
    
  
  
  
  
  1. testingTaskBarrier...  
    2. 
  
  
  
  
  2. I'mworking02  
    3. 
  
  
  
  
  3. I'mworking00  
    4. 
  
  
  
  
  4. I'mworking01  
    5. 
  
  
  
  
  5. Cycleis0;3parties  
    6. 
  
  
  
  
  6. I'mworking12  
    7. 
  
  
  
  
  7. I'mworking10  
    8. 
  
  
  
  
  8. I'mworking11  
    9. 
  
  
  
  
  9. Cycleis1;3parties  
    10. 
  
  
  
  
  10. I'mworking20  
    11. 
  
  
  
  
  11. I'mworking21  
    12. 
  
  
  
  
  12. I'mworking22  
    13. 
  
  
  
  
  13. Cycleis2;3parties  
    14. 
  
  
  
  
  14. I'mworking30  
    15. 
  
  
  
  
  15. I'mworking32  
    16. 
  
  
  
  
  16. I'mworking31  
    17. 
  
  
  
  
  17. Cycleis3;3parties  
    18. 
  
  
  
  
  18. I'mworking42  
    19. 
  
  
  
  
  19. I'mworking40  
    20. 
  
  
  
  
  20. I'mworking41  
    21. 
  
  
  
  
  21. Cycleis4;3parties  
    22. 
  
  
  
  
  22. I'mworking51  
    23. 
  
  
  
  
  23. I'mworking50  
    24. 
  
  
  
  
  24. I'mworking52  
    25. 
  
  
  
  
  25. Cycleis5;3parties  
    26. 
  
  
  
  
  26. I'mworking60  
    27. 
  
  
  
  
  27. I'mworking62  
    28. 
  
  
  
  
  28. I'mworking61  
    29. 
  
  
  
  
  29. Cycleis6;3parties  
    30. 
  
  
  
  
  30. I'mworking72  
    31. 
  
  
  
  
  31. I'mworking70  
    32. 
  
  
  
  
  32. I'mworking71  
    33. 
  
  
  
  
  33. Cycleis7;3parties  
    34. 
  
  
  
  
  34. I'mworking81  
    35. 
  
  
  
  
  35. I'mworking80  
    36. 
  
  
  
  
  36. I'mworking82  
    37. 
  
  
  
  
  37. Cycleis8;3parties  
    38. 
  
  
  
  
  38. I'mworking90  
    39. 
  
  
  
  
  39. I'mworking92  
    40. 
  
  
  
  
  40.  
    41. 
  
  
  
  
  41. testingFibonacci...  
    42. 
  
  
  
  
  42. forknewthreadfor13  
    43. 
  
  
  
  
  43. forknewthreadfor12  
    44. 
  
  
  
  
  44. forknewthreadfor11  
    45. 
  
  
  
  
  45. forknewthreadfor10  
    46. 
  
  
  
  
  46. forknewthreadfor12  
    47. 
  
  
  
  
  47. forknewthreadfor11  
    48. 
  
  
  
  
  48. forknewthreadfor10  
    49. 
  
  
  
  
  49. forknewthreadfor9  
    50. 
  
  
  
  
  50. forknewthreadfor10  
    51. 
  
  
  
  
  51. forknewthreadfor9  
    52. 
  
  
  
  
  52. forknewthreadfor11  
    53. 
  
  
  
  
  53. forknewthreadfor10  
    54. 
  
  
  
  
  54. forknewthreadfor10  
    55. 
  
  
  
  
  55. forknewthreadfor9  
    56. 
  
  
  
  
  56. Fibonacci(14)=610  
    57.

結(jié)  論

從以上的例子中可以看到，通過使用 Fork/Join 模式，軟件開發(fā)人員能夠方便地利用多核平臺的計算能力。盡管還沒有做到對軟件開發(fā)人員完全透明，F(xiàn)ork/Join 模式已經(jīng)極大地簡化了編寫并發(fā)程序的瑣碎工作。對于符合 Fork/Join 模式的應(yīng)用，軟件開發(fā)人員不再需要處理各種并行相關(guān)事務(wù)，例如同步、通信等，以難以調(diào)試而聞名的死鎖和 data race 等錯誤也就不會出現(xiàn)，提升了思考問題的層次。你可以把 Fork/Join 模式看作并行版本的 Divide and Conquer 策略，僅僅關(guān)注如何劃分任務(wù)和組合中間結(jié)果，將剩下的事情丟給 Fork/Join 框架。

在實際工作中利用 Fork/Join 模式，可以充分享受多核平臺為應(yīng)用帶來的免費午餐。

參考資料

◆  閱讀文章“The Free Lunch Is Over: A Fundamental Turn Toward Concurrency in Software”：了解為什么從現(xiàn)在開始每個嚴肅的軟件工作者都應(yīng)該了解并行編程方法。

◆ 閱讀 Doug Lea 的文章“A Java Fork/Join Framework”：了解 Fork/Join 模式的實現(xiàn)機制和執(zhí)行性能。

◆ 閱讀 developerWorks 文章“馴服 Tiger：并發(fā)集合”：了解如何使用并行 Collection 庫。

◆ 閱讀 developerWorks 文章“Java 理論與實踐：非阻塞算法簡介”：介紹了 JDK 5 在并行方面的重要增強以及在 JDK5 平臺上如何實現(xiàn)非阻塞算法的一般介紹。

◆ 書籍“Java Concurrency in Practice”：介紹了大量的并行編程技巧、反模式、可行的解決方案等，它對于 JDK 5 中的新特性也有詳盡的介紹。

獲得產(chǎn)品和技術(shù)

◆ 訪問 Doug Lea 的 JSR 166 站點獲得最新的源代碼。

◆ 從 Sun 公司 網(wǎng)站下載 Java SE 6。


分享標題：JDK7中的Fork/Join模式
本文鏈接：http://www.5511xx.com/article/cojjpih.html