选择数据类型和函数
“创建并行计算”具体是指什么？我们可以从一个相对简单的例子入手——求一组整数的和。例如下面就是利用左折叠的方法计算求和：
def sum(ints: Seq[Int]): Int =
ints.foldLeft(0)(_ + _)
除了叠加算法， 还有一个分治的算法，代码如下：
def sum1(ints: IndexedSeq[Int]): Int =
if (ints.length < 1)
ints.headOption.getOrElse(0)
else {
val (l, r) = ints.splitAt(ints.length)
sum1(l) + sum1(r)
}
我们使用splitAt函数将序列一分为二，并各自递归求和，最后合并它们的结果。这种实现可以实现并行化，即对两部分的求和可以同时进行。
一种用于并行计算的数据类型
用于表示并行计算的任何数据类型都包含一个结果，这个结果是一个有意义的类型（这里是Int），且能够获取。为此我们设计一个这样的数据类型Par[A]（Par是Paralle的简写），它就像一个装有结果的容器，并具备下面的方法：
trait Par[A] {
}
object Par {
//接受一个未求值的A，返回结果将会在另一个线程中执行
def unit[A](a: => A): Par[A] = ???
//从并行计算中抽取结果
def get[A](pa: Par[A]): A = ???
}
现在我们用自定义的数据类型更新求和算法：
import Par._
def sum2(ints: IndexedSeq[Int]): Int =
if (ints.length <= 1)
ints.headOption.getOrElse(0)
else {
val (l, r) = ints.splitAt(ints.length)
val pl = unit(sum2(l))
val pr = unit(sum2(r))
get(pl) + get(pr)
}
现在我们面临一个选择， 是让unit在一个独立的逻辑线程中立即求值，还是等到get被调用的时候再求值。unit立即求值会导致程序无法并行计算，但是unit返回一个代表一步计算的Par[Int]，那在调用get的时候无法避免产生副作用。如何才能避免unit和get的缺陷呢？
组合并行计算
我们可以不调用get函数， 那么函数的代码将如下：
def sum3(ints: IndexedSeq[Int]): Par[Int] =
if (ints.length <= 1)
unit(ints.headOption.getOrElse(0))
else {
val (l, r) = ints.splitAt(ints.length)
map2(sum3(l), sum3(r))(_ + _)
}
练习 7.1
Par.map2是一个新的高阶函数，用于组合两个并行计算的结果。实现map2函数：
def map2[A, B, C](pa: Par[A], pb: Par[B])(f: (A, B) => C): Par[C] = {
val a = get(pa)
val b = get(pb)
unit(f(a, b))
}
显性分流
目前的API没有明确的表明何时应该将计算从主线程中分流出去，换句话说程序员也不知道在哪儿会发生并行计算。如何让分流更加明确呢？我们引入另一个函数来做：
//将par[A]分配另一个独立的线程中去运行
def folk[A](pa: => Par[A]): Par[A] = ???
让我们来重写sum函数：
def sum4(ints: IndexedSeq[Int]): Par[Int] =
if (ints.length <= 1)
unit(ints.headOption.getOrElse(0))
else {
val (l, r) = ints.splitAt(ints.length)
map2(folk(sum3(l)), folk(sum3(r)))(_ + _)
}
对于length <= 1情况我们并不需要folk到一个独立线程中计算。
现在回到unit是严格还是惰性的问题，有了folk，即便unit是严格也不会有什么损失。至于非严格版本我们叫它lazyUnit吧：
//接受一个已求值的A，返回结果将会在另一个线程中执行
def unit[A](a: A): Par[A] = ???
//接受一个未求值的A，返回结果将会在另一个线程中执行
def lazyUnit[A](a: => A): Par[A] = folk(unit(a))
到此，我们看出Par只是一个值的函数，表明并行计算。而实际执行并行计算的是get函数，所以我们将get函数改名为run函数，表明这个并行计算实际执行的地方。
//从并行计算中抽取结果
def run[A](pa: Par[A]): A = ???
确定表现形式
经过各种思考和选择之后，我们有了下面大致的API。
//接受一个已求值的A，返回结果将会在另一个线程中执行
def unit[A](a: A): Par[A] = ???
//接受一个未求值的A，返回结果将会在另一个线程中执行
def lazyUnit[A](a: => A): Par[A] = folk(unit(a))
//从并行计算中抽取结果
def run[A](pa: Par[A]): A = ???
//将par[A]分配另一个独立的线程中去运行
def folk[A](pa: => Par[A]): Par[A] = ???
def map2[A, B, C](pa: Par[A], pb: Par[B])(f: (A, B) => C): Par[C] = {
val a = run(pa)
val b = run(pb)
unit(f(a, b))
}
练习 7.2
在继续之前，我们尽可能实现API中的函数
如上代码
让我们根据run函数反推Par类型，让我们试着假设run可以访问一个ExecutorService，看能不能搞清Par的样子：
def run[A](s: ExecutorService)(pa: Par[A]): A = ???
最简单的莫过于，Par[A]是ExecutorService => A，当然这也未免太简单了，为此Par[A]，应该是ExecutorService => Future[A]，而run直接返回Future：
type Par[A] = ExecutorService => Future[A]
def run[A](s: ExecutorService)(pa: Par[A]): Future[A] = pa(s)
完善API
既然有了Par的表现形式，不妨就简单直接点，基于Par的表现类型最简单的实现：
object Par {
import java.util.concurrent.{ExecutorService, Future, Callable}
type Par[A] = ExecutorService => Future[A]
private case class UnitFuture[A](a: A) extends Future[A] {
override def isCancelled: Boolean = false
override def get(): A = a
override def get(timeout: Long, unit: TimeUnit): A = a
override def cancel(mayInterruptIfRunning: Boolean): Boolean = false
override def isDone: Boolean = true
}
//接受一个已求值的A，返回结果将会在另一个线程中执行
def unit[A](a: A): Par[A] = es => UnitFuture(a)
//接受一个未求值的A，返回结果将会在另一个线程中执行
def lazyUnit[A](a: => A): Par[A] = folk(unit(a))
//从并行计算中抽取结果
def run[A](s: ExecutorService)(pa: Par[A]): Future[A] = pa(s)
//将par[A]分配另一个独立的线程中去运行
def folk[A](pa: => Par[A]): Par[A] = es => {
es.submit(new Callable[A] {
override def call(): A = pa(es).get()
})
}
def map2[A, B, C](pa: Par[A], pb: Par[B])(f: (A, B) => C): Par[C] =
es => {
val af = pa(es)
val bf = pb(es)
UnitFuture(f(af.get(), bf.get()))
}
}
练习7.3
改进Map2的实现，支持超时设置。
def map2[A, B, C](pa: Par[A], pb: Par[B],
timeout: Long, timeUnit: TimeUnit)(f: (A, B) => C): Par[C] =
es => {
val af = pa(es)
val bf = pb(es)
val a = af.get(timeout, timeUnit)
val b = bf.get(timeout, timeUnit)
UnitFuture(f(a, b))
}
练习 7.4
使用lazyUnit写一个函数将另一个函数A => B转换为一个一步计算
def asyncF[A, B](f: A => B): A => Par[B] =
a => lazyUnit(f(a))
练习 7.5
实现一个叫做sequence的函数。不能使用而外的基础函数，不能调用run。
def sequence[A](li: List[Par[A]]): Par[List[A]] = {
def loop(n: Int, res: Par[List[A]]): Par[List[A]] = n match {
case m if m < 0 => res
case _ => loop(n - 1, map2(li(n), res)(_ :: _))
}
loop(li.length - 1, unit(Nil))
}
练习 7.6
实现parFilter，并过滤列表元素
def parFilter[A](li: List[A])(f: A => Boolean): Par[List[A]] = {
def loop(n: Int, res: List[Par[A]]): List[Par[A]] = n match {
case m if m < 0 => res
case _ =>
if (f(li(n))) loop(n - 1, lazyUnit(li(n)) :: res)
else loop(n - 1, res)
}
sequence(loop(li.length - 1, Nil))
}
