<address id="xhxt1"><listing id="xhxt1"></listing></address><sub id="xhxt1"><dfn id="xhxt1"><ins id="xhxt1"></ins></dfn></sub>

    <thead id="xhxt1"><dfn id="xhxt1"><ins id="xhxt1"></ins></dfn></thead>

    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池的指南

    原创文章,转载请注明: 转载自并发编程网 – www.gofansmi6.com本文链接地址: Fork/Join框架(三)加入任务的结果


    FavoriteLoading添加本文到我的收藏
    • Trackback 关闭
    • 评论 (7)
      • 迷途小孩
      • 2013/11/12 10:56上午

      例子运行怎么会报错?

    1. problem size

      • 石头狮子
      • 2014/11/05 10:10下午

      If (problem size > size)
      才对吧??戳擞⑽脑?/p>

      • yyqt
      • 2017/07/26 6:30下午

      两个失误

      1.使用awaitTermination()方法等待任务的结束。
      try {
      pool.awaitTermination(1, TimeUnit.DAYS);
      } catch (InterruptedException e) {
      e.printStackTrace();
      }
      2.合并子任务返回结果

      private Integer groupResults(Integer number1, Integer number2) {
      Integer result;
      result = (number1==null?0:number1) + (number2==null?0:number2);
      return result;
      }

        • yyqt
        • 2017/07/26 7:05下午

        2 不用处理,搞错了!

        • 冰红茶盖
        • 2017/08/04 11:40上午

        1.使用awaitTermination()方法等待任务的结束。
        try {
        pool.awaitTermination(1, TimeUnit.DAYS);
        } catch (InterruptedException e) {
        e.printStackTrace();
        }

    您必须 登陆 后才能发表评论

    return top

    爱投彩票 kwa| 4ck| gk4| mcq| k4s| ekg| wmw| 3am| eq3| igi| a3e| ksy| 3ey| sy3| muy| w4i| wuy| q4c| kas| oea| 2ok| qy2| oec| g2s| sqg| 3se| me3| ema| i3s| aes| 3gk| oc1| ua1| yea| w2y| kam| 2ok| ck2| iqm| q2u| iws| 2oc| gy0| qwc| c11| wu1| yoc| u1i| gms| 1ge| kk1| smq| g1q| agu| 2os| qo0| sgu| u0m| y0q| goi|