多线程的归并排序算法

Question

我有一个在泛型列表上执行递归合并排序的类，只要元素实现了可比较。我试图让代码多线程来提高性能，为此，我有一个静态变量maxThreads，它可以防止我创建的线程数量爆炸，我有一个静态变量currentThreads，它可以跟踪我当前运行的线程数量。在我的currentThreads变量上似乎有一个竞争条件，但我还没能想出一个解决方案来修复它。

import java.util.ArrayList;
import java.util.List;

public class ThreadedMergeSorter<E extends Comparable<? super E>> implements, Runnable  
{
  private List<E> list;
  private List<E> left, right;
  private Thread t1, t2;
  private static final int maxThreads = 4;
  private static AtomicInteger currentThreads = new AtomicInteger(0);

  private ThreadedMergeSorter(List<E> list)
  {
    this.list = list;
  }

  public ThreadedMergeSorter(){}


  /**
   * Sorts a List<E> using the merge sorting algorithm
   * @param list the list to be merge sorted
   * @return 
   * @throws InterruptedException 
   */
  public void sort(List<E> list) 
  {
    if(list.size() > 1)
    {                  
      left = new ArrayList<E>(list.subList(0, list.size()/2));
      right = new ArrayList<E>(list.subList(list.size()/2, list.size()));

      list.clear();

      if(currentThreads.get() < maxThreads)
      {
        t1 = new Thread(new ThreadedMergeSorter<E>(left));
        t1.start();
        currentThreads.incrementAndGet();
      }
      else sort(left);

      if(currentThreads.get() < maxThreads)
      {
        t2 = new Thread(new ThreadedMergeSorter<E>(right));
        t2.start();
        currentThreads.incrementAndGet();
      }
      else sort(right);

      try{
        if(t1 != null)
        {
          t1.join();
          currentThreads.decrementAndGet();
        }
        if(t2 != null)
        {
          t2.join();
          currentThreads.decrementAndGet();
        }
      }catch(InterruptedException e){}

      list.addAll(mergeSortedLists(left, right)); 
    } 
  }

  /**
   * Merges two previously sorted List<E extends Comparable<E>> into a single List
   * @param leftArray a List of previously sorted elements
   * @param rightArray a List of previously sorted elements
   * @return an new sorted List
   */
  private List<E> mergeSortedLists(List<E> leftList, List<E> rightList)
  {
    ArrayList<E> list = new ArrayList<E>();

    while(!leftList.isEmpty() && !rightList.isEmpty())
    {
      if((leftList.get(0)).compareTo(rightList.get(0)) <= 0)
        list.add(leftList.remove(0));        
      else
        list.add(rightList.remove(0));
    }

    if(!leftList.isEmpty())
      list.addAll(leftList);
    if(!rightList.isEmpty())
      list.addAll(rightList);

    return list;
  }


  @Override
  public void run() 
  {
    sort(this.list);
  }
}

问题出在由if语句和try catch块组成的sort(List<E> list)方法中。

Answer 1

首先，你没有并行运行任何东西。线程是用start()启动的，而不是用run()启动的，后者只是在当前线程上调用run方法。

其次，如果您有正在更新的共享变量，请尝试将它们声明为AtomicInteger

private static AtomicInteger currentThreads = new AtomicInteger(0);

然后使用这些方法自动递增/递减：

currentThreads.incrementAndGet();
currentThreads.decrementAndGet();

Answer 2

不要不断地创建、终止和销毁线程。不要试图对线程进行微管理--正如你已经发现的那样，这是非常困难和容易出错的。

如果您想对合并排序进行线程化(这不是一个坏主意：)，那么可以看看ThreadPoolExecutor和CountDownLatch。

Answer 3

如果您使用的是Java7，我建议您使用新的Fork/Join，并使用AtomicReferenceArray<E>而不是List，这样您就可以以线程安全的方式就地进行排序。