Java 理论与实践: 应用 fork-join 框架，第 2 部分

用 Java 7 中的 ParallelArray 类加速排序和搜索

未显示需要 JavaScript 的文档选项



打印本页



将此页作为电子邮件发送



英文原文 级别： 中级 Brian Goetz (brian.goetz@sun.com), 高级工程师, Sun Microsystems 2008 年 4 月 15 日 在 上一期 Java 理论与实践 中，我们研究了 fork-join 库，这个库将添加到 Java 7 的

java.util.concurrent

包中。fork-join 技术提供了一种表示 divide-and-conquer 并行算法的简单方式，这种方式能够在大量硬件上有效执行，无需修改代码。 随着处理器数量的增加，为了有效利用可用的硬件，我们需要识别并利用程序中更细粒度的并行性。最近几年中，选择粗粒度的任务边界（例如在 Web 应用程序中处理单一请求）和在线程池中执行任务，通常能够提供足够的并行性，实现可接受的硬件利用效率。但是如果要再进一步，就必须深入挖掘更多的并行性，以让硬件全速运转。一个成熟的并行领域就是大数据集中的排序和搜索。用 fork-join 可以很容易地表示这类问题，正如您在 上一期 中看到的那样。但是由于这些问题非常普遍，所以该类库提供了一种更简单的方法 —

ParallelArray

。 回顾：fork-join 分解 fork-join 将 divide-and-conquer 技术具体化；它接受一个问题并将其递归分解为子问题，直到子问题小到用顺序方法解决更有效。递归步骤包括将一个问题分解成两个或更多子问题，将子问题排队求解（分叉（fork）步骤），等待子问题的结果（合并（join） 步骤），然后合并结果。这类算法的一个示例是使用 fork-join 库进行合并排序，如清单 1 所示： 清单 1.使用 fork-join 库进行合并排序

public class MergeSort extends RecursiveAction {
final int[] numbers;
final int startPos, endPos;
final int[] result;

private void merge(MergeSort left, MergeSort right) {
int i=0, leftPos=0, rightPos=0, leftSize = left.size(), rightSize = right.size();
while (leftPos < leftSize && rightPos < rightSize)
result[i++] = (left.result[leftPos] <= right.result[rightPos])
? left.result[leftPos++]
: right.result[rightPos++];
while (leftPos < leftSize)
result[i++] = left.result[leftPos++];
while (rightPos < rightSize)
result[i++] = right.result[rightPos++];
}

public int size() {
return endPos-startPos;
}

protected void compute() {
if (size() < SEQUENTIAL_THRESHOLD) {
System.arraycopy(numbers, startPos, result, 0, size());
Arrays.sort(result, 0, size());
}
else {
int midpoint = size() / 2;
MergeSort left = new MergeSort(numbers, startPos, startPos+midpoint);
MergeSort right = new MergeSort(numbers, startPos+midpoint, endPos);
coInvoke(left, right);
merge(left, right);
}
}
}

合并排序本身并非并行算法，因为它可以顺序执行。当数据集太大，内存无法容纳，必须分片保存的时候，经常使用合并排序。合并排序的最差性能和平均性能为 O(n log n)。但是由于很难在原地进行合并，所以合并排序的内存需求比能够原地进行的排序算法（例如快速排序）更高。但是因为子问题的排序能够并行完成，所以合并排序的并行性比快速排序好。
在处理器数量固定的情况下，并行化并不能将 O(n log n) 问题转变为 O(n) 问题，但是问题越适合并行化，它就越接近 O(n) 问题， 从而减少总体运行时）。减少总体运行时间意味着用户能更快得到结果 — 即使并行执行需要比顺序执行更多的 CPU 周期。
使用 fork-join 技术的主要好处是，它提供了一种编写并行执行的算法的简便方法。程序员无需知道部署中可用的 CPU 数量；运行时能够很好地协调可用 CPU 的工作量、在大范围硬件上产生合理的结果。
更细粒度的并行性
在主流服务器应用程序中，最适合更细粒度并行性的地方是数据集的排序、搜索、选择和汇总。其中的每个问题都可以用 divide-and-conquer 轻松地并行化，并能轻松地表示为 fork-join 任务。例如，要将对大数据集求平均值的操作并行化，可以递归地将大数据集分解成更小的数据集 — 就像在合并排序中做的那样 — 对子集求均值，然后在合并步骤中求出各子集的平均值的加权平均值。

ParallelArray

对于排序和搜索问题，fork-join 库提供了一种表示可以并行化的数据集操作的非常简单的途径：

