java - 如何为 Scala ExecutionContexts 配置关闭策略？

Question

我最近在一台机器上遇到了一些奇怪的行为，其中 map 返回 Future[T] 的函数是按顺序执行的。同样的问题不会发生在其他机器上:工作如人们所期望的那样交错进行。后来我发现这可能是因为 Scala 有点太聪明了，它选择了一个与机器资源相匹配的 ExecutionContext:一个核心，一个 worker 。

下面是一些重现问题的简单代码:

import scala.concurrent._
import scala.concurrent.duration._

val ss = List("the", "quick", "brown", "fox",
              "jumped", "over", "the", "lazy", "dog")

def r(s: String) : Future[String] = future {
        for (i <- 1 to 10) yield {
                println(String.format("r(%s) waiting %s more seconds...",
                        s, (10 - i).toString))
                Thread.sleep(1000)

        }
        s.reverse
}

val f_revs = ss.map { r(_) }

println("Look ma, no blocking!")

val rev = f_revs.par.map { Await.result(_, Duration.Inf) }.mkString(" ")

println(rev)

在表现出奇怪行为的机器上运行它会产生这样的顺序输出:

Look ma, no blocking!
r(the) waiting 9 more seconds...
r(the) waiting 8 more seconds...
r(the) waiting 7 more seconds...

提供自定义ExecutionContext:

val pool = Executors.newFixedThreadPool(1)
implicit val ec = ExecutionContext.fromExecutor(pool)

允许线程在这台机器上交错。但是我现在有一个新问题:线程池没有关闭，导致程序无限期挂起。显然，这是 expected behavior对于 FixedThreadPool，我可以通过将 pool.shutdown() 放在某个地方来关闭它。

尽管我固执，但我不想让线程池关闭。有没有办法将池配置为在所有队列为空时(可能在一些延迟之后)关闭，就像它与默认池一起工作一样？我查看了 ExecutionContext 文档，但没有找到我要查找的内容。

Answer 1

Scala 使用其自己的 fork-join 实现，其行为与 Java 不同，因此默认 ExecutionContext 与您使用 Executors 创建的行为不同。

一种更简单的方法是设置以下系统属性来配置默认的 ExecutionContext:

1. scala.concurrent.context.minThreads 施加最小线程数。默认为 1。
2. scala.concurrent.context.numThreads 设置线程数。默认为 x1。
3. scala.concurrent.context.maxThreads 施加最大线程数。默认为 x1。

这些中的每一个都可以是一个数字，或者一个前面有 x 的数字，以指示处理器数量的倍数。要增加线程数，您必须更改 numThreads 和 maxThreads。在您的情况下，将两者都设置为 x2 应该可以。