首页 技术 正文

Elasticsearch Java Rest Client简述

技术 2022年11月6日
0 收藏 851 点赞 913 浏览 16379 个字

ESJavaClient的历史

JavaAPI Client

  • 优势：基于transport进行数据访问，能够使用ES集群内部的性能特性，性能相对好
  • 劣势：client版本需要和es集群版本一致，数据序列化通过java实现，es集群或jdk升级，客户端需要伴随升级。

     ES官网最早提供的Client，spring-data-elasticsearch也基于该client开发，使用transport接口进行通信，其工作方式是将webserver当做集群中的节点，获取集群数据节点(DataNode)并将请求路由到Node获取数据将，返回结果在webserver内部进行结果汇总。 client需要与es集群保持相对一致性，否则会出现各种『奇怪』的异常。由于ES集群升级很快，集群升级时客户端伴随升级的成本高。 官网已声明es7.0将不在支持transport client(API Client),8.0正时移除 

REST Client

  • 优势：REST风格交互，符合ES设计初衷；兼容性强；
  • 劣势：性能相对API较低

 ESREST基于http协议，客户端发送请求到es的任何节点，节点会通过transport接口将请求路由到其他节点，完成数据处理后汇总并返回客户端，客户端负载低，。 RestFul是ES特性之一，但是直到5.0才有了自己的客户端，6.0才有的相对好用的客户端。在此之前JestClient作为第三方客户端，使用非常广泛。 本文将对Java Rest Client、Java High Level Client、Jest三种Restfull风格的客户端做简单的介绍，个人推荐使用JestClient，详见后文。 

Java Rest Client

三种客户端中，感觉最『原始』的客户端，感觉在使用HttpClient，请求参数繁多，几乎没有提供Mapping能力，请求结束后需要拿着json一层层解析，对于开发来说大部分情况下不需要关心”took”、”_shards”、”timed_out”这些属性，这些更多依赖集群的稳定性，如何方便快捷的拿到_source中的数据才是首要编码内容。

{
    "took": 9,
    "timed_out": false,
    "_shards": {
        "total": 5,
        "successful": 5,
        "skipped": 0,
        "failed": 0
    },
    "hits": {
        "total": 2,
        "max_score": 0.2876821,
        "hits": [
            {
                "_index": "pangu",
                "_type": "normal",
                "_id": "2088201805281551",
                "_score": 0.2876821,
                "_source": {
                    "age": 8,
                    "country": "UK",
                    "id": "2088201805281551",
                    "name": "baby"
                }
            },
            {
                "_index": "pangu",
                "_type": "normal",
                "_id": "2088201805281552",
                "_score": 0.2876821,
                "_source": {
                    "age": 8,
                    "country": "UK",
                    "id": "2088201805281552",
                    "name": "baby"
                }
            }
        ]
    }
}

 

Demo

import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;import org.apache.http.HttpEntity;
import org.apache.http.HttpHost;
import org.apache.http.HttpStatus;
import org.apache.http.auth.AuthScope;
import org.apache.http.auth.UsernamePasswordCredentials;
import org.apache.http.client.CredentialsProvider;
import org.apache.http.entity.ContentType;
import org.apache.http.impl.client.BasicCredentialsProvider;
import org.apache.http.impl.nio.client.HttpAsyncClientBuilder;
import org.apache.http.nio.entity.NStringEntity;
import org.apache.http.util.EntityUtils;
import org.elasticsearch.client.Response;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestClientBuilder.HttpClientConfigCallback;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.search.aggregations.AggregationBuilders;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.elasticsearch.search.sort.SortOrder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;/**
 * @author yanlei
 * @version $Id: RestClientDemo.java, v 0.1 2018年05月26日 下午12:27 yanlei Exp $
 */