ParallelArray

类。其思路是：用

ParallelArray

表示一组结构上类似的数据项，用

ParallelArray

上的方法创建一个对分解数据的具体方法的描述。然后用该描述并行地执行数组操作（幕后使用的是 fork-join 框架）。这种方法支持声明性地指定数据选择、转换和后处理操作，允许框架计算出合理的并行执行计划，就像数据库系统允许用 SQL 指定数据操作并隐藏操作的实现机制一样。

ParallelArray

的一些实现可用于不同的数据类型和大小，包括对象数组和各种原语组成的数组。
清单 2 显示的示例使用

ParallelArray

对学生成绩进行汇总，演示了选择、处理和汇总的基本操作。

Student

类包含学生的信息（姓名、毕业年份、GPA）。helper 对象

isSenior

用来选择今年毕业的学生，healper 对象

getGpa

提取指定学生的 GPA 字段。清单开始部分的表达式创建一个

ParallelArray

，代表一组学生，然后从今年毕业的学生中选出最高的 GPA。
清单 2. 使用

ParallelArray

选择、处理和汇总数据

ParallelArray<Student> students = new ParallelArray<Student>(fjPool, data);
double bestGpa = students.withFilter(isSenior)
.withMapping(selectGpa)
.max();

public class Student {
String name;
int graduationYear;
double gpa;
}

static final Ops.Predicate<Student> isSenior = new Ops.Predicate<Student>() {
public boolean op(Student s) {
return s.graduationYear == Student.THIS_YEAR;
}
};

static final Ops.ObjectToDouble<Student> selectGpa = new Ops.ObjectToDouble<Student>() {
public double op(Student student) {
return student.gpa;
}
};

表示并行数组上的操作的代码有点欺骗性。

withFilter()

和

withMapping()

方法实际上并不搜索或转换数据；它们只是设置 “查询” 的参数。实际的工作在最后一步进行，在本例中就是对

max()

的调用。

ParallelArray

支持以下基本操作：

筛选：选择计算过程中包含的元素子集。在清单 2 中，筛选器由

withFilter()

方法指定。
应用：将一个过程应用到每个选中的元素。清单 2 没有展示这个技术，但是

apply()

方法允许对每个选中元素执行一个操作。
映射：将选中的元素转换为另一种形式（例如从元素中提取数据字段）。在本例中，这个转换由

withMapping()

方法执行，我们将

Student

转换为学生的 GPA。其结果是指定选择和映射结果的

ParallelArray

。
替换：将每个元素替换为由它派生的另一个元素，创建新的并行数组。此技术与映射类似，但是形成新的

ParallelArray

，可以在其上执行进一步查询。替换的一种情况是排序，将元素替换为不同的元素，从而对其进行排序（内置的

sort()

方法可用于此操作）。另一种特殊情况是

cumulate()

方法，该方法根据指定的组合操作用累积值替换每个元素。替换操作也可用于组合多个

ParallelArray

，例如创建一个

ParallelArray

，其元素为对并行数组

a

和

b

执行

a+b[i]

操作得到的值。
汇总：将所有值组合为一个值，例如计算总和、平均值、最小值或最大值。清单 2 中的示例使用了

max()

方法。预定义的汇总方法，例如

min()

、

sum()

和

max()

，是用更通用的

reduce()

构建的。

常见汇总操作
在 清单 2 中，我们能够计算出任意个学生中的最高 GPA，但是我们想要的结果稍微有点不同 — 哪个学生的 GPA 最高。这个任务可以通过两次计算完成（一次计算最高 GPA，另一次选择拥有这个 GPA 的学生），但是

ParallelArray

提供了更加简便的方法来实现常用的汇总统计，例如最大值、最小值、总和、平均值，以及最大和最小元素的索引。

summary()

方法通过一个并行操作计算这些汇总统计信息。
清单 3 演示了计算汇总信息的

summary()

方法，包括求出最小和最大元素的索引，从而避免反复传递数据：
清单 3. 找到 GPA 最高的学生

