过滤Scala的并行集合,并在发现所需的结果数量时提前中止

尝试在线程安全的“外部”数据结构中累积中间匹配或保留具有结果计数的外部AtomicInteger在4个内核上的使用速度似乎比使用常规collection.mutable.HashMap慢2到3倍,并将单个内核与100 ％.

我知道在Par *集合上查找或存在会在内部中止.有没有一种方法可以推广这个来找到多个结果？

这里的代码似乎比ParHashMap的速度慢了2到3倍,而且还有一个问题,比较多的maxResults结果进入结果CHM(这可能是因为线程被preemptAndGet抢占之前但之前的允许其他线程在)中添加更多的元素.更新：似乎减速是由于在counter.incrementAndGet()上的工作线程竞争,这当然违反了整个并行扫描的目的:-(

def find(filter: Node => Boolean,maxResults: Int): Iterable[Node] =
{
  val counter = new AtomicInteger(0)
  val results = new ConcurrentHashMap[Key,Node](maxResults)

  import util.control.Breaks._

  breakable
  {
    for ((key,node) <- parHashMap if filter(node))
    {
      results.put(key,node)
      val total = counter.incrementAndGet()
      if (total > maxResults) break
    }
  }

  results.values.toArray(new Array[Node](results.size))
}