加强本单位政务网站建设,自己做网站上市,怎么样免费做网站,吉安市建设规划局网站我想使用Java 8 CompletableFuture和Rx-Java Observable探索一个简单的分散聚集场景。 场景很简单–产生大约10个任务#xff0c;每个任务返回一个字符串#xff0c;最终将结果收集到一个列表中。 顺序的 其顺序版本如下#xff1a; public void testSequentialScatterGa… 我想使用Java 8 CompletableFuture和Rx-Java Observable探索一个简单的分散聚集场景。 场景很简单–产生大约10个任务每个任务返回一个字符串最终将结果收集到一个列表中。 顺序的 其顺序版本如下 public void testSequentialScatterGather() throws Exception {ListString list IntStream.range(0, 10).boxed().map(this::generateTask).collect(Collectors.toList());logger.info(list.toString());
}private String generateTask(int i) {Util.delay(2000);return i - test;
}随着CompletableFuture 可以使用称为supplyAsync的实用程序方法来使方法返回CompletableFuture我正在使用此方法的一种变体它接受要使用的显式Executor 而且我故意为其中一个输入抛出异常 private CompletableFutureString generateTask(int i,ExecutorService executorService) {return CompletableFuture.supplyAsync(() - {Util.delay(2000);if (i 5) {throw new RuntimeException(Run, it is a 5!);}return i - test;}, executorService);
} 现在分散任务 ListCompletableFutureString futures IntStream.range(0, 10).boxed().map(i - this.generateTask(i, executors).exceptionally(t - t.getMessage())).collect(Collectors.toList()); 在分散任务结束时结果是CompletableFuture列表。 现在要从中获取String列表有些棘手这里我使用Stackoverflow中建议的一种解决方案 CompletableFutureListString result CompletableFuture.allOf(futures.toArray(new CompletableFuture[futures.size()])).thenApply(v - futures.stream().map(CompletableFuture::join).collect(Collectors.toList())); 这里仅使用CompletableFuture.allOf方法来构成下一步操作一旦所有分散的任务都完成了则一旦完成任务期货就会再次流式传输并收集到一个字符串列表中。 然后可以异步显示最终结果 result.thenAccept(l - {logger.info(l.toString());
});使用Rx-java Observable 使用Rx-java进行分散收集相对比CompletableFuture版本更干净因为Rx-java提供了更好的方法将结果组合在一起这也是执行分散任务的方法 private ObservableString generateTask(int i, ExecutorService executorService) {return Observable.Stringcreate(s - {Util.delay(2000);if ( i 5) {throw new RuntimeException(Run, it is a 5!);}s.onNext( i -test);s.onCompleted();}).onErrorReturn(e - e.getMessage()).subscribeOn(Schedulers.from(executorService));
} 并分散任务 ListObservableString obs IntStream.range(0, 10).boxed().map(i - generateTask(i, executors)).collect(Collectors.toList()); 我又有了一个Observable的列表而我需要的是一个结果列表Observable提供了一个合并方法来做到这一点 ObservableListString merged Observable.merge(obs).toList(); 可以订阅并在可用时打印结果 merged.subscribe(l - logger.info(l.toString()));翻译自: https://www.javacodegeeks.com/2015/08/using-java-8-completablefuture-and-rx-java-observable.html
