Elasticsearch Java Rest Client简述

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版本)。

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模式在在两个客户端中都有比较好的实现，值得学习研究。