Fork/Join框架（三）加入任务的结果

声明：本文是《 Java 7 Concurrency Cookbook 》的第五章，作者： Javier Fernández González 译者：许巧辉校对：方腾飞

加入任务的结果

Fork/Join框架提供了执行返回一个结果的任务的能力。这些任务的类型是实现了RecursiveTask类。这个类继承了ForkJoinTask类和实现了执行者框架提供的Future接口。

在任务中，你必须使用Java API方法推荐的结构：

[code lang=”java”]

If (problem size < size){
tasks=Divide(task);
execute(tasks);
groupResults()
return result;
} else {
resolve problem;
return result;
}

[/code]

如果这个任务必须解决一个超过预定义大小的问题，你应该将这个任务分解成更多的子任务，并且用Fork/Join框架来执行这些子任务。当这些子任务完成执行，发起的任务将获得所有子任务产生的结果，对这些结果进行分组，并返回最终的结果。最终，当在池中执行的发起的任务完成它的执行，你将获取整个问题地最终结果。

在这个指南中，你将学习如何使用Fork/Join框架解决这种问题，开发一个在文档中查找单词的应用程序。你将会实现以下两种任务类型：

一个文档任务，将在文档中的行集合中查找一个单词。
一个行任务，将在文档的一部分数据中查找一个单词。

所有任务将返回单词在文档的一部分中或行中出现的次数。

如何做…

根据以下这些步骤来实现这个例子：

1.创建一个Document类，它将产生用来模拟文档的字符串的二维数组。

[code lang=”java”]

