深入了解Java语言中的并发性选项有何不同

前言

Java™ 工程师在努力让并发性容易为开发人员所用。尽管做了不少的改进，但并发性仍然是 Java 平台的一个复杂、容易出错的部分。一部分复杂之处在于理解语言本身中的并发性的低级抽象，这些抽象在您的代码中填满了同步的代码块。另一个复杂之处来自一些新库，比如 fork/join，这些库在某些场景中非常有用，但在其他场景中收效甚微。了解容易混乱的大量低级选项需要专业经验和时间。

脱离 Java 语言的优势之一是，能够改善和简化并发性等区域。每种 Java 下一代语言都为此问题提供了独特的答案，利用了该语言的默认编程风格。在本期文章中，我首先将会介绍函数式编程风格的优势：轻松并行化。我会深入分析 Scala 和 Groovy 的细节（下一期文章将全面介绍 Clojure）。然后介绍 Scala actor。

完美数

数学家尼科马库斯（诞生于公元前 6 世纪）将自然数分为惟一的完美数（perfect number）、过剩数（abundant number） 或亏数（deficient number）。一个完美数等于它的正因数（不包括它本身）之和。例如，6 是一个完美数，因为它的因数是 1、2、3 和 6，28 也是完美数 (28 = 1 + 2 + 4 + 7 + 14)。过剩数的因素之和大于该数，亏数的因数之和小于该数。

这里使用完美数分类法是为了方便介绍。除非要处理大量数字，是否查找因素对于从并行化中获益而言是一个微不足道的问题。使用更多线程可带来一些益处，但线程之间的切换开销对细粒度的作业而言代价很高。

让现有代码并行化

在 “函数式编码风格” 那一期的文章中，我们鼓励您使用更高级的抽象，比如化简、映射和过滤器，而不是迭代。此方法的优势之一是容易并行化。

我的 函数式思维 系列的读者熟悉包含完美数 的数字分类模式（参见 完美数 边栏）。我在该系列中展示的任何解决方案都没有利用并发性。但是因为这些解决方案使用了转换函数，比如 map，所以我可以在每种 Java.net 语言中做极少的工作来创建并行化的版本。

清单 1 是完美数分类器的一个 Scala 示例。

清单 1. Scala 中的并行完美数分类器

object NumberClassifier {
def isFactor(factor: Int, number: Int) =
number % factor == 0
def factors(number: Int) = {
val factorsBelowSqrt = (1 to Math.sqrt(number).toInt).par.filter (isFactor(_, number))
val factorsAboveSqrt = factorsBelowSqrt.par.map(number / _)
(factorsBelowSqrt ++ factorsAboveSqrt).toList.distinct
}
def sum(factors: Seq[Int]) =
factors.par.foldLeft(0)(_ + _)
def isPerfect(number: Int) =
sum(factors(number)) - number == number
}

清单 1 中的 factors() 方法返回一个数的因数列表，使用 isFactor() 方法过滤所有可能的值。factors() 方法使用了我在 “函数式思维：转换和优化” 中更详细地介绍的一种优化。简单来讲，过滤每个数来查找因素的效率很低，因为根据定义，一个因数是其乘积等于目标数的两个数之一。

相反，我仅过滤不超过目标数的平方根的数，然后通过将目标数除以每个小于平方根的因数来生成对称因数列表。在 清单 1 中，factorsBelowSqrt 变量包含过滤操作的结果。factorsAboveSqrt 的值是现有列表的映射，用于生成这些对称值。最后，factors() 的返回值是一个串联的列表，它从一个并行的List 转换为常规的 List。

请注意，清单 1 中添加了 par 修饰符。该修饰符会导致 filter、map 和 foldLeft 并行运行，从而能够使用多个线程来处理请求。par 方法（在整个 Scala 集合库中都是一致的）将该序列转换为并行序列。因为两种类型的序列反映了它们的签名，所以 par 函数变成了并行化某个操作的临时方式。

在 Scala 中并行化常见问题的简单性，在语言设计和函数模式上都经过证实。函数式编程鼓励使用通用的函数，比如 map、filter 和 reduce，运行时以不可见的方式可以进一步优化它们。Scala 语言设计人员考虑到了这些优化，最终产生了集合 API 的设计。

边缘情况

在 清单 1 的 factors() 方法实现中，整数的平方根（例如，16 的平方根：4）显示在两个列表中。因此，factors() 方法返回的最后一行是对 distinct 函数的调用，它从列表中删除了重复值。您也可以在每一处都使用 Set，而不是只在列表中使用它，但 List 常常拥有 Set 中所没有的有用函数。

Groovy 也允许轻松地修改现有的函数代码，通过 GPars 库让它并行化，该库捆绑在各个 Groovy 发行版中。GPars 框架在内置的 Java 并行性原语之上创建有用的抽象，常常将它们包装在语法糖中。GPars 提供了令人眼花缭乱的并行机制，其中一种机制可用于分配线程池，然后将操作分布到这些池中。清单 2 中给出了一个使用 Groovy 编写的，使用 GPars 线程池的完美数分类器。

清单 2. Groovy 中的并行完美数分类器

class NumberClassifierPar {
static def factors(number) {
GParsPool.withPool {
def factors = (1..round(sqrt(number) + 1)).findAllParallel { number % it == 0 }
(factors + factors.collectParallel { number / it }).unique()
}
}
static def sumFactors(number) {
factors(number).inject(0, { i, j -> i + j })
}
static def isPerfect(number) {
sumFactors(number) - number == number
}
}

