Java7引入了Fork Join的概念,來更好的支持并行運算。顧名思義,F(xiàn)ork Join類似與流程語言的分支,合并的概念。也就是說Java7 SE原生支持了在一個主線程中開辟多個分支線程,并且根據(jù)分支線程的邏輯來等待(或者不等待)匯集,當(dāng)然你也可以fork的某一個分支線程中再開辟Fork Join,這也就可以實現(xiàn)Fork Join的嵌套。
有兩個核心類ForkJoinPool和ForkJoinTask。
ForkJoinPool實現(xiàn)了ExecutorService接口,起到線程池的作用。所以他的用法和Executor框架的使用時一樣的,當(dāng)然Fork Join本身就是Executor框架的擴展。ForkJoinPool有3個關(guān)鍵的方法,來啟動線程,execute(…),invoke(…),submit(…)。具體描述如下:
首先,用戶需要創(chuàng)建一個自己的ForkJoinTask。代碼如下:
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
|
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(…):
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
|
@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,此處阻塞,直到另一個線程調(diào)用了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 ); // 是否在此任務(wù)運行完畢后結(jié)束阻塞 ForkJoinTask result = forkJoinPool.submit(task); result.get(); // 如果exec()返回值是false,在此處會阻塞,直到調(diào)用complete } |
測試ForkJoinPool.submit(…):
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
|
@Test public void testForkJoinSubmit() throws InterruptedException, ExecutionException { final ForkJoinPool forkJoinPool = new ForkJoinPool(); final MyForkJoinTask task = new MyForkJoinTask(); task.setSuccess( true ); // 是否在此任務(wù)運行完畢后結(jié)束阻塞 ForkJoinTask result = forkJoinPool.submit(task); result.get(); // 如果exec()返回值是false,在此處會阻塞,直到調(diào)用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,此處阻塞,直到另一個線程調(diào)用了task.complete(...) result.get(); Thread.sleep( 1000 ); } |
測試ForkJoinPool.execute(…):
1
2
3
4
5
6
|
@Test public void testForkJoinExecute() throws InterruptedException, ExecutionException { ForkJoinPool forkJoinPool = new ForkJoinPool(); MyForkJoinTask task = new MyForkJoinTask(); forkJoinPool.execute(task); // 異步執(zhí)行,無視task.exec()返回值。 } |
在實際情況中,很多時候我們都需要面對經(jīng)典的“分治”問題。要解決這類問題,主要任務(wù)通常被分解為多個任務(wù)塊(分解階段),其后每一小塊任務(wù)被獨立并行計算。一旦計算任務(wù)完成,每一快的結(jié)果會被合并或者解決(解決階段)。ForkJoinTask天然就是為了支持“分治”問題的。
分支/合并的完整過程如下:
下面列舉一個分治算法的實例。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
|
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)); } } |
以上所示是小編給大家介紹的Java7之forkjoin簡介_動力節(jié)點Java學(xué)院整理,希望對大家有所幫助,如果大家有任何疑問歡迎給我留言,小編會及時回復(fù)大家的,在此也非常感謝大家對服務(wù)器之家網(wǎng)站的支持!