时间戳

首页 > 解决方案 > 我正在尝试使用 Spring webflux 使用弹性搜索 API 以使我的 API 端点非阻塞

spring-boot - 我正在尝试使用 Spring webflux 使用弹性搜索 API 以使我的 API 端点非阻塞

问题描述

我在 Spring boot 2 中创建一个端点并使用 Spring webflux。在此端点中，我将从调用者那里获取纬度和经度，并将基于此返回状态。为了获取状态，我正在调用弹性搜索 API 来获取数据。

我能够从弹性搜索 API 获得响应，如下所示：

{
  "took": 11,
  "timed_out": false,
  "_shards": {
    "total": 1,
    "successful": 1,
    "failed": 0
  },
  "hits": {
    "total": 117252,
    "max_score": null,
    "hits": [
      {
        "_index": "geolocation",
        "_type": "geolocationdata",
        "_id": "AWt0m6GJqkN7DgSP9Lsd",
        "_score": null,
        "_source": {
          "network": "117.254.200.0/22",
          "geonameId": 1262062,
          "registeredCountrygeonameId": 1269750,
          "representedCountrygeonameId": "",
          "postalCode": "370655",
          "location": "23.2667,68.8333",
          "accuracyRadius": 100,
          "localecode": "en",
          "continentcode": "AS",
          "continentname": "Asia",
          "countryisocode": "IN",
          "countryname": "India",
          "subdivision1isocode": "GJ",
          "subdivision1nname": "Gujarat",
          "subdivision2isocode": "",
          "subdivision2nname": "",
          "cityName": "Naliya",
          "metroCode": "",
          "timeZone": "Asia/Kolkata"
        },
        "sort": [
          6986.775031169917
        ]
      }
    ]
  }
}

一旦我有了这个 JSON，我想只从中获取必要的字段并构建我的 API 所需的模型，并将其返回给调用者。

这是我使用 Elastic 搜索 API 并获得结果的方式

private WebClient webClient;

@PostConstruct
public void init() {
    this.webClient = WebClient.builder()
        .baseUrl("http://172.24.5.162:9200/geolocation")
        .defaultHeader(
            HttpHeaders.CONTENT_TYPE, 
            MediaType.APPLICATION_JSON_VALUE)
        .build();
}

public String getGeoname(String latitude, String longitude) throws Exception {
    try {
        String req = "{\"from\": 0,\"size\": 1,\"sort\": {\"_geo_distance\": {\"location\": {\"lat\": " + latitude
                + ",\"lon\": " + longitude
                + "},\"order\": \"asc\",\"unit\": \"km\",\"distance_type\": \"plane\"}}}";

        final String test;

        //result from Elastic search API
        Mono<String> result = webClient.post()
                                 .uri("/_search")
                                 .body(Mono.just(req), String.class)
                                 .retrieve().bodyToMono(String.class);

    } catch (Exception ex) {
        log.error("Exception while sending request to Elastic search Lat: " + latitude + " Long: " + longitude, ex);
        return gson.toJson(new ErrorModel(ErrorCodes.BAD_INPUT, "Bad Input"));
    }
    return "";
}

在结果变量中，我有上面显示的 JSON 作为 Mono。如果我将在结果变量上使用 block() 方法来获取 JSON IWant 字符串，那么它将阻塞主线程并变为阻塞。我的要求是使用这个 Mono 以便我可以进行如下操作（基本上我正在构建 GeoLocation 我的模型）

String hits = "";
JSONObject jsonObject = new JSONObject(o);

if (jsonObject.has("hits") && jsonObject.getJSONObject("hits").has("hits")) {
    hits = jsonObject.getJSONObject("hits")
        .getString("hits");

    hits = hits.substring(1);

    JSONObject hitsJson = new JSONObject(hits);
    JSONObject source = new JSONObject();
    if (hitsJson.has("_source")) {
        source = hitsJson.getJSONObject("_source");
        GeoLocation geolocation = new GeoLocation(source.getString("continentname"),
        source.getString("countryname"), 
        source.getString("subdivision1nname"),
        source.getString("cityName"));

        geoLocationResponse = Mono.just(gson.toJson(geolocation));

如何以非阻塞方式执行上述操作并将结果返回给我的端点调用者？我正在考虑从我的 RestController 返回 Mono

标签： spring-bootspring-webflux

解决方案

首先，您不要在反应式世界中使用 try catch，我们Mono.error会在信号链的后期创建并处理这些错误。

您真的应该查看一些教程或尝试反应堆项目文档中的基本入门。这会对你有很大帮助。

您应该利用 Spring Boot 中捆绑的 jackson，而不是使用 JSONObject 和 Gson。

public Mono<GeoLocation> getGeoname(String latitude, String longitude) {
    final String req = "{\"from\": 0,\"size\": 1,\"sort\": {\"_geo_distance\": {\"location\": {\"lat\": " + latitude
                + ",\"lon\": " + longitude
                + "},\"order\": \"asc\",\"unit\": \"km\",\"distance_type\": \"plane\"}}}";

    return webClient.post()
            .uri("/_search")
            .body(Mono.just(req), String.class)
            .exchange()
            .flatMap(clientResponse -> {
                // Check status if not 200 create a mono error that we will act upon later
                if(clientResponse.rawStatusCode() != 200)
                    return Mono.error(RuntimeException::new);
                return clientResponse.bodyToMono(String.class);
            }).map(body -> {

                // if you want to work with strings then do your mapping here
                // otherwise replace bodyToMono(FooBar.class) the FooBar class with a              
                // representation of the returned body and jackson will map to it.

                return new GeoLocation();
            }).doOnError(throwable -> {
                // handle the mono error we created earlier here, and throw an exception.
                // Then handle the exception later in a global exception handler
                throw new RuntimeException(throwable);
            });
}

推荐阅读