并发排序从ForkJoin到Stream

import java.util.concurrent.RecursiveTask;

/**
 * 通过ForkJoin并行计算斐波那契start~end的和
 *
 * @author zk
 */
public class MultiThreadFibonacci extends RecursiveTask<Long> {

    private int start;

    private int end;

    /**
     * 当计算10以下的斐波那契, 直接计算(避免创建过多调用栈, 进行优化)
     */
    private static final long THRESHOLD = 10L;

    public MultiThreadFibonacci(int start, int end) {
        this.start = start;
        this.end = end;
    }

    @Override
    protected Long compute() {
        int len = end - start;

        if (len <= THRESHOLD) {
            long sum = 0;
            for (int i = start; i <= end; ++i) {
                sum += fibByFormula(i);
            }
            return sum;
        } else {
            int mid = start + (end - start) / 2;
            MultiThreadFibonacci left = new MultiThreadFibonacci(start, mid);
            left.fork();

            MultiThreadFibonacci right = new MultiThreadFibonacci(mid + 1, end);
            right.fork();

            return left.join() + right.join();
        }
    }

    /**
     * 通过公式计算fib(n)
     */
    private static long fibByFormula(int n) {
        return (long) ((Math.pow((1 + Math.sqrt(5)) / 2, n) - Math.pow((1 - Math.sqrt(5)) / 2, n)) / Math.sqrt(5));
    }
}

import algorithm.util.iostream.StdOut;
import org.junit.Test;

import static algorithm.basic.Fibonacci.fibByFormula;

public class MultiThreadFibonacciTest {

    @Test
    public void compute() {
        int left = 1, right = 50;
        MultiThreadFibonacci f = new MultiThreadFibonacci(left, right);
        long res = f.compute();
        System.out.println(String.format("The sum of %dth~%dth fibonacci is: %d", left, right, res));
    }

    /**
     * 对比单线程和并行的斐波那契
     */
    @Test
    public void compare() {
        int left = 1, right = 100000000;
        long res1 = 0, res2;

        long current = System.currentTimeMillis();
        for (int i = left; i <= right; i++) {
            res1 += fibByFormula(i);
        }
        StdOut.printf("%s (%d milliseconds)\n", "fibByFormula:", System.currentTimeMillis() - current);

        current = System.currentTimeMillis();
        MultiThreadFibonacci f = new MultiThreadFibonacci(left, right);
        res2 = f.compute();
        StdOut.printf("%s (%d milliseconds)\n", "MultiThreadFibonacci:", System.currentTimeMillis() - current);

        System.out.println(String.format("The sum of %dth~%dth fibonacci is: %d", left, right, res2));
        assert res1 == res2;
    }
}

fibByFormula: (4547 milliseconds)
MultiThreadFibonacci: (960 milliseconds)
The sum of 1th~100000000th fibonacci is: 1293530146058730797

import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.RecursiveAction;

/**
 * 归并排序的多线程版本
 *
 * @author zk
 */
public class ParallelMergeSort extends BaseSort {

    @SuppressWarnings("unchecked")
    public static <K extends Comparable<K>> void parallelMergeSort(K[] arr) {
        K[] aux = (K[]) new Comparable[arr.length];

        ForkJoinPool pool = new ForkJoinPool();
        // 创建任务
        RecursiveAction mainTask = new MergeTask<>(arr, aux, 0, arr.length - 1);
        // 执行任务
        pool.invoke(mainTask);
    }

    public static class MergeTask<K extends Comparable<K>> extends RecursiveAction {
        /**
         * 小数组排序优化:
         * <p>
         * 小于500的子数组不再使用递归调用
         * <p>
         * 而是直接使用插入排序
         */
        private final int THRESHOLD = 15;

        private K[] arr;

        private K[] tmp;

        private int start;

        private int end;

        public MergeTask(K[] arr, K[] tmp, int start, int end) {
            this.arr = arr;
            this.tmp = tmp;
            this.start = start;
            this.end = end;
        }

        @Override
        protected void compute() {
            if (end - start <= THRESHOLD) {
                insertionSort(arr, start, end);
            } else {
                // 获取左一半和右一半
                int mid = start + (end - start) / 2;
                MergeTask leftTask = new MergeTask<>(arr, tmp, start, mid);
                MergeTask rightTask = new MergeTask<>(arr, tmp, mid + 1, end);

                // Recursively sort the two halves
                leftTask.fork();
                rightTask.fork();

                leftTask.join();
                rightTask.join();

                // Merge firstHalf with second
                merge(arr, tmp, start, mid + 1, end);
            }
        }
    }

    /**
     * 不使用辅助数组进行复制归并(但是仍需要空间)
     *
     * @param arr   源数组
     * @param tmp   目标数组
     * @param left  归并区间左边界
     * @param mid   归并区间切分位置
     * @param right 归并右边界
     */
    private static <K extends Comparable<K>> void merge(K[] arr, K[] tmp, int left, int mid, int right) {
        int leftEnd = mid - 1, tmpPos = left;
        int numElements = right - left + 1;

        while (left <= leftEnd && mid <= right) {
            if (less(arr[left], arr[mid]))
                tmp[tmpPos++] = arr[left++];
            else
                tmp[tmpPos++] = arr[mid++];
        }

        while (left <= leftEnd)
            tmp[tmpPos++] = arr[left++];

        while (mid <= right)
            tmp[tmpPos++] = arr[mid++];

        for (int i = 0; i < numElements; i++, right--)
            arr[right] = tmp[right];
    }

    /**
     * 使用插排排序a[lo...hi]子区间
     *
     * @param a  排序数组
     * @param lo 排序左边界
     * @param hi 排序右边界
     */
    private static <K extends Comparable<K>> void insertionSort(K[] a, int lo, int hi) {
        for (int i = lo; i <= hi; i++)
            for (int j = i; j > lo && less(a[j], a[j - 1]); j--)
                exch(a, j, j - 1);
    }
}

public class ParallelMergeSortTest {
    @Test
    public void parallelMergeSortTest() {
        // 排序大小
        final int sortSize = 1000000;

        Integer[] arr1 = RandomArrayUtil.getRandomBoxedIntArray(sortSize);
        System.out.println("Arrays created!");

        long startTime = System.currentTimeMillis();
        parallelMergeSort(arr1);
        long endTime = System.currentTimeMillis();
        System.out.println("Sequent time is: " + (endTime - startTime) + " milliseconds");
        assert BaseSort.isSorted(arr1);
    }
}

public void streamSortTest() {
    // 排序大小
    final int sortSize = 1000000;
    
    // 通过stream和一个Integer数组创建一个list
    List<Integer> list = Arrays
        .stream(RandomArrayUtil.getRandomBoxedIntArray(sortSize))
        .collect(Collectors.toList());
    System.out.println("Arrays created!");

    long startTime = System.currentTimeMillis();
    int[] res = list.stream().parallel().mapToInt(Integer::intValue).sorted().toArray();
    long endTime = System.currentTimeMillis();
    System.out.println("ParallelStream time is: " + (endTime - startTime) + " milliseconds");
    assert BaseSort.isSorted(res);
}

public class ParallelMergeSortTest {
    /**
     * 比较单线程归并排序, ForkJoin并发排序和Stream流排序
     */
    @Test
    public void compareSort() {
        // 排序大小
        final int sortSize = 100000000;

        Integer[] arr1 = RandomArrayUtil.getRandomBoxedIntArray(sortSize);
        Integer[] arr2 = Arrays.copyOf(arr1, arr1.length);
        List<Integer> arr3 = Arrays.stream(arr1).collect(Collectors.toList());
        System.out.println("Arrays created!");

        long startTime = System.currentTimeMillis();
        MergeSort.sortTopDown(arr2);
        long endTime = System.currentTimeMillis();
        System.out.println("Sequent time is: " + (endTime - startTime) + " milliseconds");
        assert BaseSort.isSorted(arr2);

        startTime = System.currentTimeMillis();
        parallelMergeSort(arr1);
        endTime = System.currentTimeMillis();
        System.out.println("Parallel time is: " + Runtime.getRuntime().availableProcessors() + " processors is " + (endTime - startTime) + " milliseconds");
        assert BaseSort.isSorted(arr1);

        startTime = System.currentTimeMillis();
        int[] res = arr3.stream().parallel().mapToInt(Integer::intValue).sorted().toArray();
        endTime = System.currentTimeMillis();
        System.out.println("ParallelStream time is: " + (endTime - startTime) + " milliseconds");
        assert BaseSort.isSorted(res);
    }
}