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

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

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

有了這個，我可以直接進入示例–我執(zhí)行了一個簡單的任務（使用延遲模擬），該任務被生成了幾次，我需要同時執(zhí)行這些任務，然后收集結(jié)果，使用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)); }

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

@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類型的區(qū)別很小，我個人認為該類型是反應式庫的不錯的介紹，因為它可以很清楚地表明是否發(fā)出了多個項目，而只發(fā)出了一個。樣品。 這些對我來說仍然是早期的探索，我希望對這個優(yōu)秀的圖書館更加熟悉。

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

總結(jié)

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

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯，歡迎將生活随笔推薦給好友。