Spring 5（以及Boot 2，在數(shù)周之內(nèi)到貨）是一次革命。 不是“ XML上的注釋 ”或“ Java上的注釋類 ”的革命。 這是一個真正的革命性框架，可以編寫全新的應(yīng)用程序類別。 近年來，我對此框架感到有些恐懼。 “ Spring Cloud是簡化了Spring Boot使用的框架，Spring簡化了Spring使用的框架，是簡化了企業(yè)開發(fā)的框架。” start.spring.io （也稱為“ 開始…點彈簧…點I…O ”）列出了120個可以添加到服務(wù)中的不同模塊（！）。 這些天的春天變成了一個龐大的傘式項目，我可以想象為什么有些人（仍然！）偏愛Java EE（或這些天叫什么）。

但是Spring 5帶來了革命性的革命。 它不再只是阻止servlet API和各種Web框架的包裝器。 在Project Reactor之上的Spring 5允許編寫高性能，超快速和可伸縮的服務(wù)器，完全避免了servlet堆棧。 該死的，CLASSPATH上沒有Jetty甚至servlet API！ 在Spring 5 Web-flux的核心，我們將找到Netty ，這是一個用于編寫異步客戶端和服務(wù)器的低級框架。 最終，Spring成為反應(yīng)框架家族的一等公民。 Java開發(fā)人員可以實現(xiàn)快速服務(wù)，而不必離開自己的舒適區(qū)，也可以使用https://doc.akka.io/docs/akka-http/current/或https://www.playframework.com/ 。 Spring 5是用于構(gòu)建高度可擴(kuò)展且具有彈性的應(yīng)用程序的完全被動的現(xiàn)代工具。 盡管如此，諸如控制器，Bean，依賴注入之類的基本原理都是相同的。 而且，升級路徑很順利，我們可以逐步添加功能，而不是學(xué)習(xí)全新的外來框架。 足夠多的談話，讓我們寫一些代碼。

在本文中，我們將編寫一個簡單的無頭應(yīng)用程序，該應(yīng)用程序可以在ElasticSearch中大量索引文檔。 我們的目標(biāo)是即使服務(wù)器速度變慢，也只需要幾個線程即可實現(xiàn)數(shù)千個并發(fā)連接。 但是，與Spring Data MongoDB不同， Spring Data ElasticSearch本身不支持非阻塞存儲庫。 好吧，后者似乎不再維護(hù)了，當(dāng)前版本已經(jīng)3年了。 許多文章定位Spring 5 +的MongoDB其倉庫返回?zé)o阻塞流（ Flux或Flowable的RxJava）。 這一點會更高級。

ElasticSearch 6 Java API使用RESTful接口，并使用非阻塞HTTP客戶端實現(xiàn)。 不幸的是，它使用回調(diào)而不是像CompletableFuture這樣的理智的東西。 因此，讓我們自己構(gòu)建客戶端適配器。

使用Fluxes和Monos的ElasticSearch客戶端

本文的源代碼可在react reactive-elastic-search分支上的github.com/nurkiewicz/elastic-flux上找到。

我們想通過返回Flux或Mono來構(gòu)建一個支持Project Reactor的ElasticSearch Java客戶端。 當(dāng)然，如果基礎(chǔ)流是完全異步的并且不消耗線程，則將獲得最大的好處。 幸運的是，Java API就是這樣。 首先，讓我們將ElasticSearch的客戶端設(shè)置為Spring Bean：

import org.apache.http.HttpHost; import org.elasticsearch.client.RestClient; import org.elasticsearch.client.RestHighLevelClient;@Bean RestHighLevelClient restHighLevelClient() {return new RestHighLevelClient(RestClient.builder(new HttpHost("localhost", 9200)).setRequestConfigCallback(config -> config.setConnectTimeout(5_000).setConnectionRequestTimeout(5_000).setSocketTimeout(5_000)).setMaxRetryTimeoutMillis(5_000)); }

在現(xiàn)實生活中，我們顯然會參數(shù)化大多數(shù)這些東西。 我們將為簡單的JSON文檔建立索引，目前它們的內(nèi)容并不重要：

@Value class Doc {private final String username;private final String json; }

我們將編寫的代碼包裝RestHighLevelClient并通過返回Mono<IndexResponse>使它更高級 。 Mono非常類似于CompletableFuture但有兩個例外：

• 這很懶–只要您不訂閱，就不會開始計算
• 與CompletableFuture不同， Mono可以正常完成而不會發(fā)出任何值

第二個區(qū)別總是對我有些誤導(dǎo)。 在RxJava 2.x中，有兩種不同的類型： Single （總是帶有值或錯誤來完成）和Maybe （類似于Mono ）。 太糟糕的Reactor并沒有做到這一點。 沒關(guān)系，適配器層是什么樣的？ 普通的Elastic API如下所示：

client.indexAsync(indexRequest, new ActionListener() {@Overridepublic void onResponse(IndexResponse indexResponse) {//got response}@Overridepublic void onFailure(Exception e) {//got error} });

您可以看到前進(jìn)的方向： callback hell 。 與其將自定義ActionListener公開為該邏輯的參數(shù)，不如將其包裝在Mono ：

import org.elasticsearch.action.ActionListener; import org.elasticsearch.action.index.IndexRequest; import org.elasticsearch.action.index.IndexResponse; import org.elasticsearch.client.RestHighLevelClient; import org.elasticsearch.common.xcontent.XContentType;import reactor.core.publisher.Mono; import reactor.core.publisher.MonoSink;private Mono<IndexResponse> indexDoc(Doc doc) {return Mono.create(sink -> {IndexRequest indexRequest = new IndexRequest("people", "person", doc.getUsername());indexRequest.source(doc.getJson(), XContentType.JSON);client.indexAsync(indexRequest, new ActionListener<IndexResponse>() {@Overridepublic void onResponse(IndexResponse indexResponse) {sink.success(indexResponse);}@Overridepublic void onFailure(Exception e) {sink.error(e);}});}); }

我們必須創(chuàng)建IndexRequest包裝JSON文檔，并通過RESTful API發(fā)送它。 但這不是重點。 我們正在使用Mono.create()方法，它有一些缺點，但稍后會介紹更多。 Mono是懶惰的，因此僅調(diào)用indexDoc()還不夠，沒有對ElasticSearch發(fā)出HTTP請求。 但是，每次有人訂閱此單元素源時，都會執(zhí)行create()內(nèi)部的邏輯。 關(guān)鍵行是sink.success()和sink.error() 。 它們將結(jié)果從ElasticSearch（來自后臺異步線程）傳播到流中。 在實踐中如何使用這種方法？ 非常簡單！

Doc doc = //... indexDoc(doc).subscribe(indexResponse -> log.info("Got response"));

當(dāng)然，反應(yīng)流處理的真正能力來自于組合多個流。 但是我們邁出了第一步：將基于回調(diào)的異步API轉(zhuǎn)換為通用流。 如果您不愿意使用MongoDB，它會在存儲庫中內(nèi)置支持諸如Mono或Flux類的反應(yīng)類型。 Cassandra和Redis也是如此 。 在下一篇文章中，我們將學(xué)習(xí)如何生成一些虛假數(shù)據(jù)并對其進(jìn)行索引。

翻譯自: https://www.javacodegeeks.com/2018/01/spring-reactor-elasticsearch-callbacks-reactive-streams.html

總結(jié)

以上是生活随笔為你收集整理的Spring，Reactor和ElasticSearch：从回调到反应流的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

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