Chinaunix首页 | 论坛 | 博客
  • 博客访问: 26277385
  • 博文数量: 2065
  • 博客积分: 10377
  • 博客等级: 上将
  • 技术积分: 21525
  • 用 户 组: 普通用户
  • 注册时间: 2008-11-04 17:50
文章分类

全部博文(2065)

文章存档

2012年(2)

2011年(19)

2010年(1160)

2009年(969)

2008年(153)

分类: Java

2010-06-17 21:10:49

Java™ 7 中的 java.util.concurrent 包的新增功能之一是一个用于 fork-join 风格的并行分解的库。在本系列的 第 1 部分 中,Brian Goetz 演示了 fork-join 如何为众多的算法提供一种自然的分解机制,以有效地利用硬件的并行性。在本文中,他将介绍 ParallelArray 类,这个类简化了内存中数据结构上的并行排序和搜索操作。

上一期 Java 理论与实践 中,我们研究了 fork-join 库,这个库将添加到 Java 7 的 java.util.concurrent 包中。fork-join 技术提供了一种表示 divide-and-conquer 并行算法的简单方式,这种方式能够在大量硬件上有效执行,无需修改代码。

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

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



                
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 任务。例如,要将对大数据集求平均值的操作并行化,可以递归地将大数据集分解成更小的数据集 — 就像在合并排序中做的那样 — 对子集求均值,然后在合并步骤中求出各子集的平均值的加权平均值。

对于排序和搜索问题,fork-join 库提供了一种表示可以并行化的数据集操作的非常简单的途径:ParallelArray 类。其思路是:用 ParallelArray 表示一组结构上类似的数据项,用 ParallelArray 上的方法创建一个对分解数据的具体方法的描述。然后用该描述并行地执行数组操作(幕后使用的是 fork-join 框架)。这种方法支持声明性地指定数据选择、转换和后处理操作,允许框架计算出合理的并行执行计划,就像数据库系统允许用 SQL 指定数据操作并隐藏操作的实现机制一样。ParallelArray 的一些实现可用于不同的数据类型和大小,包括对象数组和各种原语组成的数组。

清单 2 显示的示例使用 ParallelArray 对学生成绩进行汇总,演示了选择、处理和汇总的基本操作。 Student 类包含学生的信息(姓名、毕业年份、GPA)。helper 对象 isSenior 用来选择今年毕业的学生,healper 对象 getGpa 提取指定学生的 GPA 字段。清单开始部分的表达式创建一个 ParallelArray,代表一组学生,然后从今年毕业的学生中选出最高的 GPA。



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

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

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

static final Ops.ObjectToDouble selectGpa = new Ops.ObjectToDouble() {
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,其元素为对并行数组 ab 执行 a[i]+b[i] 操作得到的值。

  • 汇总:将所有值组合为一个值,例如计算总和、平均值、最小值或最大值。清单 2 中的示例使用了 max() 方法。预定义的汇总方法,例如 min()sum()max(),是用更通用的 reduce() 构建的。

清单 2 中,我们能够计算出任意个学生中的最高 GPA,但是我们想要的结果稍微有点不同 — 哪个学生的 GPA 最高。这个任务可以通过两次计算完成(一次计算最高 GPA,另一次选择拥有这个 GPA 的学生),但是 ParallelArray 提供了更加简便的方法来实现常用的汇总统计,例如最大值、最小值、总和、平均值,以及最大和最小元素的索引。 summary() 方法通过一个并行操作计算这些汇总统计信息。

清单 3 演示了计算汇总信息的 summary() 方法,包括求出最小和最大元素的索引,从而避免反复传递数据:



                
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 个学生,一个线程)的性能对比:





线程


1248
学生10001.000.300.351.20
100002.112.311.021.62
1000009.995.283.635.53
100000039.3424.6720.9435.11
10000000340.25180.28160.21190.41


虽 然结果相当混乱(受到多个因素的影响,包括 GC 活动),但是能够看出,不仅通过等于内核数量的线程池大小能够实现最好的结果(如果任务是纯计算型的,这个结果可想而知),而且 4 个内核的性能是 1 个内核的性能的 2-3 倍速,这表明不用进行调节,使用高级、便捷的机制也有可能得到不错的并行性。

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

清单 4 显示了使用 BGGA 闭包方案重写的求最高 GPA 的查询。(在使用函数类型扩展的 ParallelArray 版本中,withFilter() 的参数类型不是 Ops.Predicate,而是函数类型 { T => boolean })。闭包标注取消了与内部类相关的引用,允许更紧凑(重点更突出、更直接)地表示需要的数据操作。现在代码缩减为三行,几乎所有代码都表示了我们试图实现的结果的某个重要方面。



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

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

阅读(1238) | 评论(0) | 转发(0) |
给主人留下些什么吧!~~