java fork/join簡單實踐

咱們知道,java8中有並行流,而並行流在後臺的實現是經過fork/join池來完成的,例如:

List<Integer> a = buildList();

List<Integer> b = buildList();

a.parallelStream().foreach(System.out::println);

b.parallelStream().foreach(System.out::println);

若是查看線程dump會發現,這兩個並行流公用的是同一套線程池,.這在I/O密集型任務中會是個問題,應爲兩個並行流會因阻塞而有沒必要要的等待.在查閱相關文章以後,發現能夠經過java中的fork/join框架來改良.這裏簡單介紹一下:

要用fork/join框架須要繼承RecursiveTask類或者RecursiveAction,這兩個的區別是一個有返回值,一個沒有返回值,按字面上的意思就是遞歸任務.

這類task有fork()方法,就是拆分任務,還有一個join方法,合併任務.

這些task放到ForkJoinPool中執行.ForkJoinPool能夠指定線程池大小.

想到fork/join思想其實就是分治的思想,又和歸併排序的思想一致,就寫了一個用fork/join框架實現的歸併排序,供你們參考使用:

package com.wl.test2;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.RecursiveTask;

/**
 * Created by wally on 9/14/17.
 */
public class MergeSortByRecursiveTask extends RecursiveTask<List<Integer>> {

    private int start;

    private int end;

    private List<Integer> numbers;


    public MergeSortByRecursiveTask(int start, int end, List<Integer> numbers) {
        this.start = start;
        this.end = end;
        this.numbers = numbers;
    }

    @Override
    protected List<Integer> compute() {
        if (start == end) {
            return numbers;
        }
        int mid = (start + end) / 2;
        int aa = mid+1;
        MergeSortByRecursiveTask mergeSortByRecursiveTask = new MergeSortByRecursiveTask(start, mid, numbers);
        MergeSortByRecursiveTask mergeSortByRecursiveTask1 = new MergeSortByRecursiveTask(aa, end, numbers);

        mergeSortByRecursiveTask.fork();
        mergeSortByRecursiveTask1.fork();

        mergeSortByRecursiveTask.join();
        mergeSortByRecursiveTask1.join();

        merge();
        return numbers;
    }


    private void merge() {

        List<Integer> newNumbers = new ArrayList<>(numbers);
        int mid = (end + start) / 2;

        int tempMid = mid + 1;
        int i = start;
        int k = start;
        int m = start;
        while (m <= mid && tempMid <= end) {
            if (numbers.get(m) <= numbers.get(tempMid)) {
                newNumbers.set(k, numbers.get(m));
                m++;
            }else{
                if(tempMid==2){
                    System.out.println("here");
                }
                newNumbers.set(k,numbers.get(tempMid));
                tempMid++;
            }
            k++;
        }

        while(m<=mid){
            newNumbers.set(k,numbers.get(m));
            m++;
            k++;
        }
        while(tempMid<=end){
            newNumbers.set(k,numbers.get(tempMid));
            tempMid++;
            k++;
        }
        System.out.println("start is :"+ start+", mid is " + mid+ ", end is " + end);
        while(i<=end){

            numbers.set(i,newNumbers.get(i));
            System.out.print(newNumbers.get(i)+",");
            i++;
        }
        System.out.println(";");
        System.out.println(Thread.currentThread().getName());

    }
}

main函數:

package com.wl.test2;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.concurrent.*;

/**
 * Created by wally on 3/27/17.
 */
public class TestDIr {

    private static boolean a = true;

    public static void main(String[] args) throws NoSuchFieldException, IllegalAccessException, InterruptedException {
        List<Integer> temp = new ArrayList<>();
        temp.add(3);
        temp.add(9);
        temp.add(4);
        temp.add(3);
        temp.add(1);
        temp.add(8);
        temp.add(5);
        List<Integer> result = new ArrayList<>(temp);
        MergeSortByRecursiveTask mergeSortByRecursiveTask = new 
       MergeSortByRecursiveTask(0,temp.size()-1,temp);
        ForkJoinPool pool = new ForkJoinPool();
        Future<List<Integer>> sortResult = pool.submit(mergeSortByRecursiveTask);

        try{
            System.out.println(Arrays.toString(sortResult.get().toArray()));
         
        } catch (ExecutionException e) {
            e.printStackTrace();
        }

    }
}

輸出結果:

start is :4, mid is 4, end is 5
1,8,;
ForkJoinPool-1-worker-3
start is :4, mid is 5, end is 6
1,5,8,;
ForkJoinPool-1-worker-3
start is :0, mid is 0, end is 1
3,9,;
ForkJoinPool-1-worker-2
start is :2, mid is 2, end is 3
3,4,;
ForkJoinPool-1-worker-2
here
start is :0, mid is 1, end is 3
3,3,4,9,;
ForkJoinPool-1-worker-2
start is :0, mid is 3, end is 6
1,3,3,4,5,8,9,;
ForkJoinPool-1-worker-1
[1, 3, 3, 4, 5, 8, 9]

 

最後推薦幾篇文章:

http://www.infoq.com/cn/articles/fork-join-introduction

http://www.importnew.com/16801.html