清单 2 中的 factors() 方法使用了与 清单 1 相同的算法：它生成不超过目标数的平方根的所有因数，然后生成剩余的因数并返回串联的集合。与 清单 1 中一样，我使用 unique() 方法来确保整数的平方根不会生成重复值。

无需像 Scala 中一样放大集合来创建对称并行版本，Groovy 的设计人员创建了该语言的转换方法的 xxxParallel() 版本（例如 findAllParallel() 和 collectParallel()）。但除非这些方法包装在 GPars 线程池代码块中，否则它们不会起作用。

在 清单 2 中，我创建了一个线程池，调用 GParsPool.withPool 创建一个代码块，支持在该代码块中使用 xxxParallel() 方法。withPool 方法存在其他变体。例如，您可指定池中的线程数量。

Clojure 通过 化简器 库提供了一种类似的临时并行化机制。使用转换函数的化简器版本来实现自动并行化，例如，
使用 r/map 代替 map。（r/ 是化简器命名空间。）化简器的实现是 Clojure 的语法灵活性中的一个引人注目的案例分析，它通过极小的更改实现了强大的添加功能。

Scala 中的 actor

Scala 包含众多并发性和并行性机制。一种较流行的机制是 actor 模型，它提供了将工作分布到线程上的优势，而没有同步的复杂性。在概念上，actor 有能力完成工作，然后将一个非阻塞的结果发送给协调器。要创建一个 actor，需要创建 Actor 类的子类并实现 act() 方法。通过使用 Scala 的语法糖，可绕过许多定义仪式，在代码块内定义 actor。

我没有为 清单 1 中的数字分类器执行的一种优化是，使用线程对作业的因数查找部分进行分区。如果我的计算机上有 4 个处理器，我可为每个处理器创建一个线程并拆分工作。例如，如果我尝试找到数字 16 的因数之和，那么我可以安排处理器 1 来查找从 1 到 4 的因数（并求和），安排处理器 2 来处理 5 到 8，依此类推。使用 actor 是一种自然的选择：我为每个范围创建了一个 actor，独立地执行每个 actor（通过语法糖隐式执行或通过调用它的 act() 方法来显式执行），然后收集结果，如清单 3 所示。

清单 3. 使用 Scala 中的 actor 识别完美数

object NumberClassifier extends App {
def isPerfect(candidate: Int) = {
val RANGE = 10000
val numberOFPartitions = (candidate.toDouble / RANGE).ceil.toInt
val coordinator = self
for (i <- 0 until numberOFPartitions) {
val lower = i * RANGE + 1
val upper = candidate.min((i + 1) * RANGE)
actor {
var partialSum = 0
for (j <- lower to upper)
if (candidate % j == 0) partialSum += j
coordinator ! partialSum
}
}
var responsesExpected = numberOFPartitions
var sum = 0
while (responsesExpected > 0) {
receive {
case partialSum : Int =>
responsesExpected -= 1
sum += partialSum
}
}
sum == 2 * candidate
}
}

为了保持此示例的简单性，我将 isPerfect() 编写为单个完整的函数。我首先基于常量 RANGE 创建了一些分区。其次，我需要一种方式来收集 actor 所生成的消息。在 coordinator 变量中，我有一个引用可供 actor 向其发送消息，其中 self 是 Actor 的一个成员，表示 Scala 中获取线程标识符的可靠方式。

我然后为分区编号创建一个循环，使用 RANGE 偏移来生成范围的下限和上限。接下来，为该范围创建一个 actor，使用 Scala 的语法糖来避免正式的类定义。在 actor 内，我为 partialSum 创建了一个临时保存器，然后分析该范围，将找到的因数收集到 partialSum 中。收集部分和（此范围内的所有因数的和）后， (coordinator ! partialSum) 向协调器发回一条消息，使用感叹号运算符。（这种消息传递语法的灵感来源于 Erlang 语言，用作一种对另一个线程执行非阻塞调用的途径。）

接下来，我启动了一个循环，等待所有 actor 完成处理。在等待过程中，我进入了一个 receive 代码块。在该代码块内，我想要一条 Int 消息，我在本地将它分配给 partialSum，然后递减想要的响应数量，将该部分添加到总和中。所有 actor 完成且报告结果后，该方法的最后一行将该和与候选数的 2 倍相比较。如果比较结果为 true，那么我的候选数就是一个完美数，该函数的返回值为 true。

actor 的一个不错的优势是所有权分区。每个 actor 都有一个 partialSum 局部变量，但它们从不彼此联系。通过协调器收到消息时，底层执行机制是不可见的：您创建了一个 receive 块，其他实现细节是不可见的。

Scala 中的 actor 机制是 Java 下一代语言封装 JVM 的现有工具并使用一致的抽象来扩展它们的优秀示例。用 Java 语言编写类似的代码，并使用低级并发性原语，这些操作都需要非常复杂地协调多个线程。Scala 中的 actor 隐藏了所有复杂性，留下的是容易理解的抽象。

结束语

Java 下一代语言都为 Java 语言中的并发性难题提供了答案，而且每种语言以不同方式解决了这些问题。在本期文章中，我演示了所有三种 Java 下一代语言如何实现临时并行化。我还演示了 Scala 中的 actor 模型，构建了一个数字分类器来并行计算因数之和。