网站免费建设价格,app软件下载免费,广西营销型网站公司,网站统计付费rx.observable我想探索一个使用Java 8 CompletableFuture和Rx-Java Observable的简单分散聚集场景。 场景很简单–产生大约10个任务#xff0c;每个任务返回一个字符串#xff0c;最终将结果收集到一个列表中。 顺序的 其顺序版本如下#xff1a; public void testSequen… rx.observable 我想探索一个使用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.htmlrx.observable