public class Document {

[/code]

2.创建一个带有一些单词的字符串数组。这个数组将被用来生成字符串二维数组。

[code lang=”java”]

private String words[]={"the","hello"，"goodbye","packt", "java","thread","pool","random","class","main"};

[/code]

3.实现generateDocument()方法。它接收以下参数：行数、每行的单词数。这个例子返回一个字符串二维数组，来表示将要查找的单词。

[code lang=”java”]

public String[][] generateDocument(int numLines, int numWords,String word){

[/code]

4.首先，创建生成这个文档必需的对象：字符串二维对象和生成随机数的Random对象。

[code lang=”java”]

int counter=0;
String document[][]=new String[numLines][numWords];
Random random=new Random();

[/code]

5.用字符串填充这个数组。存储在每个位置的字符串是单词数组的随机位置，统计这个程序将要在生成的数组中查找的单词出现的次数。你可以使用这个值来检查程序是否执行正确。

[code lang=”java”]

for (int i=0; i<numLines; i++){
for (int j=0; j<numWords; j++) {
int index=random.nextInt(words.length);
document[i][j]=words[index];
if (document[i][j].equals(word)){
counter++;
}
}
}

[/code]

6.将单词出现的次数写入控制台，并返回生成的二维数组。

[code lang=”java”]

System.out.println("DocumentMock: The word appears "+counter+" times in the document");

return document;
[/code]

7.创建一个DocumentTask类，指定它继承RecursiveTask类，并参数化为Integer类型。该类将实现统计单词在一组行中出现的次数的任务。

[code lang=”java”]

public class DocumentTask extends RecursiveTask<Integer> {

[/code]

8.声明一个私有的String类型的二维数组document，两个私有的int类型的属性名为start和end，一个私有的String类型的属性名为word。

[code lang=”java”]

private String document[][];
private int start, end;
private String word;

[/code]

9.实现这个类的构造器，用来初始化这些属性。

[code lang=”java”]

public DocumentTask (String document[][], int start, int end, String word){
this.document=document;
this.start=start;
this.end=end;
this.word=word;
}

[/code]

10.实现compute()方法。如果属性end和start的差小于10，那么这个任务统计单词位于行在调用processLines()方法的这些位置中出现的次数。

[code lang=”java”]

@Override
protected Integer compute() {
int result;
if (end-start<10){
result=processLines(document, start, end, word);

[/code]

11.否则，用两个对象分解行组，创建两个新的DocumentTask对象用来处理这两个组，并且在池中使用invokeAll()方法来执行它们。

[code lang=”java”]

} else {
int mid=(start+end)/2;
DocumentTask task1=new DocumentTask(document,start,mid,word);
DocumentTask task2=new DocumentTask(document,mid,end,word);
invokeAll(task1,task2);

[/code]

12.然后，使用groupResults()方法将这两个任务返回的结果相加。最后，返回这个任务统计的结果。

[code lang=”java”]

try {
result=groupResults(task1.get(),task2.get());
} catch (InterruptedException | ExecutionException e) {
e.printStackTrace();
}
}
return result;

[/code]

13.实现processLines()方法。它接收以下参数：字符串二维数组、start属性、end属性、任务将要查找的word属性。

[code lang=”java”]

private Integer processLines(String[][] document, int start, int
end,String word) {

[/code]

14.对于任务要处理的每行，创建LineTask对象来处理整行，并且将它们存储在任务数列中。

[code lang=”java”]

List<LineTask> tasks=new ArrayList<LineTask>();
for (int i=start; i<end; i++){
LineTask task=new LineTask(document[i], 0, document[i].
length, word);
tasks.add(task);
}

[/code]

15.在那个数列中使用invokeAll()执行所有任务。

[code lang=”java”]

invokeAll(tasks);

[/code]

16.合计所有这些任务返回的值，并返回这个结果。

[code lang=”java”]

int result=0;
for (int i=0; i<tasks.size(); i++) {
LineTask task=tasks.get(i);
try {
result=result+task.get();
} catch (InterruptedException | ExecutionException e) {
e.printStackTrace();
}
}
return new Integer(result);

[/code]

17.实现groupResults()方法。它相加两个数，并返回这个结果。

[code lang=”java”]

private Integer groupResults(Integer number1, Integer number2) {
Integer result;
result=number1+number2;
return result;
}

[/code]

18.创建LineTask类，指定它继承RecursiveTask类，并参数化为Integer类型。这个类将实现统计单词在一行中出现的次数的任务。

[code lang=”java”]

public class LineTask extends RecursiveTask<Integer>{

[/code]

19.声明这个类的序列号版本UID。这个元素是必需的，因为RecursiveTask类的父类，ForkJoinTask类实现了Serializable接口。声明一个私有的、String类型的属性line，两个私有的、int类型的属性start和end，一个私有的、String类型的属性word。

[code lang=”java”]

private static final long serialVersionUID = 1L;
private String line[];
private int start, end;
private String word;

[/code]

20.实现这个类的构造器，初始化这些属性。

[code lang=”java”]

public LineTask(String line[], int start, int end, String word)
{
this.line=line;
this.start=start;
this.end=end;
this.word=word;
}

[/code]

21.实现这个类的compute()方法。如果属性end和start之差小于100，这个任务在行中由start和end属性使用count()方法决定的片断中查找单词。

[code lang=”java”]

@Override
protected Integer compute() {
Integer result=null;
if (end-start<100) {
result=count(line, start, end, word);

[/code]

22.否则，将行中的单词组分成两部分，创建两个新的LineTask对象来处理这两个组，在池中使用invokeAll()方法执行它们。

[code lang=”java”]

} else {
int mid=(start+end)/2;
LineTask task1=new LineTask(line, start, mid, word);
LineTask task2=new LineTask(line, mid, end, word);
invokeAll(task1, task2);

[/code]

23.然后，使用groupResults()方法将这两个任务返回的值相加。最后，返回这个任务计算的结果。

[code lang=”java”]

try {
result=groupResults(task1.get(),task2.get());
} catch (InterruptedException | ExecutionException e) {
e.printStackTrace();
}
}
return result;

[/code]

24.实现count()方法。它接收以下参数：完整行的字符串数组、start属性、end属性、任务将要查找的word属性。

[code lang=”java”]

private Integer count(String[] line, int start, int end, String
word) {

[/code]

25.比较这个任务将要查找的word属性中的在start和end属性之间的位置的单词，如果它们相等，则增加count变量。

[code lang=”java”]

int counter;
counter=0;
for (int i=start; i<end; i++){
if (line[i].equals(word)){

counter++;
}
}

[/code]

26.为了显示示例的执行，令任务睡眠10毫秒。

[code lang=”java”]

try {
Thread.sleep(10);
} catch (InterruptedException e) {
e.printStackTrace();
}

[/code]

27.返回counter变量的值。

[code lang=”java”]

return counter;

[/code]

28.实现groupResults()方法。它合计两个数的值，并返回这个结果。

[code lang=”java”]

private Integer groupResults(Integer number1, Integer number2) {
Integer result;
result=number1+number2;
return result;
}

[/code]

29.实现示例的主类，通过创建Main类，并实现main()方法。

[code lang=”java”]

public class Main{
public static void main(String[] args) {

[/code]

30.使用DocumentMock类，创建一个带有100行，每行1000个单词的Document。

[code lang=”java”]

DocumentMock mock=new DocumentMock();
String[][] document=mock.generateDocument(100, 1000, "the");

[/code]

31.创建一个新的DocumentTask对象，用来更新整个文档的产品。参数start值为0，参数end值为100。

[code lang=”java”]

DocumentTask task=new DocumentTask(document, 0, 100, "the");

[/code]

32.使用无参构造器创建一个ForkJoinPool对象，在池中使用execute()方法执行这个任务。

[code lang=”java”]

ForkJoinPool pool=new ForkJoinPool();
pool.execute(task);

[/code]

33.实现一个代码块，用来显示关于池变化的信息。每秒向控制台写入池的某些参数的值，直到任务完成它的执行。

[code lang=”java”]

do {
System.out.printf("******************************************\n");
System.out.printf("Main: Parallelism: %d\n",pool.getParallelism());
System.out.printf("Main: Active Threads: %d\n",pool.getActiveThreadCount());
System.out.printf("Main: Task Count: %d\n",pool.getQueuedTaskCount());
System.out.printf("Main: Steal Count: %d\n",pool.getStealCount());
System.out.printf("******************************************\n");
try {
TimeUnit.SECONDS.sleep(1);
} catch (InterruptedException e) {
e.printStackTrace();
}
} while (!task.isDone());

[/code]

34.使用shutdown()方法关闭这个池。

[code lang=”java”]
pool.shutdown();
[/code]

35.使用awaitTermination()方法等待任务的结束。

[code lang=”java”]

try {
System.out.printf("Main: The word appears %d in the document",task.get());
} catch (InterruptedException | ExecutionException e) {
e.printStackTrace();
}

[/code]

36.打印单词在文档中出现的次数。检查这个数是否与DocumentMock类中写入的数一样。

[code lang=”java”]

try {
System.out.printf("Main: The word appears %d in the document",task.get());
} catch (InterruptedException | ExecutionException e) {
e.printStackTrace();
}

[/code]

它是如何工作的…

在这个示例中，你已经实现了两种不同的任务：

DocumentTask类：这个类的任务将处理由start和end属性决定的文档中的行组。如果这个行组的大小小于10，它为每行创建LineTask对象，并且当它们完成它们的执行时，它合计这些任务的结果，并返回这个合计值。如果这个任务要处理的行组大小不小于10，它将这个组分成两个并创建两个DocumentTask对象来处理这些新组。当这些任务完成它们的执行时，这个任务合计它们的结果，并返回这个合计值。
LineTask类：这个类的任务将处理文档中的一行的单词组。如果这个单词组小于10，这个任务直接在这个单词组中查找单词，并且返回这个单词出现的次数。否则，它将这个单词组分成两个并创建两个LineTask对象来处理。当这些任务完成它们的执行，这个任务合计这些任务的结果并返回这个合计值。

在Main类中，你已经使用默认构造器一个ForkJoinPool对象，并且你在它里面执行一个DocumentTask类，这个类将处理一个拥有100行，每行有1000个单词的文档。这个任务将使用其他的DocumentTask对象和LineTask对象来分解这个问题，当所有任务完成它们的执行，你可以使用启动任务来获取单词在整个文档中出现的总次数。由于任务返回一个结果，所以它们继承RecursiveTask类。

为了获取Task返回的结果，你已经使用了get()方法。这个方法是在Future接口中声明的，由RecursiveTask类实现的。

当你执行这个程序，你可以比较在控制台中的第一行和最后一行。第一行是文档生成时计算的单词出现的次数，最后一行是由Fork/Join任务计算的。

不止这些…

ForkJoinTask类提供其他的方法来完成一个任务的执行，并返回一个结果，这就是complete()方法。这个方法接收一个RecursiveTask类的参数化类型的对象，并且当join()方法被调用时，将这个对象作为任务的结果返回。它被推荐使用在：提供异步任务结果。

由于RecursiveTask类实现了Future接口，get()方法其他版本如下：

get(long timeout, TimeUnit unit)：这个版本的get()方法，如果任务的结果不可用，在指定的时间内等待它。如果超时并且结果不可用，那么这个方法返回null值。TimeUnit类是一个枚举类，它有以下常量：DAYS, HOURS，MICROSECONDS，MILLISECONDS， MINUTES， NANOSECONDS和SECONDS。

参见