forkjoin-分叉合并框架

朱雀 2023-02-19 12:24 8阅读 0赞

# 前言 #

随着多处理器时代的到来，越来越多的计算机支持多处理器并行计算。语言层面也提供了支持，fork/join就是java语言利用多处理器计算的实现框架。更多信息参考[官方文档][Link 1]。

# Fork/Join #

fork/join框架是`ExecutorService`接口的实现，它可以帮助您利用多处理器的优势。它是为那些可以递归地分解成更小部分的工作而设计的。目标是使用所有可用的处理能力来提高应用程序的性能。

与任何ExecutorService实现一样，fork/join框架将任务分配给线程池中的工作线程。fork/join框架与众不同，因为它使用了**窃取工作**的算法。工作线程耗尽了要做的事情，可以从其他仍然忙碌的线程那里窃取任务。

fork/join框架的中心是ForkJoinPool类，它是AbstractExecutorService类的扩展。ForkJoinPool实现了核心的工作窃取算法，可以执行ForkJoinTask进程。

## 基本使用 ##

使用fork/join框架的第一步是编写执行部分工作的代码。你的代码应该类似于下面的伪代码:

if (my portion of the work is small enough)
      do the work directly
    else
      split my work into two pieces
      invoke the two pieces and wait for the results

将这段代码包装在`ForkJoinTask`子类中，通常使用它更特殊的类型之一，`RecursiveTask`(可以返回结果)或`RecursiveAction`。

在`ForkJoinTask`子类准备好之后，创建表示所有要完成的工作的对象，并将其传递给`ForkJoinPool`实例的`invoke()`方法。

## 模糊清晰的图像 ##

为了帮助您理解fork/join框架是如何工作的，请考虑下面的示例。假设你想模糊一幅图像。原始源图像由一个整数数组表示，其中每个整数包含单个像素的颜色值。模糊后的目标图像也用与源图像大小相同的整数数组表示。

执行模糊是通过工作通过源阵列一次一个像素。每个像素与其周围的像素(红色、绿色和蓝色组件均为平均值)进行平均，然后将结果放置在目标数组中。由于图像是一个大数组，这个过程可能会花费很长时间。通过使用fork/join框架实现该算法，您可以利用多处理器系统上的并发处理。这里有一个可能的实现:

public class ForkBlur extends RecursiveAction { 
        private int[] mSource;
        private int mStart;
        private int mLength;
        private int[] mDestination;
      
        // Processing window size; should be odd.
        private int mBlurWidth = 15;
      
        public ForkBlur(int[] src, int start, int length, int[] dst) { 
            mSource = src;
            mStart = start;
            mLength = length;
            mDestination = dst;
        }
    
        protected void computeDirectly() { 
            int sidePixels = (mBlurWidth - 1) / 2;
            for (int index = mStart; index < mStart + mLength; index++) { 
                // Calculate average.
                float rt = 0, gt = 0, bt = 0;
                for (int mi = -sidePixels; mi <= sidePixels; mi++) { 
                    int mindex = Math.min(Math.max(mi + index, 0),
                                        mSource.length - 1);
                    int pixel = mSource[mindex];
                    rt += (float)((pixel & 0x00ff0000) >> 16)
                          / mBlurWidth;
                    gt += (float)((pixel & 0x0000ff00) >>  8)
                          / mBlurWidth;
                    bt += (float)((pixel & 0x000000ff) >>  0)
                          / mBlurWidth;
                }
              
                // Reassemble destination pixel.
                int dpixel = (0xff000000     ) |
                       (((int)rt) << 16) |
                       (((int)gt) <<  8) |
                       (((int)bt) <<  0);
                mDestination[index] = dpixel;
            }
        }
      
      ...

现在实现了抽象的`compute()`方法，它可以直接执行模糊，也可以将其分割成两个更小的任务。一个简单的数组长度阈值可以帮助确定是执行工作还是分割工作。

protected static int sThreshold = 100000;
    