SummaryStatistics summary = students.withFilter(isSenior)
.withMapping(selectGpa)
.summary();
System.out.println("Worst student: " + students.get(summary.minIndex()).name;
System.out.println("Average GPA: " + summary.getAverage();

局限性

ParallelArray

并不是一种通用的内存中数据库，也不是一种指定数据转换和提取的通用机制（例如 .NET 3.0 中的特性 — 语言集成查询 LinQ）；它只是用于简化特定范围的数据选择和转换操作的表达方式，以将这些操作轻松、自动地并行化。所以，它存在一些局限性；例如，必须在映射操作之前指定筛选操作。（允许多个筛选操作，但是将它们组合成一个复合筛选操作通常会更有效）。它的主要目的是使开发人员不用思考如何将工作并行化；如果能够用

ParallelArray

提供的操作表示转换，那么就能轻松实现并行化。
性能
为了评估

ParallelArray

的效率，我编写了一个简单的程序，针对各种大小的数组和 fork-join 池运行查询。在运行 Windows 的 Core 2 Quad 系统中进行测试。表 1 显示的是相对于基准情形（1000 个学生，一个线程）的性能对比：
表 1. 最高 GPA 查询的性能测量
线程
1
2
4
8
学生
1000
1.00
0.30
0.35
1.20
10000
2.11
2.31
1.02
1.62
100000
9.99
5.28
3.63
5.53
1000000
39.34
24.67
20.94
35.11
10000000
340.25
180.28
160.21
190.41
虽然结果相当混乱（受到多个因素的影响，包括 GC 活动），但是能够看出，不仅通过等于内核数量的线程池大小能够实现最好的结果（如果任务是纯计算型的，这个结果可想而知），而且 4 个内核的性能是 1 个内核的性能的 2-3 倍速，这表明不用进行调节，使用高级、便捷的机制也有可能得到不错的并行性。
连接闭包

ParallelArray

提供了一种不错的方法，可用于声明性地指定数据集上的筛选、处理和聚合操作，还方便自动并行化。但是，尽管它的语法比使用原始的 for-join 库更容易表达，但还是有些麻烦；每个筛选器、映射器、reducer 通常被指定为内部类，所以即使像 “查找今年毕业的学生中最高的 GPA” 这样简单的查询，仍然需要十几行代码。Java 7 可能会在 Java 语言中加入闭包；支持闭包的一种说法是：闭包使得小段代码 — 例如

ParallelArray

中的筛选器、映射器、reducer —的表示更加紧凑。
清单 4 显示了使用 BGGA 闭包方案重写的求最高 GPA 的查询。（在使用函数类型扩展的

ParallelArray

版本中，

withFilter()

的参数类型不是

Ops.Predicate<T>

，而是函数类型

{ T => boolean }

）。闭包标注取消了与内部类相关的引用，允许更紧凑（重点更突出、更直接）地表示需要的数据操作。现在代码缩减为三行，几乎所有代码都表示了我们试图实现的结果的某个重要方面。
清单 4. 用闭包重写求最高 GPA 示例

double bestGpa = students.withFilter({Student s => (s.graduationYear == THIS_YEAR) })
.withMapping({ Student s => s.gpa })
.max();

结束语
随着可用的处理器数量增加，我们需要发现程序中更细粒度的并行性来源。最有吸引力候选方案之一是聚合数据操作 — 排序、搜索和汇总。JDK 7 中将引入的 fork-join 库提供了一种 “轻松表示” 某类可并行化算法的途径，从而让程序能够在一些硬件平台上有效运行。通过声明性地描述想要执行的操作，然后让

ParallelArray

确定具体的执行方法，fork-join 库的

ParallelArray

组件使并行聚合操作的表示变得更加简单。
参考资料
学习

您可以参阅本文在 developerWorks 全球站点上的 英文原文 。
“Java 理论和实践：应用 fork-join 框架，第 1 部分”（developerWorks，2007 年 11 月）：研究 fork-join 如何为众多算法提供自然的分解机制，以有效地利用硬件的并行性。
[i]Concurrent Programming in Java 的 4.4 节（Doug Lea，Prentice Hall PTR，1999 年 11 月）：详细介绍了并行分解。
Doug Lea 维护的关于并发的 Web 站点：下载 fork-join 框架，它是 jsr166y 包一部分，或者阅读与其设计相关的 论文。
浏览 技术书店，查找与这些和其他技术主题相关的图书。
developerWorks Java 技术专区：其中有数百篇关于 Java 编程各方面的文章。

讨论

查阅 developerWorks blog 并加入 developerWorks 社区。

关于作者





Brian Goetz 作为一名专业软件开发人员已经 20 年了。他是 Sun Microsystems 的高级工程师，并且效力于多个 JCP 专家组。Brian 的著作 Java Concurrency In Practice 在 2006 年 5 月由 Addison-Wesley 出版。请参阅 Brian 在流行的业界出版物上 已发表和即将发表的文章。