RecursiveTask和RecursiveAction的使用 以及java 8 并行流和顺序流
轉載自?https://blog.csdn.net/weixin_41404773/article/details/80733324
什么是Fork/Join框架
????????Fork/Join框架是Java7提供了的一個用于并行執行任務的框架, 是一個把大任務分割成若干個小任務,最終匯總每個小任務結果后得到大任務結果的框架。
?????????我們再通過Fork和Join這兩個單詞來理解下Fork/Join框架,Fork就是把一個大任務切分為若干子任務并行的執行,Join就是合并這些子任務的執行結果,最后得到這個大任務的結果。比如計算1+2+。。+10000,可以分割成10個子任務,每個子任務分別對1000個數進行求和,最終匯總這10個子任務的結果。Fork/Join的運行流程圖如下:
工作竊取算法
?????????工作竊取(work-stealing)算法是指某個線程從其他隊列里竊取任務來執行。工作竊取的運行流程圖如下:
????????
????????那么為什么需要使用工作竊取算法呢?假如我們需要做一個比較大的任務,我們可以把這個任務分割為若干互不依賴的子任務,為了減少線程間的競爭,于是把這些子任務分別放到不同的隊列里,并為每個隊列創建一個單獨的線程來執行隊列里的任務,線程和隊列一一對應,比如A線程負責處理A隊列里的任務。但是有的線程會先把自己隊列里的任務干完,而其他線程對應的隊列里還有任務等待處理。干完活的線程與其等著,不如去幫其他線程干活,于是它就去其他線程的隊列里竊取一個任務來執行。而在這時它們會訪問同一個隊列,所以為了減少竊取任務線程和被竊取任務線程之間的競爭,通常會使用雙端隊列,被竊取任務線程永遠從雙端隊列的頭部拿任務執行,而竊取任務的線程永遠從雙端隊列的尾部拿任務執行。
?????
????????工作竊取算法的優點是充分利用線程進行并行計算,并減少了線程間的競爭,其缺點是在某些情況下還是存在競爭,比如雙端隊列里只有一個任務時。并且消耗了更多的系統資源,比如創建多個線程和多個雙端隊列。
ForkJoinPool
????????Java提供了ForkJoinPool來支持將一個任務拆分成多個“小任務”并行計算,再把多個“小任務”的結果合成總的計算結果。
????????ForkJoinPool是ExecutorService的實現類,因此是一種特殊的線程池。ForkJoinPool提供了如下兩個常用的構造器。
?public ForkJoinPool(int parallelism):創建一個包含parallelism個并行線程的ForkJoinPool
?public ForkJoinPool() :以Runtime.getRuntime().availableProcessors()的返回值作為parallelism來創建ForkJoinPool
?????????創建ForkJoinPool實例后,可以釣魚ForkJoinPool的submit(ForkJoinTask<T> task)或者invoke(ForkJoinTask<T> task)來執行指定任務。其中ForkJoinTask代表一個可以并行、合并的任務。ForkJoinTask是一個抽象類,它有兩個抽象子類:RecursiveAction和RecursiveTask。
RecursiveTask代表有返回值的任務
RecursiveAction代表沒有返回值的任務。
RecursiveAction
下面以一個沒有返回值的大任務為例,介紹一下RecursiveAction的用法。
大任務是:打印0-100的數值。
小任務是:每次只能打印20個數值。
代碼執行
package com.example.jedis.test;
import java.util.concurrent.RecursiveAction;
/**
?*
?* @Author : Wukn
?* @Date : 2018/2/5
?*/
public class RaskDemo extends RecursiveAction {
? ? /**
? ? ?* ?每個"小任務"最多只打印20個數
? ? ? */
? ? private static final int MAX = 20;
? ? private int start;
? ? private int end;
? ? public RaskDemo(int start, int end) {
? ? ? ? this.start = start;
? ? ? ? this.end = end;
? ? }
? ? @Override
? ? protected void compute() {
? ? ? ? //當end-start的值小于MAX時,開始打印
? ? ? ? if((end-start) < MAX) {
? ? ? ? ? ? for(int i= start; i<end;i++) {
? ? ? ? ? ? ? ? System.out.println(Thread.currentThread().getName()+"i的值"+i);
? ? ? ? ? ? }
? ? ? ? }else {
? ? ? ? ? ? // 將大任務分解成兩個小任務
? ? ? ? ? ? int middle = (start + end) / 2;
? ? ? ? ? ? RaskDemo left = new RaskDemo(start, middle);
? ? ? ? ? ? RaskDemo right = new RaskDemo(middle, end);
? ? ? ? ? ? left.fork();
? ? ? ? ? ? right.fork();
? ? ? ? }
? ? }
}
public static void main(String[] args) throws Exception{
? ? // 創建包含Runtime.getRuntime().availableProcessors()返回值作為個數的并行線程的ForkJoinPool
? ? ForkJoinPool forkJoinPool = new ForkJoinPool();
? ? // 提交可分解的PrintTask任務
? ? forkJoinPool.submit(new RaskDemo(0, 1000));
? ? //阻塞當前線程直到 ForkJoinPool 中所有的任務都執行結束
? ? forkJoinPool.awaitTermination(2, TimeUnit.SECONDS);
? ? // 關閉線程池
? ? forkJoinPool.shutdown();
}
運行結果
從上面結果來看,ForkJoinPool啟動了四個線程來執行這個打印任務,我的計算機的CPU是四核的。大家還可以看到程序雖然打印了0-999這一千個數字,但是并不是連續打印的,這是因為程序將這個打印任務進行了分解,分解后的任務會并行執行,所以不會按順序打印。
RecursiveTask
下面以一個有返回值的大任務為例,介紹一下RecursiveTask的用法。
大任務是:計算隨機的1000個數字的和。
小任務是:每次只能70個數值的和。
package com.example.jedis.test;
import java.util.concurrent.RecursiveTask;
/**
?*
?* @Author : Wukn
?* @Date : 2018/2/5
?*/
public class RecursiveTaskDemo extends RecursiveTask<Integer> {
? ? /**
? ? ?* ?每個"小任務"最多只打印70個數
? ? ?*/
? ? private static final int MAX = 70;
? ? private int arr[];
? ? private int start;
? ? private int end;
? ? public RecursiveTaskDemo(int[] arr, int start, int end) {
? ? ? ? this.arr = arr;
? ? ? ? this.start = start;
? ? ? ? this.end = end;
? ? }
? ? @Override
? ? protected Integer compute() {
? ? ? ? int sum = 0;
? ? ? ? // 當end-start的值小于MAX時候,開始打印
? ? ? ? if((end - start) < MAX) {
? ? ? ? ? ? for (int i = start; i < end; i++) {
? ? ? ? ? ? ? ? sum += arr[i];
? ? ? ? ? ? }
? ? ? ? ? ? return sum;
? ? ? ? }else {
? ? ? ? ? ? System.err.println("=====任務分解======");
? ? ? ? ? ? // 將大任務分解成兩個小任務
? ? ? ? ? ? int middle = (start + end) / 2;
? ? ? ? ? ? RecursiveTaskDemo left = new RecursiveTaskDemo(arr, start, middle);
? ? ? ? ? ? RecursiveTaskDemo right = new RecursiveTaskDemo(arr, middle, end);
? ? ? ? ? ? // 并行執行兩個小任務
? ? ? ? ? ? left.fork();
? ? ? ? ? ? right.fork();
? ? ? ? ? ? // 把兩個小任務累加的結果合并起來
? ? ? ? ? ? return left.join() + right.join();
? ? ? ? }
? ? }
}
? ? @Test
? ? public void dfs() throws Exception{
? ? ? ? int arr[] = new int[1000];
? ? ? ? Random random = new Random();
? ? ? ? int total = 0;
? ? ? ? // 初始化100個數字元素
? ? ? ? for (int i = 0; i < arr.length; i++) {
? ? ? ? ? ? int temp = random.nextInt(100);
? ? ? ? ? ? // 對數組元素賦值,并將數組元素的值添加到total總和中
? ? ? ? ? ? total += (arr[i] = temp);
? ? ? ? }
? ? ? ? System.out.println("初始化時的總和=" + total);
? ? ? ? // 創建包含Runtime.getRuntime().availableProcessors()返回值作為個數的并行線程的ForkJoinPool
? ? ? ? ForkJoinPool forkJoinPool = new ForkJoinPool();
? ? ? ? // 提交可分解的PrintTask任務
// ? ? ? ?Future<Integer> future = forkJoinPool.submit(new RecursiveTaskDemo(arr, 0, arr.length));
// ? ? ? ?System.out.println("計算出來的總和="+future.get());
? ? ? ? Integer integer = forkJoinPool.invoke( new RecursiveTaskDemo(arr, 0, arr.length) ?);
? ? ? ? System.out.println("計算出來的總和=" + integer);
? ? ? ? // 關閉線程池
? ? ? ? forkJoinPool.shutdown();
? ? }
從上面結果來看,ForkJoinPool將任務分解了15次,程序通過SumTask計算出來的結果,和初始化數組時統計出來的總和是相等的,這表明計算結果一切正常。
總結
第一步分割任務 ?
????????首先我們需要有一個fork類來把大任務分割成子任務,有可能子任務還是很大,所以還需要不停的分割,直到分割出的子任務足夠小。
第二步執行任務并合并結果。
??????分割的子任務分別放在雙端隊列里,然后幾個啟動線程分別從雙端隊列里獲取任務執行。子任務執行完的結果都統一放在一個隊列里,啟動一個線程從隊列里拿數據,然后合并這些數據。
能夠輕松的利用多個 CPU 提供的計算資源來協作完成一個復雜的計算任務,提高運行效率!
java8新的寫法
/************************************** ?并行流 與 順序流 ?******************************************************/
? ? /**
? ? ?*并行流 與 順序流
? ? ?*/
? ? @Test
? ? public void test03() {
? ? ? ? Instant start = Instant.now();
? ? ? ? LongStream.rangeClosed( 0,110 )
? ? ? ? ? ? ? ? //并行流
? ? ? ? ? ? ? ? .parallel()
? ? ? ? ? ? ? ? .reduce( 0,Long::sum );
? ? ? ? LongStream.rangeClosed( 0,110 )
? ? ? ? ? ? ? ? //順序流
? ? ? ? ? ? ? ? .sequential()
? ? ? ? ? ? ? ? .reduce( 0,Long::sum );
? ? ? ? Instant end = Instant.now();
? ? ? ? System.out.println("耗費時間"+ Duration.between( start,end ).toMillis());
}
?
總結
以上是生活随笔為你收集整理的RecursiveTask和RecursiveAction的使用 以及java 8 并行流和顺序流的全部內容,希望文章能夠幫你解決所遇到的問題。
- 上一篇: RunTime.getRunTime()
- 下一篇: JDK6u25里添加的按线程统计分配内存