    protected void compute() { 
        if (mLength < sThreshold) { 
            computeDirectly();
            return;
        }
        
        int split = mLength / 2;
        
        invokeAll(new ForkBlur(mSource, mStart, split, mDestination),
                  new ForkBlur(mSource, mStart + split, mLength - split,
                               mDestination));
    }

如果前面的方法在`RecursiveAction`类的子类中，那么设置在`ForkJoinPool`中运行的任务是很简单的，包括以下步骤:

1.  创建一个表示要完成的所有工作的任务。

// source image pixels are in src
    // destination image pixels are in dst
    ForkBlur fb = new ForkBlur(src, 0, src.length, dst);

1.  创建将运行该任务的ForkJoinPool。

ForkJoinPool pool = new ForkJoinPool();

1.  运行任务。

pool.invoke(fb);

有关完整的源代码，包括一些创建目标图像文件的额外代码，请参见[ForkBlur][]示例。

## 标准实现 ##

除了使用fork/join框架为要在多处理器系统上并发执行的任务实现自定义算法(如上一节中的ForkBlur.java示例)之外，Java SE中还有一些通常有用的特性已经使用fork/join框架实现了。Java SE 8中引入的一个这样的实现是由`java.util.Arrays`类使用的用于其`parallelSort()`方法。这些方法与`sort()`类似，但是通过fork/join框架利用并发性。在多处理器系统上运行时，大数组的并行排序比顺序排序快。但是，这些方法如何使用fork/join框架超出了Java教程的范围。有关这些信息，请参阅Java API文档。

fork/join框架的另一个实现由`java.util.streams`包中的方法使用，它是计划在Java SE 8发布中的Lambda项目的一部分。有关更多信息，请参见Lambda表达式部分。

# `ForkJoinTask<V>`的方法 #

fork/join框架的难点在于任务的定义。

下面先详细看看`ForkJoinTask<V>`的方法：

三个虚方法，用于子类实现。

public abstract V getRawResult();
    protected abstract void setRawResult(V value);
    protected abstract boolean exec();

分叉：异步执行任务

public final ForkJoinTask<V> fork()

合并：获取异步执行任务的返回值

public final V join()

执行：直接执行任务并获取返回值

public final V invoke()

## RecursiveAction ##

无返回值的递归，比如下面【打印】的例子：

public static class Print extends RecursiveAction { 
        private final List<Integer> list;
    
        public Print(List<Integer> list) { 
          this.list = list;
        }
    
        @Override
        protected void compute() { 
          //如果任务足够小，直接返回，不需要分叉和合并
          if (list.size() == 1) { 
            try { 
              Thread.sleep(5000);//模拟一段时间
            } catch (InterruptedException e) { 
              e.printStackTrace();
            }
            System.err.println(Thread.currentThread().getName() + "|" + list.get(0));
            return;
          }
          //否则一分为二，
          int middle = list.size() / 2;
          List<Integer> head = Take.take(list, 0, middle);
          List<Integer> tail = Take.take(list, middle, list.size());
          Print one = new Print(head);
          Print two = new Print(tail);
          two.fork();//分叉
          one.invoke();//执行当前任务
          two.join();//合并
        }
      }

## RecursiveTask ##

有返回值的递归，比如下面【[斐波那契数][Link 2]】的例子。

public static class Fibonacci extends RecursiveTask<BigDecimal> { 
        private final int n;
    
        Fibonacci(int n) { 
          this.n = n;
        }
    
        @Override
        protected BigDecimal compute() { 
          System.err.println(Thread.currentThread().getName() + "=" + n);
    	  //如果足够小，直接返回，不需要分叉合并
          if (n <= 2)	//第一和第二个数都是1
            return BigDecimal.ONE;
          Fibonacci f1 = new Fibonacci(n - 1);
          Fibonacci f2 = new Fibonacci(n - 2);
          f2.fork();//分叉
          return f1.invoke().add(f2.join());//第一个任务的执行结果加上第二个任务的合并结果
        }
      }

# 实践：获取maven中的前20个仓库 #

maven center中注册的前20个仓库地址可以通过以下网址获取：

*  https://mvnrepository.com/repos?p=1
 *  https://mvnrepository.com/repos?p=2

一个网址可以获取10个。如果采用顺序获取的方法，要消耗两倍的时间。而采用fork/join框架，可以并行获取，大大节省了时间。

下面是获取一页数据的方法：

public static List<Map<String, Object>> repository(int page) { 
      return Try.ofCallable(() -> { 
        Document document = Jsoup.connect("https://mvnrepository.com/repos?p=" + page).get();
        Element body = document.body();
        Elements orders = body.select(".im-title span");
        Elements titles = body.select(".im-title a");
        Elements urls = body.select(".im-subtitle");
        List<Map<String, Object>> list = new ArrayList<>();
        Do.loop(t -> { 
          Map<String, Object> map = new HashMap<>();
          String order = orders.get(t).text();
          map.put("order", Integer.parseInt(order.substring(0, order.indexOf("."))));
          map.put("name", titles.get(t).text());
          map.put("url", urls.get(t).text());
          list.add(map);
        }, titles.size());
        return list;
      }).getOrElse(new ArrayList<>());
    }

然后是分叉合并任务：

public static class Task extends RecursiveTask<List<Map<String, Object>>> { 
    
