multithreading - Scala 并行集合 : Change default Pool

Question

精简版

我正在寻找一种方法来一劳永逸地设置在调用集合的 .par 函数时全局使用的 Pool...

到目前为止，我只找到了如何在全局 ExecutionContext 中设置线程数，但没有找到如何更改默认使用的实际 Pool。

我只想明确指定 ForkJoinPool 以使并行集合 ExecutionContext 独立于我使用的 Scala 版本。

---

长版这个要求是在我们遇到问题之后出现的，因为 Scala 2.10 不支持 JDK 1.8

Scala 根本无法识别 java 版本并认为我们仍在 1.5，因此池是不同类型并且线程数不受处理器数限制

问题是由这段代码引起的:

if (scala.util.Properties.isJavaAtLeast("1.6")) new ForkJoinTaskSupport
    else new ThreadPoolTaskSupport

def isJavaAtLeast(version: String) = {
    val okVersions = version match {
      case "1.5"    => List("1.5", "1.6", "1.7")
      case "1.6"    => List("1.6", "1.7")
      case "1.7"    => List("1.7")
      case _        => Nil
    }
    okVersions exists (javaVersion startsWith _)
  }

由于我们管理线程的方式在我们的应用程序中非常关键，我们不希望仅更改版本就出现意想不到的惊喜，我想知道是否可以强制 Scala 使用 ForkJoinPool 以及我们决定的预设线程数 全局(我不想要此处描述的单实例解决方案 Scala Parallel Collections: How to know and configure the number of threads )

希望它足够清楚!

Answer 1

从我的角度来看，您的问题包含两个不同的要求:

一个是我只想显式指定 ForkJoinPool 以使并行集合 ExecutionContext 独立于我使用的 Scala 版本。

我不知道这是可能的。最重要的是，我对构造函数 class ForkJoinTaskSupport(val environment: ForkJoinPool) 表示怀疑。这个构造函数是用 ForkJoinPool 调用的，它支持 .par 使用的当前执行上下文，如果我没记错的话，这是全局的。几层之后，我们意识到这个池是在 ExecutionContextImpl 中定义的:

def createExecutorService: ExecutorService = {

    [...]

    val desiredParallelism = range(
      getInt("scala.concurrent.context.minThreads", "1"),
      getInt("scala.concurrent.context.numThreads", "x1"),
      getInt("scala.concurrent.context.maxThreads", "x1"))

    val threadFactory = new DefaultThreadFactory(daemonic = true)

    try {
      new ForkJoinPool(
        desiredParallelism,
        threadFactory,
        uncaughtExceptionHandler,
        true) // Async all the way baby
    } catch {
      [...]
    }
  }

所以它不完全是一个你可以改变的池，但它仍然是一个你绝对可以配置的池，这将解决你的需求的重新制定，也就是我想知道是否有可能强制 Scala 使用 ForkJoinPool 和由我们全局决定的预设线程数

完全免责声明:我从未尝试这样做，因为到目前为止我还不需要它，但你的问题让我想调查一下!