企业官网建设流程全解析

• 示例工程

本篇技术指南围绕《Functional Programming in Scala》配套仓库 fpinscala 中 answerkey/parallelism/02.hint.md 这道习题提示展开：它引导读者思考"如果run由一个java.util.concurrent.ExecutorService支撑会怎样"，并鼓励翻阅整个java.util.concurrent包。我们将结合仓库内 Par.scala 与 Nonblocking.scala 的源码实现，讲清这道"开放式设计题"的来龙去脉：为什么答案是"继续读下去"，以及最终选定的Par[A] = ExecutorService => Future[A]表示从何而来。读完本文，你将理解纯函数并行 API 的设计动机、java.util.concurrent中与本主题直接相关的工具组件，以及阻塞式实现与非阻塞 CPS 实现的差异。

习题背景：第 7 章在构建什么

fpinscala 仓库按章节组织练习与答案，README 中明确列出了章节对应关系：parallelism对应书的第 7 章（"Purity, functional programming, and the push for parallelism" 一章）。这一章的目标是在不暴露线程、锁等底层细节的前提下，设计一个纯函数的并行计算库——一个Par（parallel）类型，以及围绕它的一组组合子。

从仓库的练习骨架 src/main/scala/fpinscala/exercises/parallelism/Par.scala 可以看到，Par被建模为一个object，其中汇聚了unit、run、map2、fork、lazyUnit、asyncF、sequence、parMap、parFilter、choice系列、flatMap、join等一系列函数。练习文件保留了大量带注释的提示，例如unit的注释说明"unit被表示为一个返回UnitFuture的函数"、map2的注释说明"map2不会把f的求值放到独立的逻辑线程中，fork才是 API 中唯一控制并行的函数"——这些注释正是理解本章设计哲学的钥匙。

answerkey/parallelism/目录下的 hint 序列与练习一一对应：

• 01.hint.md：关于map2（"两个Par输入不必是同一类型"）；
• 02.hint.md（本文主题）：关于run与ExecutorService；
• 03.hint.md：关于map2如何尊重超时（timeout）；
• 05.hint.md：关于sequence（"实现结构与之前为Option写的某个函数非常相似"）；
• 08.hint.md：关于固定大小线程池的问题（"如果线程池恰好被限制为大小 1 会怎样？"）；
• 10.hint.md：关于给Future增加错误处理回调。

02.hint.md位于这条设计链的最前端，它解决的不是"怎么写某一行代码"，而是"这个库的类型签名应该长什么样"。

读懂 hint：让run由 ExecutorService 支撑意味着什么

02.hint.md全文如下：

What ifrunwere backed by ajava.util.concurrent.ExecutorService? You may want to spend some time looking through thejava.util.concurrentpackage to see what other useful things you can find.

这句话有两层含义：

1. run的形态应该围绕ExecutorService来设计。run是"执行"一个并行计算、得到结果的入口函数。当时读者可能倾向于让run自己创建线程（例如直接new Thread(...)），而 hint 提示：更合理的做法是让run接受一个由调用方提供的ExecutorService，由它来调度计算。这样Par就与具体的执行资源解耦——Par本身只是一段"待执行的描述"，执行策略完全由外部注入。

2. java.util.concurrent包里藏着大量可直接复用的基础设施。ExecutorService只是入口，顺着它往下翻，还能找到Future、Callable、TimeUnit、CountDownLatch、AtomicReference、Executors工厂类等组件。本章后续的许多练习（如map2Timeouts）和仓库里的非阻塞实现，都直接建立在这些组件之上。

答案解读：为什么 7.2 的答案只有 "Keep reading"

与 hint 配套的答案文件 answerkey/parallelism/02.answer.md 内容极为简短：

Keep reading - we choose a representation forParlater in the chapter.

这不是敷衍，而是一个刻意的教学设计。这道习题的本质是让读者思考：Par到底是什么？它的run应该返回什么？map2的类型里两个Par参数怎么协同？但在章节的这个位置，最终表示尚未确定——你无法写出run的具体实现，因为你还没决定Par是一个函数、一个数据结构还是一种效果（effect）。答案要求读者"继续读下去"，直到章节后面引入正式表示后再回头看这道题。

