Параллельный map/foreach в scala
Значок
Рекурсивная функция
procedure factorial(n)
return (0<n) * factorial(n-1) | 1
end
я обманул немного отрицательных сторон разрешения для возврата 1. Если Вы хотите, чтобы он перестал работать, учитывая отрицательный аргумент, это немного менее кратко:
return (0<n) * factorial(n-1) | (n=0 & 1)
Тогда
write(factorial(3))
write(factorial(-1))
write(factorial(20))
печать
6
2432902008176640000
Повторяющийся генератор
procedure factorials()
local f,n
f := 1; n := 0
repeat suspend f *:= (n +:= 1)
end
Тогда
every write(factorials() \ 5)
печать
1
2
6
24
120
Для понимания этого: оценка направлена на цель и отслеживает в обратном порядке при отказе. Нет никакого булева типа и бинарных операторов, которые возвратили бы булевскую переменную на других языках, или перестать работать или возвратить их второй аргумент - за исключением |, который в контексте единственного значения возвращает его первый аргумент, если он успешно выполняется, иначе пробует его второй аргумент. (в контексте нескольких-значений это возвращается, его первый аргумент тогда его второй аргумент)
suspend похож yield на других языках, за исключением того, что генератор явно не называют многократно для возврата его результатов. Вместо этого every спрашивает его аргумент в пользу всех значений, но ничего не возвращает по умолчанию; это полезно с побочными эффектами (в этом вводе-выводе случая).
\ пределы количество значений, возвращенных генератором, который в случае [1 111] было бы бесконечно.
5 ответов
Мне нравится ответ Futures . Однако, хотя он будет выполняться одновременно, он также будет возвращать асинхронно, что, вероятно, не то, что вам нужно. Правильный подход будет следующим:
import scala.actors.Futures._
vals map { x => future { f(x) } } foreach { _() }
У меня возникли проблемы с использованием scala.actors.Futures в Scala 2.8 (когда я проверял, в нем были ошибки). Однако использование java libs напрямую сработало для меня:
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)
}
}
Последний выпуск Функциональной Java имеет некоторые функции параллелизма более высокого порядка, которые вы можете использовать.
import fjs.F._
import fj.control.parallel.Strategy._
import fj.control.parallel.ParModule._
import java.util.concurrent.Executors._
val pool = newCachedThreadPool
val par = parModule(executorStrategy[Unit](pool))
И затем ...
par.parMap(vals, f)
Не забудьте выключите пул .
Скалаз имеет parMap . Вы можете использовать его следующим образом:
import scalaz.Scalaz._
import scalaz.concurrent.Strategy.Naive
Это оснастит каждый функтор (включая Iterable ) методом parMap , так что вы можете просто сделать:
vals.parMap(f)
Вы также получите ] parFlatMap , parZipWith и т. д.
Я бы использовал scala.actors.Futures :
vals.foreach(t => scala.actors.Futures.future(f(t)))