Gleichzeitiges Map/Foreach in Scala

Question

Ich habe eine Iteration vals: Iterable[T] und eine langlaufende Funktion ohne relevante Nebenwirkungen: f: (T => Unit) . Zurzeit gilt dies für vals auf die offensichtliche Weise:

vals.foreach(f)

Ich möchte, dass die Anrufe an f gleichzeitig durchgeführt werden (innerhalb angemessener Grenzen). Gibt es eine offensichtliche Funktion irgendwo in der Scala-Basisbibliothek? Etwas wie:

Concurrent.foreach(8 /* Number of threads. */)(vals, f)

Während f relativ lange läuft, ist es kurz genug, dass ich den Overhead des Aufrufs eines Threads für jeden Aufruf nicht möchte, also suche ich nach etwas, das auf einem Threadpool basiert.

Answer 1

Viele der Antworten aus dem Jahr 2009 verwenden noch die alte scala.actors.Futures._, die im neueren Scala nicht mehr enthalten sind. Während Akka der bevorzugte Weg ist, ist ein viel lesbarerer Weg, einfach parallel zu verwenden ( .par ) Sammlungen:

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

wird

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

Alternativ kann die Verwendung von parMap prägnanter erscheinen, allerdings mit dem Vorbehalt, dass Sie daran denken müssen, das übliche Scalaz* zu importieren. Wie üblich gibt es in Scala mehr als einen Weg, die gleiche Sache zu tun!

Answer 2

Scalaz hat parMap . Sie würden es wie folgt verwenden:

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

Damit wird jeder Funktor (einschließlich Iterable ) mit einer parMap Methode, so können Sie einfach tun:

vals.parMap(f)

Sie erhalten außerdem parFlatMap , parZipWith , usw.

Answer 3

Ich mag die Futures Antwort. Sie wird zwar gleichzeitig ausgeführt, kehrt aber auch asynchron zurück, was wahrscheinlich nicht das ist, was Sie wollen. Der richtige Ansatz wäre wie folgt:

import scala.actors.Futures._

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

Answer 4

Ich hatte einige Probleme mit scala.actors.Futures in Scala 2.8 (es war fehlerhaft, wenn ich überprüft). Mit Java-Bibliotheken direkt arbeitete für mich, obwohl:

final 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

Ich würde die scala.actors.Futures :

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