java - 将 ExecutorService 与合并排序的多线程版本一起使用

Question

我正在处理一个家庭作业问题，我必须创建一个多线程版本的合并排序。我能够实现它，但我无法停止线程的创建。我研究过使用 ExecutorService 来限制线程的创建，但我不知道如何在我当前的代码中实现它。

这是我当前的多线程合并排序。我们需要实现一个特定的策略模式，这就是我的 sort() 方法的来源。

@Override
public int[] sort(int[] list) {
    int array_size = list.length;
    list = msort(list, 0, array_size-1);
    return list;
}

int[] msort(int numbers[], int left, int right) {
    final int mid;
    final int leftRef = left;
    final int rightRef = right;
    final int array[] = numbers;
    if (left<right) {
        mid = (right + left) / 2;
        //new thread
        Runnable r1 = new Runnable(){
            public void run(){
                msort(array, leftRef, mid);     
            }
        };
        Thread t1 = new Thread(r1);
        t1.start();

        //new thread
        Runnable r2 = new Runnable(){
            public void run(){
                msort(array, mid+1, rightRef);
            }
        };
        Thread t2 = new Thread(r2);
        t2.start();
        //join threads back together
        try {
            t1.join();
            t2.join();

        } catch (InterruptedException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }
        merge(numbers, leftRef, mid, mid+1, rightRef);
    }
    return numbers;
}

void merge(int numbers[], int startA, int endA, int startB, int endB) {
    int finalStart = startA;
    int finalEnd = endB;
    int indexC = 0;
    int[] listC = new int[numbers.length];

    while(startA <= endA && startB <= endB){
        if(numbers[startA] < numbers[startB]){
            listC[indexC] = numbers[startA];
            startA = startA+1;
        }
        else{
            listC[indexC] = numbers[startB];
            startB = startB +1;
        }
        indexC++;
    }

    if(startA <= endA){
        for(int i = startA; i < endA; i++){
            listC[indexC]= numbers[i];
            indexC++;
        }
    }

    indexC = 0;
    for(int i = finalStart; i <= finalEnd; i++){
        numbers[i]=listC[indexC];
        indexC++;
    }
}

如有指点，将不胜感激。

Answer 1

根据@mcdowella 的评论，如果您想限制并行运行的线程数，我还认为 fork/join 框架是您的最佳选择。

我知道这对你的作业没有任何帮助，因为你可能不允许使用 Java7 中的 fork/join 框架。然而，它即将学习一些东西，不是吗？;)

正如我评论的那样，我认为您的合并方法是错误的。我无法查明失败的原因，但我已经重写了它。我强烈建议您编写一个测试用例，其中包含该合并方法期间可能发生的所有边缘情况，如果您验证它有效，请将其植入到您的多线程代码中。

@lbalazscs 还提示您在 javadocs 中提到了 fork/join 排序，但是我无事可做 - 所以如果您使用 Java7 实现它，我将向您展示解决方案。

public class MultithreadedMergeSort extends RecursiveAction {

  private final int[] array;
  private final int begin;
  private final int end;

  public MultithreadedMergeSort(int[] array, int begin, int end) {
    this.array = array;
    this.begin = begin;
    this.end = end;
  }

  @Override
  protected void compute() {
    if (end - begin < 2) {
      // swap if we only have two elements
      if (array[begin] > array[end]) {
        int tmp = array[end];
        array[end] = array[begin];
        array[begin] = tmp;
      }
    } else {
      // overflow safe method to calculate the mid
      int mid = (begin + end) >>> 1;
      // invoke recursive sorting action
      invokeAll(new MultithreadedMergeSort(array, begin, mid),
          new MultithreadedMergeSort(array, mid + 1, end));
      // merge both sides
      merge(array, begin, mid, end);
    }
  }

  void merge(int[] numbers, int startA, int startB, int endB) {
    int[] toReturn = new int[endB - startA + 1];
    int i = 0, k = startA, j = startB + 1;
    while (i < toReturn.length) {
      if (numbers[k] < numbers[j]) {
        toReturn[i] = numbers[k];
        k++;
      } else {
        toReturn[i] = numbers[j];
        j++;
      }
      i++;
      // if we hit the limit of an array, copy the rest
      if (j > endB) {
        System.arraycopy(numbers, k, toReturn, i, startB - k + 1);
        break;
      }
      if (k > startB) {
        System.arraycopy(numbers, j, toReturn, i, endB - j + 1);
        break;
      }
    }
    System.arraycopy(toReturn, 0, numbers, startA, toReturn.length);
  }

  public static void main(String[] args) {
    int[] toSort = { 55, 1, 12, 2, 25, 55, 56, 77 };
    ForkJoinPool pool = new ForkJoinPool();
    pool.invoke(new MultithreadedMergeSort(toSort, 0, toSort.length - 1));
    System.out.println(Arrays.toString(toSort));

  }

请注意，线程池的构造将 Activity 并行线程的数量限制为处理器的核心数量。

ForkJoinPool pool = new ForkJoinPool();

根据它的javadoc:

Creates a ForkJoinPool with parallelism equal to java.lang.Runtime.availableProcessors, using the default thread factory, no UncaughtExceptionHandler, and non-async LIFO processing mode.

另请注意我的合并方法与您的方法有何不同，因为我认为这是您的主要问题。如果我用我的方法替换您的合并方法，至少您的排序有效。