这与01.hint.md（"map2的两个输入不必同类型"）形成呼应：map2A, B, C(f: (A, B) => C): Par[C]这种多态签名可以在不指定Par内部表示的前提下先写出来；而run的签名则直接依赖Par的表示——这正是它被设计成"开放式问题"的原因。

从仓库源码可以验证这条设计路线的终点：在 src/main/scala/fpinscala/exercises/parallelism/Par.scala（答案版见 src/main/scala/fpinscala/answers/parallelism/Par.scala），Par最终被表示为：

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

run则极简地退化为对这个函数的直接调用：

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

也就是说：Par[A]就是一个"给定ExecutorService就能得到Future[A]"的函数。hint 里"让run由ExecutorService支撑"的设想，最终以最彻底的方式落到了类型定义里。

java.util.concurrent 巡礼：hint 提到的"其他有用的东西"

hint 明确建议花时间翻阅java.util.concurrent包。就本主题而言，最相关的组件如下：

仓库里这些组件都有真实落点。例如fork的实现（Par.scala）：

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

它把参数a包装成Callable提交给ExecutorService。源码注释同时点出了这个"最自然"实现的隐患：外层Callable会阻塞等待内层任务完成，占据线程池中的一个线程，"用两个线程干一个线程的活"，损失了潜在并行度——这是08.hint.md（"固定大小线程池、大小为 1 会怎样"）埋下的伏笔，最终在章节后面引向非阻塞表示。

再看unit（Par.scala）：

def unitA: Par[A] = es => UnitFuture(a) private case class UnitFutureA extends Future[A]: def isDone = true def get(timeout: Long, units: TimeUnit) = get def isCancelled = false def cancel(evenIfRunning: Boolean): Boolean = false

UnitFuture是一个"永远完成、不可取消"的最简Future实现：它不碰ExecutorService，get直接返回常量。这一设计保证了unit是纯的、立即可得的，不需要任何线程资源。

延伸练习 7.3：map2Timeouts如何把 TimeUnit 用到位

顺着 hint 对java.util.concurrent的提示，下一道练习（对应03.hint.md："为了尊重超时，我们需要一个新的Future实现，记录求值第一个 future 花掉的时间，再从分配给第二个 future 的可用时间里减去它"）在仓库中已有完整实现——map2Timeouts（Par.scala）：

extension A def map2TimeoutsB, C(f: (A, B) => C): Par[C] = es => new Future[C]: private val futureA = pa(es) private val futureB = pb(es) @volatile private var cache: Option[C] = None def isDone = cache.isDefined def get() = get(Long.MaxValue, TimeUnit.NANOSECONDS) def get(timeout: Long, units: TimeUnit) = val timeoutNanos = TimeUnit.NANOSECONDS.convert(timeout, units) val started = System.nanoTime val a = futureA.get(timeoutNanos, TimeUnit.NANOSECONDS) val elapsed = System.nanoTime - started val b = futureB.get(timeoutNanos - elapsed, TimeUnit.NANOSECONDS) val c = f(a, b) cache = Some(c) c def isCancelled = futureA.isCancelled || futureB.isCancelled def cancel(evenIfRunning: Boolean) = futureA.cancel(evenIfRunning) || futureB.cancel(evenIfRunning)

这段代码完整展示了 hint 所期待读者"翻阅java.util.concurrent"的收获：

• get(timeout, units)用TimeUnit.NANOSECONDS.convert把任意时间单位统一换算为纳秒；
• 用System.nanoTime记录等待futureA实际花费的时间elapsed；
• 把剩余预算timeoutNanos - elapsed传给futureB，从而让整个map2尊重总超时；
• @volatile cache缓存结果，isDone由缓存是否已填充决定；
• 答案注释（见 03.answer.md）提醒：多次线程并发调用get时该实现不阻止重复求值，但对纯值而言重复求值不影响结果，此处不需要同步。

对比之下，普通版map2（Par.scala）"不尊重超时"——它只是把ExecutorService传给两个Par，分别get后应用f并包进UnitFuture。map2与map2Timeouts的差别，正是"是否把java.util.concurrent的超时机制吃透"的差别。

从阻塞到非阻塞：Nonblocking 里的 CPS 表示与 Actor

