5

Em Java ConcurrentHashMap possui uma propriedade interessante. Operações como putIfAbsent, remove, replace, computeIfAbsent, computeIfPresent, compute e merge são atômicas.

Exemplo:

final Map<String, Integer> map = new ConcurrentHashMap<>();
IntStream.rangeClosed(1, 10000)
         .parallel()
         .forEach(i -> map.merge("key" + (i % 5 + 1), i, Integer::sum));
final long total = map.values().stream().mapToInt(Integer::intValue).sum(); // 50005000

Eu não encontrei nenhuma operação similar na API do TrieMap:

val map = TrieMap[String, Int]().withDefaultValue(0)
(1 to 10000).par.foreach(i => map(s"key${i % 5 + 1}") += i)
val total = map.values.sum // valor aleatório, += não é uma operação atomica

Claramente é possível sincronizar operações manualmente bem como usar construções como CAS, STM, etc:

(1 to 10000).par.foreach(i => {
  val key = s"key${i % 5 + 1}".intern
  key.synchronized {
    map(key) += i
  }
})

Alternativamente, é possível utilizar ConcurrentHashMap do Java diretamente (sem usar asScala):

val map = new java.util.concurrent.ConcurrentHashMap[String, Int]()
(1 to 10000).par.foreach(i => map.merge(s"key${i % 5 + 1}", i, Integer.sum _))
val total = map.values().asScala.sum  

Há alguma maneira mais simples / idiomática de garantir updates atomicos em um concurrent.Map?

0

1 Resposta 1

3

Scala 2.12.x

Para quem ainda está na versão 2.12.x, segue uma solução com Compare-and-swap que funciona corretamente:

val map = TrieMap[String, Int]()
(1 to 10000).par.foreach { i =>
  val key = s"key${i % 5 + 1}"
  map.putIfAbsent(key, 0)

  Iterator.continually(map(key)).find { oldValue =>
    map.replace(key, oldValue, oldValue + i)
  }
}
val total = map.values.sum

No exemplo acima o TrieMap pode ser substituído por um concurrent.Map sem prejuízos para a solução:

import scala.collection.JavaConverters._

val map = new ConcurrentHashMap[String, Int]().asScala

Scala 2.13.x

Dito isso, desde que fiz essa pergunta três anos se passaram e muita coisa mudou. A versão mais atual de Scala no momento que estou escrevendo essa resposta é a 2.13.4. De 2018 para cá:

  • Coleções paralelas foram movidas para o seu próprio módulo: scala-parallel-collections.

    Esse modulo agora precisa ser incluído como dependência do projeto:

    libraryDependencies += "org.scala-lang.modules" %% "scala-parallel-collections" % "1.0.0"
    

    E o uso do método par requer que você importe conversores implícitos:

    import scala.collection.parallel.CollectionConverters._
    
    val parallelRange = (1 to 10000).par
    
  • Algumas operações que não eram atômicas passaram a ser. Por exemplo o método: TrieMap.getOrElseUpdate (ver Issue 7943).

  • Alternativas para as operações do Java 8 foram introduzidas. E.g., a contrapartida do java.util.Map#compute no Java é scala.collection.mutable.Map#updateWith.

Em Scala 2.13.x o código acima pode ser reescrito de maneira mais idiomática:

val map = TrieMap[String, Int]()
(1 to 10000).par.foreach { i =>
  map.updateWith(s"key${i % 5 + 1}") {
    case Some(c) => Some(c + i)
    case None    => Some(i)
  }
}
val total = map.values.sum

No exemplo acima também podemos substituir o TrieMap por um concurrent.Map, porém os conversores foram migrados para outro pacote:

import scala.jdk.CollectionConverters._

val map = new ConcurrentHashMap[String, Int]().asScala

Você deve fazer log-in para responder a esta pergunta.

Esta não é a resposta que você está procurando? Pesquise outras perguntas com a tag .