快捷導(dǎo)航

Java7之forkjoin簡(jiǎn)介_動(dòng)力節(jié)點(diǎn)Java學(xué)院整理

更新時(shí)間：2017年06月01日 09:48:30 投稿：mrr

Java7引入了Fork Join的概念，來更好的支持并行運(yùn)算。接下來通過本文給大家分享Java7之forkjoin簡(jiǎn)介，感興趣的朋友一起看看吧

Java7引入了Fork Join的概念，來更好的支持并行運(yùn)算。顧名思義，F(xiàn)ork Join類似與流程語言的分支，合并的概念。也就是說Java7 SE原生支持了在一個(gè)主線程中開辟多個(gè)分支線程，并且根據(jù)分支線程的邏輯來等待(或者不等待)匯集，當(dāng)然你也可以fork的某一個(gè)分支線程中再開辟Fork Join，這也就可以實(shí)現(xiàn)Fork Join的嵌套。

有兩個(gè)核心類ForkJoinPool和ForkJoinTask。

ForkJoinPool實(shí)現(xiàn)了ExecutorService接口，起到線程池的作用。所以他的用法和Executor框架的使用時(shí)一樣的，當(dāng)然Fork Join本身就是Executor框架的擴(kuò)展。ForkJoinPool有3個(gè)關(guān)鍵的方法，來啟動(dòng)線程，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 style="BACKGROUND: white; TEXT-ALIGN: left; LINE-HEIGHT: normal;" align=left><SPAN style='FONT-SIZE: 9pt;"微軟雅黑","sans-serif"; COLOR: #333333;

ForkJoinTask是分支合并的執(zhí)行任何，分支合并的業(yè)務(wù)邏輯使用者可以再繼承了這個(gè)抽先類之后，在抽象方法exec()中實(shí)現(xiàn)。其中exec()的返回結(jié)果和ForkJoinPool的執(zhí)行調(diào)用方（execute(…),invoke(…),submit(…)），共同決定著線程是否阻塞，具體請(qǐng)看下面的測(cè)試用例。

首先，用戶需要?jiǎng)?chuàng)建一個(gè)自己的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;
 }
}

測(cè)試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，此處阻塞，直到另一個(gè)線程調(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ù)運(yùn)行完畢后結(jié)束阻塞
  ForkJoinTask result = forkJoinPool.submit(task);
  result.get(); // 如果exec()返回值是false，在此處會(huì)阻塞，直到調(diào)用complete
 }

測(cè)試ForkJoinPool.submit(…)：

@Test
 public void testForkJoinSubmit() throws InterruptedException, ExecutionException {
  final ForkJoinPool forkJoinPool = new ForkJoinPool();
  final MyForkJoinTask task = new MyForkJoinTask();
  task.setSuccess(true); // 是否在此任務(wù)運(yùn)行完畢后結(jié)束阻塞
  ForkJoinTask result = forkJoinPool.submit(task);
  result.get(); // 如果exec()返回值是false，在此處會(huì)阻塞，直到調(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，此處阻塞，直到另一個(gè)線程調(diào)用了task.complete(...)
  result.get();
  Thread.sleep(1000);
 }

測(cè)試ForkJoinPool.execute(…)：

 @Test
 public void testForkJoinExecute() throws InterruptedException, ExecutionException {
  ForkJoinPool forkJoinPool = new ForkJoinPool();
  MyForkJoinTask task = new MyForkJoinTask();
  forkJoinPool.execute(task); // 異步執(zhí)行，無視task.exec()返回值。
 }

在實(shí)際情況中，很多時(shí)候我們都需要面對(duì)經(jīng)典的“分治”問題。要解決這類問題，主要任務(wù)通常被分解為多個(gè)任務(wù)塊（分解階段），其后每一小塊任務(wù)被獨(dú)立并行計(jì)算。一旦計(jì)算任務(wù)完成，每一快的結(jié)果會(huì)被合并或者解決（解決階段）。ForkJoinTask天然就是為了支持“分治”問題的。

分支/合并的完整過程如下：

下面列舉一個(gè)分治算法的實(shí)例。

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ǎn)介_動(dòng)力節(jié)點(diǎn)Java學(xué)院整理，希望對(duì)大家有所幫助，如果大家有任何疑問歡迎給我留言，小編會(huì)及時(shí)回復(fù)大家的，在此也非常感謝大家對(duì)腳本之家網(wǎng)站的支持！

您可能感興趣的文章: