2.2 归并排序动图演示

2.3 使用归并排序实现上面的算法题

public class MergeSort {

    private final int[] arrayToSort; //要排序的数组
    private final int threshold; //拆分的阈值，低于此阈值就不再进行拆分

    public MergeSort(final int[] arrayToSort, final int threshold) {
        this.arrayToSort = arrayToSort;
        this.threshold = threshold;
    }

    /**
     * 排序
     *
     * @return
     */
    public int[] sequentialSort() {
        return sequentialSort(arrayToSort, threshold);
    }

    public static int[] sequentialSort(final int[] arrayToSort, int threshold) {
        //拆分后的数组长度小于阈值，直接进行排序
        if (arrayToSort.length < threshold) {
            //调用jdk提供的排序方法
            Arrays.sort(arrayToSort);
            return arrayToSort;
        }

        int midpoint = arrayToSort.length / 2;
        //对数组进行拆分
        int[] leftArray = Arrays.copyOfRange(arrayToSort, 0, midpoint);
        int[] rightArray = Arrays.copyOfRange(arrayToSort, midpoint,
                arrayToSort.length);
        //递归调用
        leftArray = sequentialSort(leftArray, threshold);
        rightArray = sequentialSort(rightArray, threshold);
        //合并排序结果
        return merge(leftArray, rightArray);
    }

    public static int[] merge(final int[] leftArray, final int[] rightArray) {
        //定义用于合并结果的数组
        int[] mergedArray = new int[leftArray.length + rightArray.length];
        int mergedArrayPos = 0;
        int leftArrayPos = 0;
        int rightArrayPos = 0;
        while (leftArrayPos < leftArray.length && rightArrayPos < rightArray.length) {
            if (leftArray[leftArrayPos] <= rightArray[rightArrayPos]) {
                mergedArray[mergedArrayPos] = leftArray[leftArrayPos];
                leftArrayPos++;
            } else {
                mergedArray[mergedArrayPos] = rightArray[rightArrayPos];
                rightArrayPos++;
            }
            mergedArrayPos++;
        }

        while (leftArrayPos < leftArray.length) {
            mergedArray[mergedArrayPos] = leftArray[leftArrayPos];
            leftArrayPos++;
            mergedArrayPos++;
        }

        while (rightArrayPos < rightArray.length) {
            mergedArray[mergedArrayPos] = rightArray[rightArrayPos];
            rightArrayPos++;
            mergedArrayPos++;
        }

        return mergedArray;
    }
}

Fork/Join并行归并排序

public class MergeSortTask extends RecursiveAction {

    private final int threshold; //拆分的阈值，低于此阈值就不再进行拆分
    private int[] arrayToSort; //要排序的数组

    public MergeSortTask(final int[] arrayToSort, final int threshold) {
        this.arrayToSort = arrayToSort;
        this.threshold = threshold;
    }

    @Override
    protected void compute() {
        //拆分后的数组长度小于阈值，直接进行排序
        if (arrayToSort.length <= threshold) {
            // 调用jdk提供的排序方法
            Arrays.sort(arrayToSort);
            return;
        }

        // 对数组进行拆分
        int midpoint = arrayToSort.length / 2;
        int[] leftArray = Arrays.copyOfRange(arrayToSort, 0, midpoint);
        int[] rightArray = Arrays.copyOfRange(arrayToSort, midpoint, arrayToSort.length);

        MergeSortTask leftTask = new MergeSortTask(leftArray, threshold);
        MergeSortTask rightTask = new MergeSortTask(rightArray, threshold);

        //调用任务
        invokeAll(leftTask, rightTask);

        // 合并排序结果
        arrayToSort = MergeSort.merge(leftTask.getSortedArray(),
                rightTask.getSortedArray());
    }

    public int[] getSortedArray() {
        return arrayToSort;
    }
}

        在这个示例中，我们使用Fork/Join框架实现了归并排序算法，并通过递归调用实现了并行化。使用 Fork/Join框架实现归并排序算法的关键在于将排序任务分解成小的任务，使用Fork/Join框架将这些小 任务提交给线程池中的不同线程并行执行，并在最后将排序后的结果进行合并。这样可以充分利用多 核CPU的并行处理能力，提高程序的执行效率。

测试结果对比

测试代码

public class Utils {

    /**
     * 随机生成数组
     *
     * @param size 数组的大小
     * @return
     */
    public static int[] buildRandomIntArray(final int size) {
        int[] arrayToCalculateSumOf = new int[size];
        Random generator = new Random();
        for (int i = 0; i < arrayToCalculateSumOf.length; i++) {
            arrayToCalculateSumOf[i] = generator.nextInt(100000000);
        }
        return arrayToCalculateSumOf;
    }
}

public class ArrayToSortMain {

    public static void main(String[] args) {
        //生成测试数组 用于归并排序
        int[] arrayToSortByMergeSort = Utils.buildRandomIntArray(20000000);
        //生成测试数组 用于forkjoin排序
        int[] arrayToSortByForkJoin = Arrays.copyOf(arrayToSortByMergeSort,
                arrayToSortByMergeSort.length);
        //获取处理器数量
        int processors = Runtime.getRuntime().availableProcessors();

        MergeSort mergeSort = new MergeSort(arrayToSortByMergeSort, processors);
        long startTime = System.nanoTime();
        // 归并排序
        mergeSort.mergeSort();
        long duration = System.nanoTime() - startTime;
        System.out.println("单线程归并排序时间: " + (duration / (1000f * 1000f)) + "毫秒");


        //利用forkjoin排序
        MergeSortTask mergeSortTask = new MergeSortTask(arrayToSortByForkJoin,
                processors);
        //构建forkjoin线程池
        ForkJoinPool forkJoinPool = new ForkJoinPool(processors);
        startTime = System.nanoTime();
        //执行排序任务
        forkJoinPool.invoke(mergeSortTask);
        duration = System.nanoTime() - startTime;
        System.out.println("forkjoin排序时间: " + (duration / (1000f * 1000f)) + "毫秒");

    }
}

根据测试结果可以看出，数组越大，利用Fork/Join框架实现的并行化归并排序比单线程归并排序的效 率更高：

2.4 并行实现归并排序的优化和注意事项

在实际应用中，我们需要考虑数据分布的均匀性、内存使用情况、线程切换开销等因素，以充分利用多核CPU并保证算法的正确性和效率。

以下是并行实现归并排序的一些优化和注意事项：

任务的大小： 任务大小的选择会影响并行算法的效率和负载均衡 ，如果任务太小，会造成任务划分和合并的开销过大；如果任务太大，会导致任务无法充分利用多核CPU并行处理能力。因此，在实际应用中需要根据数据量、CPU核心数等因素选择合适的任务大小。

负载均衡： 并行算法需要保证负载均衡 ，即各个线程执行的任务大小和时间应该尽可能相等，否则会导致某些线 程负载过重，而其他线程负载过轻的情况。在归并排序中，可以通过递归调用实现负载均衡，但是需要注意递归 的层数不能太深，否则会导致任务划分和合并的开销过大。

数据分布： 数据分布的均匀性也会影响并行算法的效率和负载均衡。 在归并排序中，如果数据分布不均匀，会导 致某些线程处理的数据量过大，而其他线程处理的数据量过小的情况。因此，在实际应用中需要考虑数据的分布 情况，尽可能将数据分成大小相等的子数组。

内存使用： 并行算法需要考虑内存的使用情况 ，特别是在处理大规模数据时，内存的使用情况会对算法的执行效 率产生重要影响。在归并排序中，可以通过对数据进行原地归并实现内存的节约，但是需要注意归并的实现方 式，以避免数据的覆盖和不稳定排序等问题。

线程切换： 线程切换是并行算法的一个重要开销 ，需要尽量减少线程的切换次数，以提高算法的执行效率。在归 并排序中，可以通过设置线程池的大小和调整任务大小等方式控制线程的数量和切换开销，以实现算法的最优性能。

3. Fork/Join框架介绍

3.1 什么是Fork/Join

Fork/Join是一个是一个并行计算的框架，主要就是 用来支持分治任务模型 的，这个计算框架里的 Fork 对应的是分治任务模型里的任务分解，Join 对应的是结果合并 。它的核心思想是将一个大任务分成许多小任务，然后并行执行这些小任务，最终将它们的结果合并成一个大的结果

