我在使用Netflix Rx-Java庫方面有良好的工作經驗，并且以前曾寫過關于使用Rx-Java和Java 8 CompletableFuture解決分散式問題的博客。 在這里，我想探索使用Spring Reactor Core庫應用相同的模式。

tldr –如果您熟悉Netflix Rx-Java，您已經很熟悉Spring Reactor Core，API的地圖，并且我很高興看到Spring Reactor團隊在Javadoc API中勤奮地使用了Marble圖。

另一個快速點是， rx.Observable根據是否要發射許多項目或是否要發射一個項目來映射到Flux或Mono 。

有了這個，我可以直接進入示例–我執行了一個簡單的任務（使用延遲模擬），該任務被生成了幾次，我需要同時執行這些任務，然后收集結果，使用rx表示如下。可觀察的代碼：

@Test public void testScatterGather() throws Exception {ExecutorService executors = Executors.newFixedThreadPool(5);List<Observable<String>> obs =IntStream.range(0, 10).boxed().map(i -> generateTask(i, executors)).collect(Collectors.toList());Observable<List<String>> merged = Observable.merge(obs).toList();List<String> result = merged.toBlocking().first();logger.info(result.toString());}private Observable<String> generateTask(int i, ExecutorService executorService) {return Observable.<String>create(s -> {Util.delay(2000);s.onNext( i + "-test");s.onCompleted();}).subscribeOn(Schedulers.from(executorService)); }

請注意，我純粹是出于測試目的。 現在，使用Spring Reactor Core的類似代碼可以轉換為以下代碼：

@Test public void testScatterGather() {ExecutorService executors = Executors.newFixedThreadPool(5);List<Flux<String>> fluxList = IntStream.range(0, 10).boxed().map(i -> generateTask(executors, i)).collect(Collectors.toList());Mono<List<String>> merged = Flux.merge(fluxList).toList();List<String> list = merged.get();logger.info(list.toString());}public Flux<String> generateTask(ExecutorService executorService, int i) {return Flux.<String>create(s -> {Util.delay(2000);s.onNext(i + "-test");s.onComplete();}).subscribeOn(executorService); }

它或多或少地一對一映射。 Mono類型的區別很小，我個人認為該類型是反應式庫的不錯的介紹，因為它可以很清楚地表明是否發出了多個項目，而只發出了一個。樣品。 這些對我來說仍然是早期的探索，我希望對這個優秀的圖書館更加熟悉。

翻譯自: https://www.javacodegeeks.com/2016/04/scatter-gather-using-spring-reactor-core.html

總結

以上是生活随笔為你收集整理的使用Spring Reactor Core进行分散收集的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。