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

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

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

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

使用Fluxes和Monos的ElasticSearch客戶端

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

我們想通過返回Flux或Mono來構建一個支持Project Reactor的ElasticSearch Java客戶端。 當然，如果基礎流是完全異步的并且不消耗線程，則將獲得最大的好處。 幸運的是，Java API就是這樣。 首先，讓我們將ElasticSearch的客戶端設置為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)); }

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

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

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

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

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

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

您可以看到前進的方向： callback hell 。 與其將自定義ActionListener公開為該邏輯的參數，不如將其包裝在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);}});}); }

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

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

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

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

總結

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

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