纯函数式的并行计算（2）

在上章节中我们介绍了如何使用java.util.current中的API来实现Par：这种纯函数式的并行计算结构，但是最后我们也提出了这个实现存在一个bug，那就是在固定线程池的情况下，这种实现式非常容易造成死锁的。那么问题的关键式什么呢？问题的关键就在于Par的表现类型Future在获取其内部值的时候是会阻塞线程的，而我们需要Par的类型是无阻塞的。那么如何才能实现一个无阻塞的Par呢？想法很简单，需要我们自己去实现一个Future，他可以注册一个callback，在准备好之后被回调。代码如下：

package org.fp.scala.par

import java.util.concurrent.ExecutorService

sealed trait Future[A] {
  
  private[par] def apply(f: A => Unit): Unit
}

object Future {

  type Par[+A] = ExecutorService => Future[A]

}

既然重新定义了Par，那么我们也需要重新定义run方法，代码如下：

  def run[A](es: ExecutorService)(pa: Par[A]): A = {
    val ref = new AtomicReference[A]()
    val latch = new CountDownLatch(1)
    pa(es){ a =>
      ref.set(a)
      latch.countDown()
    }
    latch.await()
    ref.get()
  }

实现了run方法我们再来看下如何实现unit方法，代码如下：

  def unit[A](a: A): Par[A] =
    es => new Future[A] {
      override private[par] def apply(f: A => Unit): Unit = f(a)
    }

注意这里并没有用到ExecutorService。现在再来看下fork的实现，他是我们实际引入并行计算的地方，代码如下：

  def fork[A](pa: => Par[A]): Par[A] =
    es => new Future[A] {
      override private[par] def apply(f: A => Unit): Unit =
        eval(es)(pa(es)(f))
    }

  def eval[A](es: ExecutorService)(f: => Unit): Unit =
    es.submit(new Callable[Unit] {
      override def call(): Unit = f
    });

当fork返回Future接受cb时，他会分流出去对a进行求值。
心在想想map2又应该怎么实现呢？先看一下map2的方法签名：

def map2[A, B, C](pa: Par[A], pb: Par[B])(f: (A, B) => C): Par[C] = ???

这里要实现无阻塞是非常棘手的，从概念上讲，我们希望map2并行运算两个Par参数。但是又几个竞态条件需要注意，毕竟java.util.concurrent里面的低级操作实现正确的非阻塞是非常困难的。
要想实现map2，我们需要使用Actor的无阻塞并发实现。

  def map2[A, B, C](pa: Par[A], pb: Par[B])(f: (A, B) => C): Par[C] =
    es => new Future[C] {
      override private[par] def apply(f: C => Unit): Unit = {
        var ar: Option[A] = None
        val br: Option[B] = None

        val combiner = Actor[Either[A, B]](es) {
          case Left(a) => br match {
            case None => ar = Some(a)
            case Some(b) => eval(es)(f(a, b))
          }
          case Right(b) => ar match {
            case None => br = Some(b)
            case Some(a) => eval(es)(f(a, b))
          }
        }

        pa(es)(a => combiner ! Left(a))
        pb(es)(b => combiner ! Right(b))
      }
    }

基于这样的实现，我们再也不必担心用完线程，可以任务run各种复杂的Par了。