fork的阻塞缺陷（占用两个线程）是本章后半部分的核心矛盾，仓库用 src/main/scala/fpinscala/exercises/parallelism/Nonblocking.scala 给出了另一种表示：

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

这里Future[A]不再是 JDK 的java.util.concurrent.Future，而是一个**延续传递风格（CPS）**的函数：给定一个回调A => Unit，它负责在结果可用时把A交给回调，期间不阻塞任何线程。Par仍然是"给定ExecutorService得到Future"的形式——run的形态（由ExecutorService支撑）保持不变，只是内部表示换了。

非阻塞版run（Nonblocking.scala）用到了 hint 提示的另外两个组件AtomicReference与CountDownLatch：

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

流程是：注册回调（把结果写入AtomicReference并countDown）→await阻塞直到回调被异步触发 → 从ref读取结果。fork则通过eval助手把求值异步提交：

def eval(es: ExecutorService)(r: => Unit): Unit = es.submit(new Callable[Unit] { def call = r }) def forkA: Par[A] = es => cb => eval(es)(a(es)(cb))

非阻塞map2的汇合逻辑借助了仓库自带的 Actor.scala（实现取自 scalaz 库的Actor，采用基于AtomicReference的无锁 MPSC 队列）：两个Par的结果以Left/Right消息形式发给一个 actor，actor 等两边都到齐后再通过eval触发最终回调。源码注释坦承这个实现"在开线程上有点过于慷慨"，但足以保证栈安全。答案版（src/main/scala/fpinscala/answers/parallelism/Nonblocking.scala）的flatMap额外用fork包裹以获得栈安全，注释明确说明"链式调用大量flatMap时避免栈溢出"。

测试如何验证这套设计

仓库为这些并行原语提供了基于属性的测试套件 src/test/scala/fpinscala/exercises/parallelism/NonblockingSuite.scala。它用Executors.newFixedThreadPool(4)创建固定大小线程池，并用Gen生成随机输入来检验choice、choiceN、choiceViaChoiceN、choiceMap、chooser、choiceViaFlatMap、choiceNViaFlatMap、join、joinViaFlatMap、flatMapViaJoin的行为。例如join的检验：

test("Nonblocking.join")(genParParString): p => val pc = Par.joinString assertEquals(pc.run(es), p.run(es).run(es))

这类断言把"实现是否符合直觉语义"落到了可执行的测试上——join的结果应与"先跑外层、再跑内层"等价。这些测试同时验证了run与ExecutorService协作的正确性：run的返回值（无论是 JDKFuture的get，还是 CPS 版经AtomicReference/CountDownLatch取出的值）在不同调度下都应稳定。

动手验证：编译与运行

仓库采用 Scala CLI 构建（见 README.md）。在项目根目录执行：

$ scala-cli compile . # 编译全部 exercises 与 answers $ scala-cli console . # 进入 REPL，可 import fpinscala.exercises.parallelism.Par $ scala-cli test . -- 'fpinscala.exercises.parallelism.*' # 运行 parallelism 相关测试

要亲自观察Par与run的行为，可以在 REPL 里构造一个ExecutorService并运行：

import fpinscala.exercises.parallelism.Par.* import java.util.concurrent.* val es: ExecutorService = Executors.newFixedThreadPool(4) val p: Par[Int] = fork(unit(42)) p.run(es).get // 阻塞式表示：返回 42 es.shutdown()

注意：如果使用Nonblocking的表示，run直接返回A而非Future[A]，调用形态略有不同。

小结

answerkey/parallelism/02.hint.md这道 hint 的价值不在于给出答案，而在于指明设计方向：run应当由ExecutorService支撑，java.util.concurrent包中满是可复用的并行基础设施。仓库源码最终验证了这一方向——Par[A] = ExecutorService => Future[A]成为整章的基石，UnitFuture、fork的Callable包装、map2Timeouts的纳秒级预算管理，以及非阻塞 CPS 表示中的AtomicReference/CountDownLatch/Actor，全部是这条设计路线的自然延伸。而02.answer.md那句 "Keep reading" 也提醒我们：在类型表示尚未定案之前，有些习题最好的答案就是先放下笔，读完章节再回来。

• 示例工程