3.2 应用场景

Fork/Join框架的应用场景包括以下几个方面：

1. 递归分解型任务

Fork/Join框架特别适用于递归分解型的任务，例如排序、归并、遍历等。这些任务通常可以将大的任 务分解成若干个子任务，每个子任务可以独立执行，并且可以通过归并操作将子任务的结果合并成一 个有序的结果。

2. 数组处理

Fork/Join框架还可以用于数组的处理，例如数组的排序、查找、统计等。在处理大型数组时，

Fork/Join框架可以将数组分成若干个子数组，并行地处理每个子数组，最后将处理后的子数组合并成 一个有序的大数组。

3. 并行化算法

Fork/Join框架还可以用于并行化算法的实现，例如并行化的图像处理算法、并行化的机器学习算法

等。在这些算法中，可以将问题分解成若干个子问题，并行地解决每个子问题，然后将子问题的结果 合并起来得到最终的解决方案。

4. 大数据处理

Fork/Join框架还可以用于大数据处理，例如大型日志文件的处理、大型数据库的查询等。在处理大数 据时，可以将数据分成若干个分片，并行地处理每个分片，最后将处理后的分片合并成一个完整的结果。

3.3 Fork/Join使用

Fork/Join框架的主要组成部分是ForkJoinPool、ForkJoinTask。ForkJoinPool是一个线程池，它用于 管理ForkJoin任务的执行。ForkJoinTask是一个抽象类，用于表示可以被分割成更小部分的任务。

ForkJoinPool

ForkJoinPool是Fork/Join框架中的线程池类，它用于管理Fork/Join任务的线程。ForkJoinPool类包

括一些重要的方法，例如submit()、invoke()、shutdown()、awaitTermination()等，用于提交任

务、执行任务、关闭线程池和等待任务的执行结果。ForkJoinPool类中还包括一些参数，例如线程池 的大小、工作线程的优先级、任务队列的容量等，可以根据具体的应用场景进行设置。

构造器

ForkJoinPool中有四个核心参数，用于控制线程池的 并行数、工作线程的创建、异常处理和模式指定 等。各参数解释如下：

int parallelism： 指定并行级别（parallelism level） 。ForkJoinPool将根据这个设定，决定工作线程的数量。如 果未设置的话，将使用Runtime.getRuntime().availableProcessors()来设置并行级别；

ForkJoinWorkerThreadFactory factory： ForkJoinPool在创建线程时，会通过factory来创建。注意，这里需要实现的是ForkJoinWorkerThreadFactory，而不是ThreadFactory。如果你不指定factory，那么将由默认的DefaultForkJoinWorkerThreadFactory负责线程的创建工作；

UncaughtExceptionHandler handler： 指定异常处理器，当任务在运行中出错时，将由设定的处理器处理；

boolean asyncMode： 设置队列的工作模式。 当asyncMode为true时，将使用先进先出队列，而为false时则使用后进先出的模式。

//获取处理器数量
 int processors = Runtime.getRuntime().availableProcessors();
//构建forkjoin线程池
 ForkJoinPool forkJoinPool = new ForkJoinPool(processors);

任务提交方式

任务提交是ForkJoinPool的核心能力之一，提交任务有三种方式：

public class BlockingTaskDemo {
    public static void main(String[] args) {
        //构建一个forkjoin线程池
        ForkJoinPool pool = new ForkJoinPool();

        //创建一个异步任务，并将其提交到ForkJoinPool中执行
        CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
            try {
                // 模拟一个耗时的任务
                TimeUnit.SECONDS.sleep(5);
                return "Hello, world!";
            } catch (InterruptedException e) {
                e.printStackTrace();
                return null;
            }
        }, pool);

        try {
            // 等待任务完成，并获取结果
            String result = future.get();

            System.out.println(result);
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ExecutionException e) {
            e.printStackTrace();
        } finally {
            //关闭ForkJoinPool，释放资源
            pool.shutdown();
        }
    }
}

3.4 ForkJoinPool工作原理

工作线程ForkJoinWorkerThread

工作队列WorkQueue