fork/join

2019-11-06 06:30:49

字體：大中小

供稿：網(wǎng)友

并行計(jì)算在處處都有大數(shù)據(jù)的今天已經(jīng)不是一個(gè)新鮮的詞匯了，現(xiàn)在已經(jīng)有單機(jī)多核甚至多機(jī)集群并行計(jì)算，注意，這里說(shuō)的是并行，而不是并發(fā)。嚴(yán)格的將，并行是指系統(tǒng)內(nèi)有多個(gè)任務(wù)同時(shí)執(zhí)行，而并發(fā)是指系統(tǒng)內(nèi)有多個(gè)任務(wù)同時(shí)存在，不同的任務(wù)按時(shí)間分片的方式切換執(zhí)行，由于切換的時(shí)間很短，給人的感覺(jué)好像是在同時(shí)執(zhí)行。 java在JDK7之后加入了并行計(jì)算的框架Fork/Join，可以解決我們系統(tǒng)中大數(shù)據(jù)計(jì)算的性能問(wèn)題。Fork/Join采用的是分治法，F(xiàn)ork是將一個(gè)大任務(wù)拆分成若干個(gè)子任務(wù)，子任務(wù)分別去計(jì)算，而Join是獲取到子任務(wù)的計(jì)算結(jié)果，然后合并，這個(gè)是遞歸的過(guò)程。子任務(wù)被分配到不同的核上執(zhí)行時(shí)，效率最高。偽代碼如下：

Result solve(PRoblem problem) { if (problem is small) directly solve problem else { split problem into independent parts fork new subtasks to solve each part join all subtasks compose result from subresults }}12345678910

12345678910

Fork/Join框架的核心類是ForkJoinPool，它能夠接收一個(gè)ForkJoinTask，并得到計(jì)算結(jié)果。ForkJoinTask有兩個(gè)子類，RecursiveTask（有返回值）和RecursiveAction（無(wú)返回結(jié)果），我們自己定義任務(wù)時(shí)，只需選擇這兩個(gè)類繼承即可。類圖如下：這里寫(xiě)圖片描述下面來(lái)看一個(gè)實(shí)例：計(jì)算一個(gè)超大數(shù)組所有元素的和。代碼如下：

import java.util.Arrays;import java.util.Random;import java.util.concurrent.ExecutionException;import java.util.concurrent.ForkJoinPool;import java.util.concurrent.RecursiveTask;/** * @author: shuang.gao Date: 2015/7/14 Time: 8:16 */public class SumTask extends RecursiveTask<Integer> { private static final long serialVersionUID = -6196480027075657316L; private static final int THRESHOLD = 500000; private long[] array; private int low; private int high; public SumTask(long[] array, int low, int high) { this.array = array; this.low = low; this.high = high; } @Override protected Integer compute() { int sum = 0; if (high - low <= THRESHOLD) { // 小于閾值則直接計(jì)算 for (int i = low; i < high; i++) { sum += array[i]; } } else { // 1. 一個(gè)大任務(wù)分割成兩個(gè)子任務(wù) int mid = (low + high) >>> 1; SumTask left = new SumTask(array, low, mid); SumTask right = new SumTask(array, mid + 1, high); // 2. 分別計(jì)算 left.fork(); right.fork(); // 3. 合并結(jié)果 sum = left.join() + right.join(); } return sum; } public static void main(String[] args) throws ExecutionException, InterruptedException { long[] array = genArray(1000000); System.out.println(Arrays.toString(array)); // 1. 創(chuàng)建任務(wù) SumTask sumTask = new SumTask(array, 0, array.length - 1); long begin = System.currentTimeMillis(); // 2. 創(chuàng)建線程池 ForkJoinPool forkJoinPool = new ForkJoinPool(); // 3. 提交任務(wù)到線程池 forkJoinPool.submit(sumTask); // 4. 獲取結(jié)果 Integer result = sumTask.get(); long end = System.currentTimeMillis(); System.out.println(String.format("結(jié)果 %s 耗時(shí) %sms", result, end - begin)); } private static long[] genArray(int size) { long[] array = new long[size]; for (int i = 0; i < size; i++) { array[i] = new Random().nextLong(); } return array; }}123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384

我們通過(guò)調(diào)整閾值（THRESHOLD），可以發(fā)現(xiàn)耗時(shí)是不一樣的。實(shí)際應(yīng)用中，如果需要分割的任務(wù)大小是固定的，可以經(jīng)過(guò)測(cè)試，得到最佳閾值；如果大小不是固定的，就需要設(shè)計(jì)一個(gè)可伸縮的算法，來(lái)動(dòng)態(tài)計(jì)算出閾值。如果子任務(wù)很多，效率并不一定會(huì)高。未完待續(xù)。。。

參考資料

http://gee.cs.oswego.edu/dl/papers/fj.pdf https://docs.Oracle.com/javase/tutorial/essential/concurrency/forkjoin.html https://www.ibm.com/developerworks/cn/java/j-lo-forkjoin/ http://www.ibm.com/developerworks/cn/java/j-jtp11137.html

本文來(lái)自：高爽|Coder，原文地址：http://blog.csdn.net/ghsau/article/details/46287769，轉(zhuǎn)載請(qǐng)注明。