• 示例工程

在《Functional Programming in Scala》(fpinscala) 第 7 章并行编程中，asyncF是连接普通函数与并行计算世界的关键组合子。本文以 answerkey/parallelism/04.answer.md 的练习答案为核心，结合仓库中Par的完整实现源码，深入讲解asyncF的构造原理、它在parMap等并行原语中的作用，以及它与惰性求值、fork调度的关系，帮助读者真正掌握函数式并行 API 的设计思路。

1. 练习背景：从Par到asyncF

在第 7 章中，Par是一个表示"可能并行计算"的抽象类型。仓库中的定义位于 answers/parallelism/Par.scala：

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

这是一个不透明类型别名（opaque type）：Par[A]实际上是一个函数，它接收一个ExecutorService，返回一个Future[A]。这一表示意味着"并行计算"被建模为一个待调度执行的描述，而不是直接执行的动作——只有在传入具体的线程池（ExecutorService）时才真正运行。

在练习 4 之前，书中已经构建了三个基础原语：

• unit：把纯值提升为Par，不消耗线程（用UnitFuture包装常量）；
• fork：显式把一个Par计算调度到线程池中异步执行，是 API 中唯一控制并行度的函数；
• lazyUnit：fork(unit(a))，把按名参数a变成异步计算的快捷方式。

这三个原语的定义（见 answers/parallelism/Par.scala）：

def unitA: Par[A] = es => UnitFuture(a) def forkA: Par[A] = es => es.submit(new Callable[A] { def call = a(es).get }) def lazyUnitA: Par[A] = fork(unit(a))

练习 4 的问题正是：给定一个普通函数A => B，如何把它提升为一个"接收参数、返回Par[B]"的函数？也就是说，要构造一个把同步函数变成异步调用的适配器。

2. 标准答案：asyncF的实现

answerkey/parallelism/04.answer.md 给出的答案只有三行，但信息密度很高：

def asyncFA,B: A => Par[B] = a => lazyUnit(f(a))

我们来逐层拆解这段代码：

1. asyncFA,B：接收一个普通同步函数f；
2. 返回A => Par[B]：返回一个函数，它接收参数a: A，返回一个Par[B]并行计算；
3. a => lazyUnit(f(a))：对任意输入a，先同步地计算f(a)——但注意，lazyUnit的参数是按名传递的（a: => A），因此f(a)并不会在构造Par时立即求值，而是延迟到Par真正被运行时才计算，且计算发生在fork调度的线程中。

2.1 为什么用lazyUnit而不是unit

对比两种写法：

// 写法 A：unit —— 立即求值 a => unit(f(a)) // 写法 B：lazyUnit —— 惰性求值 + 异步调度 a => lazyUnit(f(a))

如果使用unit(f(a))，f(a)会在调用asyncF(f)返回的函数时立即执行（因为unit的参数是严格求值的a: A），这违背了并行 API 的设计意图——我们希望计算被"推迟"到调度阶段。而lazyUnit通过fork(unit(a))实现了两层效果：

• 惰性：f(a)不会在函数返回时求值，只有Par被传入ExecutorService执行时才触发；
• 异步：fork会把求值提交给线程池，f(a)在池中某个线程上执行，从而真正获得并行性。

这正是asyncF名字的由来——它把一个同步的A => B转变成"异步化"的A => Par[B]。

3.asyncF的典型应用：parMap

asyncF最重要的价值在于它是parMap的基石。parMap并行地对集合中每个元素应用函数，其实现（见 answers/parallelism/Par.scala）：

def parMapA, B(f: A => B): Par[List[B]] = fork: val fbs: List[Par[B]] = ps.map(asyncF(f)) sequence(fbs)

执行流程可以分解为：

1. fork:把整个映射过程放到一个逻辑线程中执行；
2. ps.map(asyncF(f))：对输入列表的每个元素，用asyncF(f)构造一个Par[B]——每个元素的计算都被封装成独立的、可异步调度的并行任务；
3. sequence(fbs)：把这些Par[B]列表"合并"成一个Par[List[B]]，收集所有结果。

sequence的实现（见 answers/parallelism/Par.scala）有多个版本，从朴素版本sequenceSimple（foldRight+map2）到更高效的sequenceBalanced（利用IndexedSeq.splitAt二分并行合并）。parMap最终采用sequenceBalanced版本以获得更好的并行度。

