Fork/Join框架(三)加入任务的结果

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

加入任务的结果

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

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


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

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

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

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

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

如何做…

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

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


public class Document {

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


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

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


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

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


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

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


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++;
}
}
}

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


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

return document;

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


public class DocumentTask extends RecursiveTask<Integer> {

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


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

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


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

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


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

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


} 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);

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


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

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


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

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


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

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


invokeAll(tasks);

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


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

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


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

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


public class LineTask extends RecursiveTask&lt;Integer&gt;{

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


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

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


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

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


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

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


} 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);

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


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

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


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

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


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

counter++;
}
}

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


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

27.返回counter变量的值。


return counter;

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


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

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


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

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


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

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


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

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


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

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


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());

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

pool.shutdown();

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


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

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


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

它是如何工作的…

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

  • 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。

参见

  • 在第5章,Fork/Join框架中的创建一个Fork/Join池的指南
  • 在第8章,测试并发应用程序中的监控Fork/Join池的指南

原创文章,作者:ItWorker,如若转载,请注明出处:https://blog.ytso.com/140903.html

(0)
上一篇 2021年9月5日
下一篇 2021年9月5日

相关推荐

发表回复

登录后才能评论