Scala中使用List.par.aggregate时程序挂起问题咨询

Hey there! Let's break down why your parallel aggregate code is hanging, and how to fix it.

The Root Cause: Sharing a Single Accumulator Across Parallel Partitions #

First, let's recap how aggregate works with parallel collections:

• Parallel collections split your sequence into multiple partitions, each processed by a separate thread.
• For every partition, it starts with an initial accumulator (the z parameter you pass), then runs the seqop function to build up that partition's result.
• Finally, it uses the combop function to merge all partition-specific results into one final value.

Your critical mistake here is passing a pre-instantiated ConcurrentHashMap as the z parameter. Since z is a by-name parameter (=> B), it’s designed to create a fresh accumulator for each partition. When you pass a fixed map instance instead, all threads end up sharing the exact same map.

This breaks the parallel execution model of aggregate:

• While ConcurrentHashMap is thread-safe, having all threads modify the same map during seqop creates confusion in the combop phase. The framework tries to merge "partition results" that are actually the same map instance, which can trigger infinite waits or deadlock-like behavior in the fork-join task scheduler—this is why your code hangs.

The Fix: Create a New Accumulator for Each Partition #

Instead of passing an existing map, pass a factory expression that creates a new ConcurrentHashMap for every partition. Then update your combop to properly merge the separate partition maps into one final result.

Here's the corrected code:

val s = Seq("apple", "oranges", "apple", "banana", "apple", "oranges", "oranges")

val result = s.par.aggregate(new java.util.concurrent.ConcurrentHashMap[String, Int]())(
  // seqop: Update the current partition's map with the item count
  (map, item) => {
    map.put(item, map.getOrDefault(item, 0) + 1)
    map
  },
  // combop: Merge entries from the second map into the first
  (map1, map2) => {
    import scala.jdk.CollectionConverters._
    map2.asScala.foreach { case (key, count) =>
      map1.put(key, map1.getOrDefault(key, 0) + count)
    }
    map1
  }
)

// Print the final count map
import scala.jdk.CollectionConverters._
println(result.asScala.toMap)

Why This Works #

• Each partition gets its own isolated ConcurrentHashMap, so threads don’t compete for the same state during seqop (this is also more efficient than shared modifications).
• The combop step properly merges the independent partition maps into one final result, aligning with how aggregate is intended to work in parallel.

A Quick Alternative Note #

You mentioned you know other ways to count occurrences, but for completeness, Scala’s standard library has a simpler (though less efficient for large datasets) approach:

s.par.groupBy(identity).mapValues(_.size).toMap

But understanding the aggregate issue will help you avoid similar pitfalls with parallel collections down the line!

内容的提问来源于stack exchange，提问作者Ben Li