Java 内Fork/Join框架的尝试

整个周末 都是在咳嗽中度过。夏日的感冒 感觉要死的节奏。没法思考 没法学习 是最尴尬的事情 喝了药就想睡觉。

那么 只能把存货 上一下。

Fork / Join框架是使用并发分治法解决问题的框架。引入它们是为了补充现有的并发API。在介绍它们之前 现有的Executor Service实现是运行异步任务的流行选择 但是当任务同质且独立时 它们会发挥最佳作用。运行依赖的任务并使用这些实现来组合其结果并不容易。随着Fork / Join框架的引入 试图解决这一缺陷。

简单粗暴地直接跳入代码。在代码中创建一个任务 该任务将返回列表的所有元素的总和。以下步骤以伪代码表示我们的算法

01.查找列表的中间索引

02.在中间划分列表

03.递归创建一个新任务 该任务将计算左侧部分的总和

04.递归创建一个新任务 该任务将计算正确部分的总和

05.将左总和 中间元素和右总和的结果相加

 Slf4j

public class ListSummer extends RecursiveTask Integer {

 private final List Integer listToSum;

 ListSummer(List Integer listToSum) {

 this.listToSum listToSum;

 Override

 protected Integer compute() {

 if (listToSum.isEmpty()) {

 log.info( 空集合 

 return 0;

 int middleIndex listToSum.size() / 2;

 log.info( 集合 {}, 中部索引: {} , listToSum, middleIndex);

 List Integer leftSublist listToSum.subList(0, middleIndex);

 List Integer rightSublist listToSum.subList(middleIndex 1, listToSum.size());

 ListSummer leftSummer new ListSummer(leftSublist);

 ListSummer rightSummer new ListSummer(rightSublist);

 leftSummer.fork();

 rightSummer.fork();

 Integer leftSum leftSummer.join();

 Integer rightSum rightSummer.join();

 int total leftSum listToSum.get(middleIndex) rightSum;

 log.info( 左部分和 {}, 右部分和 {}, 总和 {} , leftSum, rightSum, total);

 return total;

}

首先 我们通过继承ForkJoinTask的子类RecursiveTask。这是能够在执行并发任务并返回结果时的子类。当任务不返回结果而仅执行效果时 继承递归操作RecursiveAction的子类。对于我们解决的大多数实际任务 这两个子类就足够了。

其次 RecursiveTask和RecursiveAction都定义了一种抽象计算方法compute。这是进行计算执行代码的地方。

第三 在计算方法内部 检查通过构造函数传递的列表的大小。如果为空 则已经知道总和的结果为零 然后立即返回。否则 将列表分为两个子列表 并创建List Summer类型的两个实例。然后 在这两个实例上调用fork 在ForkJoinTask中定义 方法

leftSummer.fork();

rightSummer.fork();

导致将这些任务安排为异步执行的原因 稍后将在本文中解释用于此目的的确切机制。

之后 调用join 方法 也在ForkJoinTask中定义 等待这两个部分的结果

Integer leftSum leftSummer.join();

Integer rightSum rightSummer.join();

然后将其与列表的中间元素相加以获得最终结果。

代码内添加了许多日志消息 以使更易于理解。

注 当我们处理包含数千个条目的列表时 进行详细的日志记录 尤其是记录整个列表 对于解决方案来说其实很不优雅。

差不多了 现在 为测试运行创建一个测试类

public class ListSummerTest {

 Test

 public void shouldSumEmptyList() {

 ListSummer summer new ListSummer(List.of());

 ForkJoinPool forkJoinPool new ForkJoinPool();

 forkJoinPool.submit(summer);

 int result summer.join();

 assertThat(result).isZero();

 Test

 public void shouldSumListWithOneElement() {

 ListSummer summer new ListSummer(List.of(5));

 ForkJoinPool forkJoinPool new ForkJoinPool();

 forkJoinPool.submit(summer);

 int result summer.join();

 assertThat(result).isEqualTo(5);

 Test

 public void shouldSumListWithMultipleElements() {

 ListSummer summer new ListSummer(List.of(

 1, 2, 3, 4, 5, 6, 7, 8, 9

 ForkJoinPool forkJoinPool new ForkJoinPool();

 forkJoinPool.submit(summer);

 int result summer.join();

 assertThat(result).isEqualTo(45);

}

测试类中 创建一个 ForkJoinPool的实例。

ForkJoinPool是用于运行Fork Join任务的独特Executor Service实现。它采用一种称为工作窃取算法的特殊算法。与其他执行器服务实现相反 在其他实现器服务实现中 只有一个队列包含要执行的所有任务 在工作窃取实现中 每个工作线程都获得其工作队列。每个线程都从其队列开始执行任务。具体的原理 可以查看源代码进行查看。

当我们检测到Fork Join Task可以分解为多个较小的子任务时 便将它们分解为较小的任务 然后在这些任务上调用fork 方法。这种调用导致子任务被推入执行线程的队列中。在执行期间 当一个线程用尽队列/没有要执行的任务时 它可以从另一线程的队列中“窃取”任务 因此称为“工作窃取” 。与使用任何其他Executor Service实现相比 这种窃取行为可以带来更高的吞吐量。

之前 当我们在代码中执行leftSummer和rightSummer对象的fork 时 它们被推入执行线程的工作队列中 之后它们被池中的其他活动线程“窃取” 依此类推)。

刚才解决的问题本质上是非阻塞的。如果想解决一个阻塞操作的问题 那么为了获得更好的吞吐量 则需要改变策略。

用另一个代码实例来描述一下。假设要创建一个非常简单的网页爬虫。该爬虫将接收HTTP链接列表 执行GET请求以获取响应体 然后计算响应长度。

这是代码–

 Slf4j

public class ResponseLengthCalculator extends RecursiveTask Map String, Integer {

 private final List String links;

 ResponseLengthCalculator(List String links) {

 this.links links;

 Override

 protected Map String, Integer compute() {

 if (links.isEmpty()) {

 log.info( 没获取更多链接 

 return Collections.emptyMap();

 int middle links.size() / 2;

 log.info( 索引: {} , links, middle);

 ResponseLengthCalculator leftPartition new ResponseLengthCalculator(links.subList(0, middle));

 ResponseLengthCalculator rightPartition new ResponseLengthCalculator(links.subList(middle 1, links.size()));

 log.info( 执行左区域加入 

 leftPartition.fork();

 log.info( 左区域被加入 现在执行右区域加入 

 rightPartition.fork();

 log.info( 右区域被加入 

 String middleLink links.get(middle);

 HttpRequester httpRequester new HttpRequester(middleLink);

 String response;

 try {

 log.info( 执行回调{} , middleLink);

 ForkJoinPool.managedBlock(httpRequester);

 response httpRequester.response;

 } catch (InterruptedException ex) {

 log.error( 发生异常 , ex);

 response 

 Map String, Integer responseMap new HashMap (links.size());

 Map String, Integer leftLinks leftPartition.join();

 responseMap.putAll(leftLinks);

 responseMap.put(middleLink, response.length());

 Map String, Integer rightLinks rightPartition.join();

 responseMap.putAll(rightLinks);

 log.info( 左部链接集合{}, 响应长度 {}, 右部链接集合{} , leftLinks, response.length(), rightLinks);

 return responseMap;

 private static class HttpRequester implements ForkJoinPool.ManagedBlocker {

 private final String link;

 private String response;

 private HttpRequester(String link) {

 this.link link;

 Override

 public boolean block() {

 HttpGet headRequest new HttpGet(link);

 CloseableHttpClient client HttpClientBuilder

 .create()

 .disableRedirectHandling()

 .build();

 try {

 log.info( 正在执行请求{} , link);

 CloseableHttpResponse response client.execute(headRequest);

 log.info( 链接url{}已经被请求 , link);

 this.response EntityUtils.toString(response.getEntity());

 } catch (IOException e) {

 log.error( 当从链接url{}中获取响应报错:{} , link, e.getMessage());

 this.response 

 return true;

 Override

 public boolean isReleasable() {

 return false;

}

上述代码创建了一个ForkJoinPool.Managed Blocker的实现类HttpRequester 在其中放置阻塞的HTTP调用。该接口定义了两个方法-block 和is Releasable 。block 方法是我们进行阻塞调用的地方。在完成阻塞操作之后 我们返回true 指示不再需要进一步的阻塞。我们从is Releasable 实现中返回false 以向fork-join工作线程指示block 方法实现实际上可能在阻塞。is Releasable 实现将在调用block 方法之前先由fork-join工作线程调用。最后 我们通过调用Fork Join Pool.managed Block 静态方法将Http Requester实例提交到池中。之后 我们的阻止任务将开始执行。当它阻止HTTP请求时 Fork Join Pool.managed Block 方法还将安排在必要时激活备用线程 以确保足够的并行性。

测试代码如下所示:

public class ResponseLengthCalculatorTest {

 Test

 public void shouldReturnEmptyMapForEmptyList() {

 ResponseLengthCalculator responseLengthCalculator new ResponseLengthCalculator(Collections.emptyList());

 ForkJoinPool pool new ForkJoinPool();

 pool.submit(responseLengthCalculator);

 Map String, Integer result responseLengthCalculator.join();

 assertThat(result).isEmpty();

 Test

 public void shouldHandle200Ok() {

 ResponseLengthCalculator responseLengthCalculator new ResponseLengthCalculator(List.of(

 http://huclele.ml/200 

 ForkJoinPool pool new ForkJoinPool();

 pool.submit(responseLengthCalculator);

 Map String, Integer result responseLengthCalculator.join();

 assertThat(result)

 .hasSize(1)

 .containsKeys( http://huclele.ml/200 )

 .containsValue(0);

 Test

 public void shouldFetchResponseForDifferentResponseStatus() {

 ResponseLengthCalculator responseLengthCalculator new ResponseLengthCalculator(List.of(

 http://huclele.ml/200 ,

 http://huclele.ml/302 ,

 http://huclele.ml/404 ,

 http://huclele.ml/502 

 ForkJoinPool pool new ForkJoinPool();

 pool.submit(responseLengthCalculator);

 Map String, Integer result responseLengthCalculator.join();

 assertThat(result)

 .hasSize(4);

}

今天 聊了Fork/Join的用法 以及两个实例 还有更多玩法 可以学习。有什么问题 留言给我o~