Java 8与availableProcessors()续

接上文Java 8与Runtime.getRuntime().availableProcessors().

通用池的并发数一般都是availableProcessors() - 1，除非我们通过系统属性指定了它的值。然而，如果你在一个单核的机器上运行Java的话，你会把这个通用池的并发数设置成1，更准确的值其实应该是0。你可能会想，地球上还有谁有单核的机器？事实上，现在给开发人员提供共享的虚拟开发环境已经越来越普遍了，这种通常都是配备单核CPU，但内存会很大。在虚拟机里，单核并不是什么稀奇事。问题在于像parallelSort这样的代码试图去以并行的方式来进行排序，而不是传统的老的排序机制。在Arrays.parallelSort() 方法中，它会去查看 ForkJoinPool.getCommonPoolParallelism()的值。由于在单核和双核的机器上这个值都是1，它会调用默认的Arrays.sort()方法。因此我认为说在双核机器上能有30%的性能提升的说法是扯淡的。

然而，对并行流来说情况会好些。由于在双核系统里，有一个大小为1的 Fork/Join池，实际上我们有两个线程在工作：1，线程池里的线程，2，提交任务的线程。这里我犯了个很常见的错误，就是让我统计教室里有多少人时我忘了把我自己算进去 了。其实parallel()还是有效果的，多多少少吧。根据availableProcessors()返回的值可以创建出合适数量的工作线程。除了在单核的机器上，因为这会使用两个线程而不是和物理处理器数对应的一个线程。

Pierre-yves Saumont给我发过来一份代码，这能证明parallel()默认是使用availableProcessors()返回的值（注：而不是availableProcessors()-1）。

import java.util.*;
import java.util.concurrent.*;
import java.util.stream.*;

public class CountThreads {
  private static Map<String, Integer> map = new ConcurrentHashMap<>();

  public static void main(String... args) {
    System.out.println(IntStream.range(1, 1_000_000).
        parallel().filter(CountThreads::isPrime).count());
    map.forEach((k, v) -> System.out.println(k + ": " + v));
  }

  public static boolean isPrime(int n) {
    map.compute(Thread.currentThread().getName(),
        (k, v) -> v == null ? 1 : v + 1);
    if (n % 2 == 0) return false;
    for (int i = 3; i * i <= n; i += 2) {
      if (n % i == 0) {
        return false;
      }
    }
    return true;
  }
}
 

在一台双核以及单核的机器 上，可以看到下面的输出：

78498
ForkJoinPool.commonPool-worker-1: 499999
main: 500000
 

在我的1-4-2机器上，有8个超线程，我们可以看到有8个线程在工作：

78498
ForkJoinPool.commonPool-worker-7: 125000
ForkJoinPool.commonPool-worker-1: 125000
ForkJoinPool.commonPool-worker-2: 125000
main: 125000
ForkJoinPool.commonPool-worker-5: 125000
ForkJoinPool.commonPool-worker-6: 156249
ForkJoinPool.commonPool-worker-3: 125000
ForkJoinPool.commonPool-worker-4: 93750
 

如果你想测试类似并行排序的性能，这需要大量的内存访问，你可能会发现你的超线程能提供接近线性的性能提升。这是由于最大的瓶颈在于填充缓存行的速度。超线程能允许它并行地执行。查找素数对内存的依赖不多，因此物理核数的增加可能对性能的提升并不大。

除此之外，parallelSort()排序的算法可能有些不同。我测试的一些特定的数据集下，单线程的Arrays.sort()比8线程的Arrays.paralleclSort()的还要快，尤其是当数据集已经排好序的情况下。

最后，我用Fork/Join框架写了一段CountThreads的代码。它大概是这样的：

import java.util.concurrent.*;

public class CountThreadsForkJoin {
  public static void main(String... args) {
    System.out.println(ForkJoinPool.commonPool().invoke(
        new PrimeSearcher(1, 1_000_000)));
  }

  private static class PrimeSearcher extends RecursiveTask<Integer> {
    private static final int THRESHOLD = 10000;
    private final int from;
    private final int to;

    public PrimeSearcher(int from, int to) {
      this.from = from;
      this.to = to;
    }

    protected Integer compute() {
      if (to - from < THRESHOLD) {
        return serialCount();
      }
      int from0 = from;
      int to0 = from + (to - from) / 2;
      int from1 = to0 + 1;
      int to1 = to;
      PrimeSearcher ps0 = new PrimeSearcher(from0, to0);
      PrimeSearcher ps1 = new PrimeSearcher(from1, to1);
      ps0.fork();
      return ps1.invoke() + ps0.join();
    }

    private Integer serialCount() {
      int count = 0;
      for (int i = from; i < to; i++) {
        if (CountThreads.isPrime(i)) count++;
      }
      return count;
    }
  }
}
 

有人觉得Java 8的并行流需要手动写 Fork/Join，这个算不上什么提高，我倒是想去会会他们。

import java.util.stream.*;

public class CountThreadsStream {
  public static void main(String... args) {
    System.out.println(IntStream.range(1, 1_000_000).
        parallel().filter(CountThreads::isPrime).count());
  }
}
 

时间会告诉你现实中parallel()究竟会有多少用处。


原创文章转载请注明出处：http://it.deepinmind.com

英文原文链接