Concurrente map/foreach en scala

Question

J'ai une itération vals: Iterable[T] et une fonction longue sans effets secondaires pertinents : f: (T => Unit). En ce moment, cela est appliqué à vals de manière évidente :

vals.foreach(f)

J'aimerais que les appels à f soient effectués de manière concurrente (dans des limites raisonnables). Y a-t-il une fonction évidente quelque part dans la bibliothèque de base de Scala ? Quelque chose comme :

Concurrent.foreach(8 /* Nombre de threads. */)(vals, f)

Alors que f est raisonnablement long à s'exécuter, il est assez court pour que je ne veuille pas supporter les frais généraux de l'invocation d'un thread pour chaque appel, donc je recherche quelque chose basé sur un pool de threads.

Answer 1

Beaucoup des réponses de 2009 utilisent encore l'ancienne scala.actors.Futures._, qui ne sont plus disponibles dans la nouvelle version de Scala. Alors qu'Akka est la façon préférée, une manière beaucoup plus lisible est simplement d'utiliser les collections parallèles (.par) :

vals.foreach { v => f(v) }

devient

vals.par.foreach { v => f(v) }

Alternativement, utiliser parMap peut sembler plus succinct bien que avec l'avertissement que vous devez vous rappeler d'importer le Scalaz* habituel. Comme d'habitude, il y a plus d'une façon de faire la même chose en Scala !

Answer 2

Scalaz a parMap. Vous l'utiliseriez comme suit:

import scalaz.Scalaz._
import scalaz.concurrent.Strategy.Naive

Cela équipera chaque foncteur (y compris Iterable) d'une méthode parMap, vous pouvez donc simplement faire:

vals.parMap(f)

Vous obtenez également parFlatMap, parZipWith, etc.

Answer 3

J'aime la réponse Futures. Cependant, bien qu'elle s'exécute de manière concurrente, elle renverra également de manière asynchrone, ce qui n'est probablement pas ce que vous voulez. L'approche correcte serait la suivante :

import scala.actors.Futures._

vals map { x => future { f(x) } } foreach { _() }

Answer 4

J'ai rencontré des problèmes en utilisant scala.actors.Futures dans Scala 2.8 (c'était buggy quand j'ai vérifié). Utiliser directement les libs java a fonctionné pour moi, cependant :

object Parallel {
  val cpus=java.lang.Runtime.getRuntime().availableProcessors
  import java.util.{Timer,TimerTask}
  def afterDelay(ms: Long)(op: =>Unit) = new Timer().schedule(new TimerTask {override def run = op},ms)
  def repeat(n: Int,f: Int=>Unit) = {
    import java.util.concurrent._
    val e=Executors.newCachedThreadPool //newFixedThreadPool(cpus+1)
    (0 until n).foreach(i=>e.execute(new Runnable {def run = f(i)}))
    e.shutdown
    e.awaitTermination(Math.MAX_LONG, TimeUnit.SECONDS)
  }
}

Answer 5

J'utiliserais scala.actors.Futures :

vals.foreach(t => scala.actors.Futures.future(f(t)))