public class RestClientDemo {    private static final Logger LOGGER = LoggerFactory.getLogger(RestClientDemo.class);    /**
     * index 名称
     */
    private static final String INDEX_NAME = "pangu";    /**
     * type 名称
     */
    private static final String TYPE_NAME = "normal";    private static RestClient restClient;    static {
        restClient = RestClient.builder(new HttpHost("search.alipay.com", 9999, "http"))
                .setFailureListener(new RestClient.FailureListener() { // 连接失败策略
                    @Override
                    public void onFailure(HttpHost host) {
                        LOGGER.error("init client error, host:{}", host);
                    }
                })
                .setMaxRetryTimeoutMillis(10000) // 超时时间
                .setHttpClientConfigCallback(new HttpClientConfigCallback() { // 认证
                    @Override
                    public HttpAsyncClientBuilder customizeHttpClient(HttpAsyncClientBuilder httpClientBuilder) {
                        final CredentialsProvider credentialsProvider = new BasicCredentialsProvider();
                        credentialsProvider.setCredentials(AuthScope.ANY, new UsernamePasswordCredentials("example", "WnhmUwjU"));
                        return httpClientBuilder.setDefaultCredentialsProvider(credentialsProvider);
                    }
                })
                .build();
    }    public boolean index(String id, String jsonString) {
        String method = "PUT";
        String endpoint = new StringBuilder().append("/").append(INDEX_NAME).append("/").append(TYPE_NAME).append("/").append(id)
                .toString();
        LOGGER.info("method={}, endpoint={}", method, endpoint);        HttpEntity entity = new NStringEntity(jsonString, ContentType.APPLICATION_JSON);
        Response response = null;
        try {
            response = restClient.performRequest(method, endpoint, Collections.emptyMap(), entity);
            if (response.getStatusLine().getStatusCode() == HttpStatus.SC_CREATED
                    || response.getStatusLine().getStatusCode() == HttpStatus.SC_OK) {
                LOGGER.info("index success");
                return true;
            } else {
                LOGGER.error("index error : {}", response.toString());
                return false;
            }
        } catch (IOException e) {
            LOGGER.error("system error, id={}, jsonString={}", id, jsonString, e);
            return false;
        }
    }    public <T> T search(String searchString, CallbackSearch<T> callbackSearch) {
        String method = "GET";
        String endpoint = new StringBuilder().append("/").append(INDEX_NAME).append("/").append(TYPE_NAME).append("/").append("_search")
                .toString();
        LOGGER.info("method={}, endpoint={}", method, endpoint);        HttpEntity entity = new NStringEntity(searchString, ContentType.APPLICATION_JSON);
        Response response = null;
        try {
            response = restClient.performRequest(method, endpoint, Collections.emptyMap(), entity);
            if (response.getStatusLine().getStatusCode() == HttpStatus.SC_OK) {
                // 提取数据
                String resultString = EntityUtils.toString(response.getEntity(), "UTF-8");
                return callbackSearch.get(resultString);
            } else {
                LOGGER.error("index false, error : {}", response);
                return null;
            }
        } catch (IOException e) {
            LOGGER.error("system error, searchString={}", searchString, e);
            return null;
        }
    }    public interface CallbackSearch<T> {
        T get(String responseString);
    }    public static void main(String[] args) {
        RestClientDemo restClientDemo = new RestClientDemo();        // 1. 索引数据
        User user = new User();
        user.setId("2088201805281345");
        user.setName("nick");
        user.setAge(16);
        user.setCountry("USA");
        boolean indexOK = restClientDemo.index(user.getId(), JSON.toJSONString(user));
        LOGGER.info("index param={} result={}", user, indexOK);        // 2. 数据检索
        SearchSourceBuilder sourceBuilder = new SearchSourceBuilder()
                .query(QueryBuilders.boolQuery().must(QueryBuilders.termQuery("name", "nick")))
                .sort("id", SortOrder.DESC)
                .from(0).size(10);        List<User> searchResult = restClientDemo.search(sourceBuilder.toString(), new CallbackSearch<List<User>>() {
            @Override
            public List<User> get(String responseString) {
                LOGGER.info("responseString={}", responseString);
                List<User> result = new ArrayList<>();
                JSONObject responseObj = JSON.parseObject(responseString);
                JSONObject hits = responseObj.getJSONObject("hits");
                if (hits.getIntValue("total") != 0) {
                    JSONArray innerHits = hits.getJSONArray("hits");
                    for (int i = 0; i < innerHits.size(); i++) {
                        JSONObject innerhit = innerHits.getJSONObject(i);
                        User user = innerhit.getObject("_source", User.class);
                        result.add(user);
                    }
                }
                return result;
            }
        });
        LOGGER.info("search param={} result={}", sourceBuilder, searchResult);        // 3. 聚合查询
        SearchSourceBuilder aggSearchSourceBuilder = new SearchSourceBuilder()
                .query(QueryBuilders.matchAllQuery())
                .aggregation(AggregationBuilders.avg("age_avg").field("age"));        Double aggResult = restClientDemo.search(aggSearchSourceBuilder.toString(), new CallbackSearch<Double>() {            @Override
            public Double get(String responseString) {
                LOGGER.info("responseString={}", responseString);
                JSONObject responseObj = JSON.parseObject(responseString);
                JSONObject aggregations = responseObj.getJSONObject("aggregations");
                Double result = aggregations.getJSONObject("age_avg").getDouble("value");                return result;
            }
        });        LOGGER.info("aggregation param={} result={}", aggSearchSourceBuilder, aggResult);
    }
}