      private final List<Integer> pages;
    
      public Task(List<Integer> pages) { 
        this.pages = pages;
      }
    
      @Override
      protected List<Map<String, Object>> compute() { 
      	//如果任务足够小，直接执行，不需要分叉与合并
        if (pages.size() == 1) { 
          return repository(pages.get(0));
        }
        int middle = pages.size() / 2;
        Task task1 = new Task(new ArrayList<>(pages.subList(0, middle)));
        Task task2 = new Task(new ArrayList<>(pages.subList(middle, pages.size())));
        List<Map<String, Object>> res = new ArrayList<>();
        task2.fork();	//任务2分叉（异步）
        res.addAll(task1.invoke());		//加入任务1的直接执行结果
        res.addAll(task2.join());	//加入任务2的合并结果（分叉的任务要合并回来才行）
        return res;
      }
    }

最后是ForkJoinPool执行：

public static void main(String[] args) { 
        ForkJoinPool pool = new ForkJoinPool();
        List<Map<String, Object>> res = pool.invoke(new Task(Arrays.asList(1, 2)));
        System.err.println(res);
      }

结果如下：

[{ name=Central, url=https://repo1.maven.org/maven2/, order=1}, { name=Sonatype, url=https://oss.sonatype.org/content/repositories/releases/, order=2}, { name=Spring Plugins, url=https://repo.spring.io/plugins-release/, order=3}, { name=Spring Lib M, url=https://repo.spring.io/libs-milestone/, order=4}, { name=Hortonworks, url=https://repo.hortonworks.com/content/repositories/releases/, order=5}, { name=Atlassian, url=https://maven.atlassian.com/content/repositories/atlassian-public/, order=6}, { name=JCenter, url=https://jcenter.bintray.com/, order=7}, { name=JBossEA, url=https://repository.jboss.org/nexus/content/repositories/ea/, order=8}, { name=JBoss Releases, url=https://repository.jboss.org/nexus/content/repositories/releases/, order=9}, { name=WSO2 Releases, url=https://maven.wso2.org/nexus/content/repositories/releases/, order=10}, { name=Spring Lib Release, url=https://repo.spring.io/libs-release/, order=11}, { name=WSO2 Public, url=https://maven.wso2.org/nexus/content/repositories/public/, order=12}, { name=IBiblio, url=https://maven.ibiblio.org/maven2/, order=13}, { name=XWiki Releases, url=https://maven.xwiki.org/releases/, order=14}, { name=Kotlin Dev, url=https://dl.bintray.com/kotlin/kotlin-dev/, order=15}, { name=Nuxeo, url=https://maven-eu.nuxeo.org/nexus/content/repositories/public-releases/, order=16}, { name=Clojars, url=https://clojars.org/repo/, order=17}, { name=Gradle Plugins, url=https://plugins.gradle.org/m2/, order=18}, { name=Geomajas, url=http://maven.geomajas.org/, order=19}, { name=Redhat GA, url=https://maven.repository.redhat.com/ga/, order=20}]

可以把这些仓库放入maven的settings.xml中，不愁找不到jar。

[Link 1]: https://docs.oracle.com/javase/tutorial/essential/concurrency/forkjoin.html
[ForkBlur]: https://docs.oracle.com/javase/tutorial/essential/concurrency/examples/ForkBlur.java
[Link 2]: https://mathworld.wolfram.com/FibonacciNumber.html