同理，parFilter（answers/parallelism/Par.scala）也用asyncF把谓词判断封装成并行任务：

def parFilterA(f: A => Boolean): Par[List[A]] = fork: val pars: List[Par[List[A]]] = l.map(asyncF(a => if f(a) then List(a) else List())) sequence(pars).map(_.flatten)

可以看到：asyncF是从"普通函数"到"并行任务集合"这一转换中最核心的胶水，没有它，parMap/parFilter将无法为每个元素独立地构造并行计算。

4. 深入源码：asyncF在非阻塞实现中的形态

仓库中还有一套基于**续体传递风格（CPS）**的非阻塞实现，位于 answers/parallelism/Nonblocking.scala。在那里Future被重新定义为回调函数：

opaque type Future[+A] = (A => Unit) => Unit opaque type Par[+A] = ExecutorService => Future[A]

对应的asyncF定义（answers/parallelism/Nonblocking.scala）在结构上与阻塞版完全一致：

def asyncFA, B: A => Par[B] = a => lazyUnit(f(a))

这印证了asyncF的实现不依赖具体的调度细节——无论底层Par是阻塞式Future还是 CPS 回调，asyncF的构造方式都是"按名参数 +lazyUnit"。差异只在于各自的lazyUnit/fork内部如何调度（阻塞版用ExecutorService.submit阻塞等待，非阻塞版用eval(es)(...)异步执行回调）。这种"高层组合子稳定、底层表示可变"的设计，正是函数式库可组合性的体现。

parMap在非阻塞版本中同样复用了asyncF（answers/parallelism/Nonblocking.scala），并额外提供了IndexedSeq重载版本直接使用sequenceBalanced，进一步验证了asyncF作为通用基础设施的地位。

5. 调用链与可验证性：从asyncF到run

asyncF构造出的Par最终通过run执行。在阻塞版中（answers/parallelism/Par.scala）：

extension A def run(s: ExecutorService): Future[A] = pa(s)

一次典型的asyncF调用链为：

asyncF(f)(x) // 返回 Par[B]（尚未执行） → lazyUnit(f(x)) // = fork(unit(f(x)))，按名捕获 f(x) → run(es) // 传入线程池 → es.submit(...) // fork 调度：f(x) 在线程池线程上求值 → Future.get // 阻塞等待结果

测试用例对这类组合子的行为有明确验证。例如 NonblockingSuite.scala 用Executors.newFixedThreadPool(4)构造固定大小线程池，对choice、choiceN、join、flatMap等组合子逐一断言"实际运行结果 == 期望结果"。虽然asyncF本身没有独立的测试，但它是parMap、parFilter、choice*系列组合子的公共底座，这些测试间接覆盖了它的正确性。

6. 边界与陷阱：惰性、死锁与线程池

理解asyncF的惰性语义后，还需注意与之相关的经典陷阱——线程饥饿导致的死锁。书中答案 answerkey/parallelism/09.answer.md 明确指出：

对大小为 2 的线程池，fork(fork(fork(x)))会死锁；更典型的例子是fork(map2(fork(x), fork(y)))——外层任务先提交并占用一个线程等待fork(x)与fork(y)，而两个内层任务只能共享剩余的一个线程，导致死锁。

asyncF(f)内部包含一次fork，因此在嵌套使用（如parMap配合fork的外层包裹，见parMap定义中的fork:）时，同样受线程池大小约束。这解释了为什么：

• parMap在fork:块内工作——外层fork让sequence的合并过程本身也异步化；
• 当输入集合很大或线程池很小时，需要警惕任务嵌套造成的阻塞；
• 书中在第 13 章用Task数据类型给出更完整的解决方案（见 answerkey/parallelism/10.answer.md 与 answers/iomonad/Task.scala），通过显式的异步调度避免阻塞占线。

7. 小结

asyncF虽只有一行核心代码，却是 fpinscala 第 7 章并行 API 从"原语"走向"应用"的关键一步：

理解asyncF，就等于理解了函数式并行编程的核心思想：把"计算什么"（Par的描述）与"何时、在哪个线程上计算"（ExecutorService的调度）彻底分离。这也是整章练习的最终目标——用少数几个原语组合出无限丰富的并行程序。

• 示例工程