在 Scala 未来阻塞？答案

【问题标题】：Blocking in scala future?在 Scala 未来阻塞？
【发布时间】：2012-08-14 17:32:21
【问题描述】：

我有一个很长的字符串序列，它们需要由某个处理函数单独处理，然后作为另一个序列对象收集。这个问题似乎非常适合 fork/join 类型的攻击。

该函数是一个实例化非常昂贵的类的成员。但是在 future 之间实例化和共享单个类对象似乎会导致问题，因此我实例化了可用处理器数量的 4 倍，然后将它们拆分到 future 之间。

// instantiate processing class objects
val processors = 1 to SomeNumber map (x=> new MyProcessor)
val processorstream = Stream.continually(processors).flatten
// the input strings
val input: Seq[String] = some sequence of strings
val splitinput = input.grouped(some large number).toStream
// create futures
val mytask = splitinput.zip(processorstream).collect {
    case (subseq of strings, processor) => future {
        map elements of subsequence of strings with processor}}

然后我像这样收集输出

val result = mytask.map(x => x.apply()).reduce(_++_) // or some appropriate concatenation operator

我的问题是，即使我有 8 个内核，这也不能让我充分利用 CPU。它只使用一个核心。

为了调查，我尝试过的另一种方法是

val input: Seq[String] = some sequence of strings
// no stage where I split the input into subsequences
val mytask = input.zip(processorstream).collect {
    case (string, processor) => future {
        process string with processor}}
val result = mytask.map(x => x.apply())

这种替代方法既有效又无效。它实现了充分的 cpu 利用率，但抛出了几个异常，因为（假设）处理器在每个字符串中运行的速度太快，有时相同的处理器对象会同时应用于不同的字符串。

我更加确信我的假设是处理器工作速度过快，因为如果我提供更长的输入（例如，整个文本文档而不是 10 个单词的标题），我可以充分利用 CPU，而不会引发任何异常。

我也尝试过使用 akka futures 和 scalaz promises，当我将输入序列拆分为子序列时，它们似乎都只使用一个 cpu。

那么在这种情况下，如何在使用字符串子序列作为输入的情况下充分利用期货？

【问题讨论】：

为什么不使用并行集合中的 .par？
@om-nom-nom 那些是什么？它在这里如何应用？几个月前我才拿起 scala。
并行集合并行执行像map 这样的操作。见docs.scala-lang.org/overviews/parallel-collections/…

标签： scala concurrency actor future

【解决方案1】：

每@om-nom-nom：

input.par.map { s => task(s) }

【讨论】：

而且你甚至会丢失一些噪音：input.par 地图任务
Viktor 这是你度假的想法吗？ :)

【解决方案2】：

您可以尝试将ThreadLocal 用于您的可变处理器。比较没用的例子：

val words = io.Source.fromFile("/usr/share/dict/words").getLines.toIndexedSeq

class Processor {
  val sb = new StringBuffer() // mutable!
  println("---created processor---")
  def map(s: String): Int = {
    sb.setLength(0)
    for (i <- 1 to s.length()) {
      sb.append(s.substring(0, i))
    }
    sb.toString().sum.toInt  // don't make any sense out of this
  }
}

val tl = new ThreadLocal[Processor] {
  override protected def initialValue() = new Processor
}

val parRes = words.par.map(w => tl.get.map(w)).sum
val serRes = words.map(    w => tl.get.map(w)).sum
assert(parRes == serRes)

默认情况下，这将创建与 CPU 内核一样多的线程，正如 ---created processor--- 消息所证明的那样。

【讨论】：