深入淺析Java7中的新特性forkjoin？很多新手對此不是很清楚，為了幫助大家解決這個難題，下面小編將為大家詳細講解，有這方面需求的人可以來學習下，希望你能有所收獲。
Java7引入了Fork Join的概念，來更好的支持并行運算。顧名思義，Fork Join類似與流程語言的分支，合并的概念。也就是說Java7 SE原生支持了在一個主線程中開辟多個分支線程，并且根據分支線程的邏輯來等待(或者不等待)匯集，當然你也可以fork的某一個分支線程中再開辟Fork Join，這也就可以實現Fork Join的嵌套。
有兩個核心類ForkJoinPool和ForkJoinTask。
ForkJoinPool實現了ExecutorService接口，起到線程池的作用。所以他的用法和Executor框架的使用時一樣的，當然Fork Join本身就是Executor框架的擴展。ForkJoinPool有3個關鍵的方法，來啟動線程，execute(…),invoke(…),submit(…)。具體描述如下：
<SPAN style='FONT-SIZE: 9pt;"微軟雅黑","sans-serif"; COLOR: #333333;"BORDER-TOP: windowtext 1pt solid; BORDER-RIGHT: windowtext 1pt solid; BORDER-BOTTOM: windowtext 1pt solid; PADDING-BOTTOM: 0cm; PADDING-TOP: 0cm; PADDING-LEFT: 0cm; BORDER-LEFT: windowtext 1pt solid; PADDING-RIGHT: 0cm; BACKGROUND-COLOR: transparent;"> <P align=left><SPAN style='FONT-SIZE: 9pt;"微軟雅黑","sans-serif"; COLOR: #333333; |
首先，用戶需要創建一個自己的ForkJoinTask。代碼如下：
public class MyForkJoinTask extends ForkJoinTask {
/**
*
*/
private static final long serialVersionUID = 1L;
private V value;
private boolean success = false;
@Override
public V getRawResult() {
return value;
}
@Override
protected void setRawResult(V value) {
this.value = value;
}
@Override
protected boolean exec() {
System.out.println("exec");
return this.success;
}
public boolean isSuccess() {
return success;
}
public void setSuccess(boolean isSuccess) {
this.success = isSuccess;
}
}測試ForkJoinPool.invoke(…)：
@Test
public void testForkJoinInvoke() throws InterruptedException, ExecutionException {
ForkJoinPool forkJoinPool = new ForkJoinPool();
MyForkJoinTask task = new MyForkJoinTask();
task.setSuccess(true);
task.setRawResult("test");
String invokeResult = forkJoinPool.invoke(task);
assertEquals(invokeResult, "test");
}
@Test
public void testForkJoinInvoke2() throws InterruptedException, ExecutionException {
final ForkJoinPool forkJoinPool = new ForkJoinPool();
final MyForkJoinTask task = new MyForkJoinTask();
new Thread(new Runnable() {
public void run() {
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
}
task.complete("test");
}
}).start();
// exec()返回值是false，此處阻塞，直到另一個線程調用了task.complete(...)
String result = forkJoinPool.invoke(task);
System.out.println(result);
}
@Test
public void testForkJoinSubmit() throws InterruptedException, ExecutionException {
final ForkJoinPool forkJoinPool = new ForkJoinPool();
final MyForkJoinTask task = new MyForkJoinTask();
task.setSuccess(true); // 是否在此任務運行完畢后結束阻塞
ForkJoinTask result = forkJoinPool.submit(task);
result.get(); // 如果exec()返回值是false，在此處會阻塞，直到調用complete
}測試ForkJoinPool.submit(…)：
@Test
public void testForkJoinSubmit() throws InterruptedException, ExecutionException {
final ForkJoinPool forkJoinPool = new ForkJoinPool();
final MyForkJoinTask task = new MyForkJoinTask();
task.setSuccess(true); // 是否在此任務運行完畢后結束阻塞
ForkJoinTask result = forkJoinPool.submit(task);
result.get(); // 如果exec()返回值是false，在此處會阻塞，直到調用complete
}
@Test
public void testForkJoinSubmit2() throws InterruptedException, ExecutionException {
final ForkJoinPool forkJoinPool = new ForkJoinPool();
final MyForkJoinTask task = new MyForkJoinTask();
forkJoinPool.submit(task);
Thread.sleep(1000);
}
@Test
public void testForkJoinSubmit3() throws InterruptedException, ExecutionException {
final ForkJoinPool forkJoinPool = new ForkJoinPool();
final MyForkJoinTask task = new MyForkJoinTask();
new Thread(new Runnable() {
public void run() {
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
}
task.complete("test");
}
}).start();
ForkJoinTask result = forkJoinPool.submit(task);
// exec()返回值是false，此處阻塞，直到另一個線程調用了task.complete(...)
result.get();
Thread.sleep(1000);
}測試ForkJoinPool.execute(…)：
@Test
public void testForkJoinExecute() throws InterruptedException, ExecutionException {
ForkJoinPool forkJoinPool = new ForkJoinPool();
MyForkJoinTask task = new MyForkJoinTask();
forkJoinPool.execute(task); // 異步執行，無視task.exec()返回值。
}在實際情況中，很多時候我們都需要面對經典的“分治”問題。要解決這類問題，主要任務通常被分解為多個任務塊（分解階段），其后每一小塊任務被獨立并行計算。一旦計算任務完成，每一快的結果會被合并或者解決（解決階段）。ForkJoinTask天然就是為了支持“分治”問題的。
分支/合并的完整過程如下：
下面列舉一個分治算法的實例。
import java.util.Random;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.RecursiveTask;
public class MaximumFinder extends RecursiveTask<Integer> {
private static final int SEQUENTIAL_THRESHOLD = 5;
private final int[] data;
private final int start;
private final int end;
public MaximumFinder(int[] data, int start, int end) {
this.data = data;
this.start = start;
this.end = end;
}
public MaximumFinder(int[] data) {
this(data, 0, data.length);
}
@Override
protected Integer compute() {
final int length = end - start;
if (length < SEQUENTIAL_THRESHOLD) {
return computeDirectly();
}
final int split = length / 2;
final MaximumFinder left = new MaximumFinder(data, start, start + split);
left.fork();
final MaximumFinder right = new MaximumFinder(data, start + split, end);
return Math.max(right.compute(), left.join());
}
private Integer computeDirectly() {
System.out.println(Thread.currentThread() + ' computing: ' + start
+ ' to ' + end);
int max = Integer.MIN_VALUE;
for (int i = start; i < end; i++) {
if (data[i] > max) {
max = data[i];
}
}
return max;
}
public static void main(String[] args) {
// create a random data set
final int[] data = new int[1000];
final Random random = new Random();
for (int i = 0; i < data.length; i++) {
data[i] = random.nextInt(100);
}
// submit the task to the pool
final ForkJoinPool pool = new ForkJoinPool(4);
final MaximumFinder finder = new MaximumFinder(data);
System.out.println(pool.invoke(finder));
}
}看完上述內容是否對您有幫助呢？如果還想對相關知識有進一步的了解或閱讀更多相關文章，請關注億速云行業資訊頻道，感謝您對億速云的支持。
免責聲明：本站發布的內容（圖片、視頻和文字）以原創、轉載和分享為主，文章觀點不代表本網站立場，如果涉及侵權請聯系站長郵箱：is@yisu.com進行舉報，并提供相關證據，一經查實，將立刻刪除涉嫌侵權內容。
