Fork/Join框架怎么用

這篇文章給大家分享的是有關(guān)Fork/Join框架怎么用的內(nèi)容。小編覺得挺實用的，因此分享給大家做個參考，一起跟隨小編過來看看吧。

fork/join框架是ExecutorService接口的一種具體實現(xiàn)，會將任務(wù)分發(fā)給線程池中的工作線程，更好地利用多處理器帶來的好處，提供程序性能。它是為那些能夠被遞歸地拆解成子任務(wù)的工作類型量身設(shè)計的。

fork/join框架的獨特之處在與它使用工作竊取(work-stealing)算法。完成自己的工作而處于空閑的工作線程能夠從其他仍然處于忙碌(busy)狀態(tài)的工作線程處竊取等待執(zhí)行的任務(wù)。

fork/join框架的核心是ForkJoinPool類，它是對AbstractExecutorService類的擴展。ForkJoinPool實現(xiàn)了工作偷取算法，并可以執(zhí)行ForkJoinTask任務(wù)。

使用fork/join框架的第一步是編寫執(zhí)行一部分工作的代碼，類似的偽代碼如下：

if (當前這個任務(wù)工作量足夠小)

  直接完成這個任務(wù)

else

  把當前任務(wù)分解成兩個部分
  調(diào)用這兩個部分并等待結(jié)果

此代包裝在ForkJoinTask的子類中。不過，通常是RecursiveTask(會返回一個結(jié)果)，或RecursiveAction。

fork分解出新任務(wù)，join匯集任務(wù)結(jié)果，其大致過程如下：

Fork/Join實際用例

fork/join的核心思想就是分而治之，將一個大任務(wù)拆分成一個一個的小任務(wù)。fork/join的使用需要定義一個任務(wù)類去實現(xiàn)RecursiveTask或RecursiveAction，重寫compute()方法，在compute()方法中定義任務(wù)拆分的邏輯，然后借助ForkJoinPool提交任務(wù)去執(zhí)行，fork/join框架會根據(jù)compute()方法中定義的拆分邏輯對任務(wù)進行具體的拆分，如果有返回值，可以借助ForkJoinTask獲取返回值。假設(shè)現(xiàn)在有很多網(wǎng)絡(luò)請求需要并發(fā)的去執(zhí)行，然后匯總結(jié)果，使用fork/join的代碼實現(xiàn)如下：

public class ForkJoinTest {

  // 測試數(shù)據(jù)  
  static ArrayList<String> urls =
      new ArrayList<String>() {
        {
          add("http://www.sina.com");
          add("http://www.baidu.com");
          add("http://www.sina.com");
          add("http://www.baidu.com");
          add("http://www.sina.com");
          add("http://www.baidu.com");
          add("http://www.sina.com");
          add("http://www.baidu.com");
          add("http://www.sina.com");
          add("http://www.baidu.com");
          add("http://www.sina.com");
          add("http://www.baidu.com");
          add("http://www.sina.com");
          add("http://www.baidu.com");
          add("http://www.sina.com");
          add("http://www.baidu.com");
          add("http://www.sina.com");
          add("http://www.baidu.com");
          add("http://www.sina.com");
          add("http://www.baidu.com");
          add("http://www.sina.com");
          add("http://www.baidu.com");
          add("http://www.sina.com");
        }
      };

  // 核心ForkJoinPool
  static ForkJoinPool firkJoinPool =
      new ForkJoinPool(3, ForkJoinPool.defaultForkJoinWorkerThreadFactory, null, true);
  public static void main(String args[]) throws ExecutionException, InterruptedException {
    Job job = new Job(urls, 0, urls.size());
    ForkJoinTask<String> forkJoinTask = firkJoinPool.submit(job);
    String result = forkJoinTask.get();
    System.out.println(result);
  }
  public static String doRequest(String url) {
    // 模擬網(wǎng)絡(luò)請求
    return "Kody ... test ... " + url + "\n";
  }

  // 自定義任務(wù)  
  static class Job extends RecursiveTask<String> {

    List<String> urls;
    int start;

    int end;

    public Job(List<String> urls, int start, int end) {
      this.urls = urls;
      this.start = start;
      this.end = end;

    }

    /** 核心任務(wù)的計算，任務(wù)拆分邏輯實現(xiàn) */
    @Override
    protected String compute() {

      // 計算任務(wù)的大小      
      int count = end - start;

      // 如果任務(wù)拆分的足夠小，則執(zhí)行任務(wù)      
      if (count <= 5) {
        // 直接執(zhí)行        
        String result = "";
        for (int i = start; i < end; i++) {
          String response = doRequest(urls.get(i));
          result += response;
        }
        return result;

      } else { // 如果任務(wù)較大，繼續(xù)拆分任務(wù)

        int x = (start + end) / 2;
        Job job1 = new Job(urls, start, x);
        job1.fork();
        Job job2 = new Job(urls, x, end);
        job2.fork();

        // 匯總fork出的子任務(wù)結(jié)果        
        return String.join("", job1.join(), job2.join());

      }
    }
  }
}

感謝各位的閱讀！關(guān)于“Fork/Join框架怎么用”這篇文章就分享到這里了，希望以上內(nèi)容可以對大家有一定的幫助，讓大家可以學到更多知識，如果覺得文章不錯，可以把它分享出去讓更多的人看到吧！