Java High Level REST Client

5.0的RestClient很难用，官网似乎也发现了这个问题。从6.0开始推出的Java High Level REST Client明显在交互易用性方面提升了很多，提供了诸如getHit等方法获取结果，开发者可以更关注与核心数据的操作但是Java High Level REST Client似乎犯了和Java API Client的问题，底层设置了很多6.0才有的特性，导致改版本无法用于低版本ES(包含ZSearch，目前ZSearch是5.3版本)。 Elasticsearch Java Rest Client简述 

Demo

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;import com.alibaba.fastjson.JSON;import org.apache.http.HttpHost;
import org.apache.http.HttpStatus;
import org.apache.http.auth.AuthScope;
import org.apache.http.auth.UsernamePasswordCredentials;
import org.apache.http.client.CredentialsProvider;
import org.apache.http.impl.client.BasicCredentialsProvider;
import org.apache.http.impl.nio.client.HttpAsyncClientBuilder;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestClientBuilder;
import org.elasticsearch.client.RestClientBuilder.HttpClientConfigCallback;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.common.xcontent.XContentType;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;/**
 * @author yanlei
 * @version $Id: RestClientDemo.java, v 0.1 2018年05月26日 下午12:27 yanlei Exp $
 */
public class HighLevelRestClientDemo {    private static final Logger LOGGER = LoggerFactory.getLogger(HighLevelRestClientDemo.class);    private static final String INDEX_NAME = "pangu";    private static final String TYPE_NAME = "normal";    private static RestHighLevelClient client;    static {
        final CredentialsProvider credentialsProvider = new BasicCredentialsProvider();
        credentialsProvider.setCredentials(AuthScope.ANY,
                new UsernamePasswordCredentials("example", "WnhmUwjU"));        RestClientBuilder builder = RestClient.builder(
                new HttpHost("127.0.0.1", 9200, "http"))
                .setFailureListener(new RestClient.FailureListener() { // 连接失败策略
                    @Override
                    public void onFailure(HttpHost host) {
                        LOGGER.error("init client error, host:{}", host);
                    }
                })
                .setMaxRetryTimeoutMillis(10000) // 超时时间
                .setHttpClientConfigCallback(new HttpClientConfigCallback() { // 认证
                    @Override
                    public HttpAsyncClientBuilder customizeHttpClient(HttpAsyncClientBuilder httpClientBuilder) {
                        return httpClientBuilder.setDefaultCredentialsProvider(credentialsProvider);
                    }
                });        client = new RestHighLevelClient(builder);
    }    public boolean index(String id, String jsonString) {
        IndexRequest request = new IndexRequest(
                INDEX_NAME,
                TYPE_NAME,
                id);
        request.source(jsonString, XContentType.JSON);        IndexResponse response = null;
        try {
            response = client.index(request);
            if(response.status().getStatus() == HttpStatus.SC_CREATED || response.status().getStatus() == HttpStatus.SC_OK) {
                LOGGER.info("index success");
                return true;
            }else {
                LOGGER.error("index error : {}", response.toString());
                return false;
            }
        } catch (IOException e) {
            LOGGER.error("系统异常", e);
            return false;
        }
    }    public <T> T search(SearchSourceBuilder sourceBuilder, CallbackSearch<T> callbackSearch) {
        SearchRequest searchRequest = new SearchRequest()
                .indices(INDEX_NAME)
                .types(TYPE_NAME)
                .source(sourceBuilder);        SearchResponse response = null;
        try {
            response = client.search(searchRequest);
            if(response.status().getStatus() == HttpStatus.SC_OK) {
                // 提取数据
                return callbackSearch.get(response);
            }else {
                LOGGER.error("index false, error : {}", response);
                return null;
            }
        } catch (IOException e) {
            LOGGER.error("系统异常", e);
            return null;
        }
    }    public interface CallbackSearch<T> {
        T get(SearchResponse response);
    }    public static void main(String[] args) {
        HighLevelRestClientDemo restClientDemo = new HighLevelRestClientDemo();        // 1. 索引数据
        User user = new User();
        user.setId("2088201805281552");
        user.setName("baby");
        user.setAge(8);
        user.setCountry("UK");
        boolean indexOK = restClientDemo.index(user.getId(), JSON.toJSONString(user));
        LOGGER.info("index param={} result={}", user, indexOK);        // 2. 数据检索
        SearchSourceBuilder sourceBuilder = new SearchSourceBuilder()
                .query(QueryBuilders.boolQuery().must(QueryBuilders.termQuery("name", "baby")))
                //.sort("age", SortOrder.DESC)
                .from(0).size(10);        List<User> searchResult = restClientDemo.search(sourceBuilder, new CallbackSearch<List<User>>() {
            @Override
            public List<User> get(SearchResponse response) {
                List<User> result = new ArrayList<>();
                response.getHits().forEach(hit -> {
                    User user = JSON.parseObject(hit.getSourceAsString(), User.class);
                    result.add(user);
                return result;
            }
        });
        LOGGER.info("search param={} result={}", sourceBuilder, searchResult);    }
}

JestClient

伴随1.*ES诞生，底层基于HttpClient+GSON提供集群访问与数据映射，Java High Level REST Client的很多设计也借鉴了Jest。Jest在ES操作易用性方面与Java High Level REST Client部分伯仲，但是其多版本兼容性比后者强很多。虽然使用有龟速之称的GSON切不能替换，但是其性能应该能满足大部分业务场景。在18台服务器（4核+64G内存）13亿数据的数据上做性能测试，QPS达到200+时，集群出现异常，所以GSON不是首要考虑的因素。PS：200+QPS在集群没有做冷热数据切分与rounting下的效果。 

DEMO

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;import com.google.gson.Gson;
import com.google.gson.GsonBuilder;
import io.searchbox.client.JestClient;
import io.searchbox.client.JestClientFactory;
import io.searchbox.client.JestResult;
import io.searchbox.client.config.HttpClientConfig;
import io.searchbox.core.Index;
import io.searchbox.core.Search;
import io.searchbox.core.SearchResult;
import io.searchbox.core.SearchResult.Hit;
import io.searchbox.core.search.aggregation.MetricAggregation;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.search.aggregations.AggregationBuilders;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.elasticsearch.search.sort.SortOrder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;/**
 * @author yanlei
 * @version $Id: JestClientDemo.java, v 0.1 2018年05月23日 下午5:45 yanlei Exp $
 */
public class JestClientDemo {    private static final Logger LOGGER = LoggerFactory.getLogger(JestClientDemo.class);    private static final String INDEX_NAME = "pangu";    private static final String TYPE_NAME = "normal";    private static JestClient client;    static {
        JestClientFactory factory = new JestClientFactory();        // 定制化gson配置
        Gson gson = new GsonBuilder().setDateFormat("yyyyMMdd hh:mm:ss").create();        factory.setHttpClientConfig(new HttpClientConfig
                .Builder("http://search.alipay.com:9999") // 集群地址
                .defaultCredentials("example", "WnhmUwjU") // 认证信息
                .gson(gson) // 启用定制化gson，可使用默认
                .multiThreaded(true)
                .defaultMaxTotalConnectionPerRoute(2)
                .maxTotalConnection(10)
                .build());        client = factory.getObject();
    }    public boolean index(String id, Object obj) {
        Index index = new Index.Builder(obj).index(INDEX_NAME).type(TYPE_NAME).id(id).build();
        try {
            JestResult result = client.execute(index);
            if(result.isSucceeded()) {
                LOGGER.info("index success");
                return true;
            }else {
                LOGGER.error("index error : {}", result.getErrorMessage());
                return false;
            }
        } catch (IOException e) {
            LOGGER.error("系统异常", e);
            return false;
        }
    }    public boolean index(User user) {
        Index index = new Index.Builder(user).index(INDEX_NAME).type(TYPE_NAME).build();
        try {
            JestResult result = client.execute(index);
            if(result.isSucceeded()) {
                LOGGER.info("index success");
                return true;
            }else {
                LOGGER.error("index error : {}", result.getErrorMessage());
                return false;
            }
        } catch (IOException e) {
            LOGGER.error("系统异常", e);
            return false;
        }
    }    public <T> List<T> search(SearchSourceBuilder sourceBuilder, CallbackSearch<T> callbackSearch, Class<T> response) {
        Search search = new Search.Builder(sourceBuilder.toString())
                .addIndex(INDEX_NAME)
                .addType(TYPE_NAME)
                .build();        SearchResult result = null;
        try {
            result = client.execute(search);
            if(result.isSucceeded()) {
                // 提取数据
                List<Hit<T, Void>> hits = result.getHits(response);
                return callbackSearch.getHits(hits);
            }else {
                LOGGER.error("index false, error : {}", result.getErrorMessage());
                return null;
            }
        } catch (IOException e) {
            LOGGER.error("系统异常", e);
            return null;
        }
    }    public <T> T aggregation(SearchSourceBuilder sourceBuilder, CallbackAggregation<T> callbackAggregation) {
        Search search = new Search.Builder(sourceBuilder.toString())
                .addIndex(INDEX_NAME)
                .addType(TYPE_NAME)
                .build();        SearchResult result = null;
        try {
            result = client.execute(search);
            if(result.isSucceeded()) {
                return callbackAggregation.getAgg(result.getAggregations());
            }else {
                LOGGER.error("index false, error : {}", result.getErrorMessage());
                return null;
            }
        } catch (IOException e) {
            LOGGER.error("系统异常", e);
            return null;
        }
    }    public interface CallbackSearch<T> {
        List<T> getHits(List<Hit<T, Void>> hits);
    }    public interface CallbackAggregation<T> {
        T getAgg(MetricAggregation metricAggregation);
    }    public static void main(String[] args) {
        JestClientDemo demo = new JestClientDemo();        // 1. 索引数据
        User user = new User();
        user.setId("2088201805281205");
        user.setName("nick");
        user.setAge(18);
        user.setCountry("CHINA");
        boolean indexOK = demo.index(user);
        LOGGER.info("index param={} result={}", user, indexOK);        // 2. 数据检索
        SearchSourceBuilder sourceBuilder = new SearchSourceBuilder()
                .query(QueryBuilders.boolQuery().must(QueryBuilders.termQuery("name", "nick")))
                .sort("id", SortOrder.DESC)
                .from(0).size(10);
        List<User> searchResult = demo.search(sourceBuilder, new CallbackSearch<User>() {
            @Override
            public List<User> getHits(List<Hit<User, Void>> hits) {
                List<User> userList = new ArrayList<>();
                hits.forEach(hit -> {
                    userList.add(hit.source);
                });
                return userList;
            }
        }, User.class);
        LOGGER.info("search param={} result={}", sourceBuilder, searchResult);        // 3. 聚合查询
        SearchSourceBuilder aggSearchSourceBuilder = new SearchSourceBuilder()
                .query(QueryBuilders.matchAllQuery())
                .aggregation(AggregationBuilders.avg("age_avg").field("age"));
        double aggResult = demo.aggregation(aggSearchSourceBuilder, metricAggregation -> {
            double avgAge = metricAggregation.getAvgAggregation("age_avg").getAvg();
            return avgAge;
        });
        LOGGER.info("aggregation param={} result={}", aggSearchSourceBuilder, aggResult);
    }
}

小结

JestClient兼容性由于其他两者，Java High Level REST Client设置了默认调优参数，若版本匹配，其性能会更加优秀。Java Rest Cilent随Java High Level REST Client推出，更名为Java Low Level REST Client了，官网也不推荐使用了。最后补充下，RestClient虽然基于ResulFul风格的请求，但是构建QueryDSL还是一件成本比较高的事情，所以Java High Level REST Client和Jest都使用ElasticSearch的APIclient构建QueryDSL，其中的Builder模式在在两个客户端中都有比较好的实现，值得学习研究。

点赞 851

贡献者

上一篇： Elasticsearch Java Rest Client API 整理总结 (一)——Document API
下一篇： java web 获取客户端操作系统信息
相关推荐
python开发_常用的python模块及安装方法
python开发_常用的python模块及安装方法
adodb：我们领导推荐的数据库连接组件bsddb3：BerkeleyDB的连接组件Cheetah-1.0：我比较喜欢这个版本的cheeta…
技术
日期：2022-11-24 点赞：878 阅读：9,023
Educational Codeforces Round 11 C. Hard Process 二分
Educational Codeforces Round 11 C. Hard Process 二分
C. Hard Process题目连接：http://www.codeforces.com/contest/660/problem/CDes…
技术
日期：2022-11-24 点赞：807 阅读：5,513
下载Ubuntn 17.04 内核源代码
下载Ubuntn 17.04 内核源代码
zengkefu@server1:/usr/src$ uname -aLinux server1 4.10.0-19-generic #21…
技术
日期：2022-11-24 点赞：569 阅读：6,361
可用Active Desktop Calendar V7.86 注册码序列号
可用Active Desktop Calendar V7.86 注册码序列号
可用Active Desktop Calendar V7.86 注册码序列号Name: www.greendown.cn Code: &nb…
技术
日期：2022-11-24 点赞：733 阅读：6,143
Android调用系统相机、自定义相机、处理大图片
Android调用系统相机、自定义相机、处理大图片
Android调用系统相机和自定义相机实例本博文主要是介绍了android上使用相机进行拍照并显示的两种方式，并且由于涉及到要把拍到的照片显…
技术
日期：2022-11-24 点赞：512 阅读：7,774
Struts的使用
Struts的使用
一、Struts2的获取　　Struts的官方网站为：http://struts.apache.org/　　下载完Struts2的jar包,…
技术
日期：2022-11-24 点赞：671 阅读：4,853
Lomu

贡献者

这家伙很懒，只想把你留下。

文章
58
收藏
0
人气
189,756
粉丝
进主页
文章展示
  • [ 技术 ] python开发_常用的python模块及安装方法
  • [ 技术 ] Educational Codeforces Round 11 C. Hard Process 二分
  • [ 技术 ] 下载Ubuntn 17.04 内核源代码
  • [ 技术 ] 可用Active Desktop Calendar V7.86 注册码序列号
  • [ 技术 ] Android调用系统相机、自定义相机、处理大图片
  • [ 技术 ] Struts的使用
    • 个人收藏笔记记录
    我的笔记
    首页 首页 菜单 首页 首页