Elasticsearch Java Rest Client 上手指南（上）

发表于 | | 阅读次数

开始看Elasticsearch Java API 的时候，被这段话浇了盆凉水

We plan on deprecating the TransportClient in Elasticsearch 7.0 and removing it completely in 8.0. Instead, you should be using the Java High Level REST Client, which executes HTTP requests rather than serialized Java requests. The migration guidedescribes all the steps needed to migrate.

The Java High Level REST Client currently has support for the more commonly used APIs, but there are a lot more that still need to be added. You can help us prioritise by telling us which missing APIs you need for your application by adding a comment to this issue: Java high-level REST client completeness.

Any missing APIs can always be implemented today by using the low level Java REST Client with JSON request and response bodies.

from：– https://www.elastic.co/guide/en/elasticsearch/client/java-api/current/java-api.html

大致意思是：我们马上就要不支持TransportClient 这个熊玩意了，你们应该用Java Rest Client，最好是还是高级别的client，用Http方式来访问ES，然后，麻烦帮我们确定这个Client的完整性，如果有发现高级Client缺失功能，可以用低版本的Clientlow level Java REST Client来发送json请求，查询数据

TransportClient这玩意儿确实很难用，抛开连接池不稳定不说，写完代码之后就是下图视感：

OMG

所以写完因为之后，准备用Java Rest Client来重构一下，用于兼容ES 5、6、7、甚至8

Java Low Level REST Client

1： 特性

  • 最小依赖
  • 针对可用节点，负载均衡
  • 针对错误和故障节点可以进行故障转移
  • 针对某个节点连接失败次数越多，客户端就会等待更长的时间，才会再次尝试这个节点
  • 持久连接
  • 请求日志追踪
  • 原子性操作

2： 依赖

至少java 1.7

maven

1
2
3
4
5
<dependency>
    <groupId>org.elasticsearch.client</groupId>
    <artifactId>elasticsearch-rest-client</artifactId>
    <version>6.3.1</version>
</dependency>

Gradle

1
2
3
dependencies {
    compile 'org.elasticsearch.client:elasticsearch-rest-client:6.3.1'
}

（具体版本号可以查询中央仓库:https://snapshots.elastic.co/maven/）

这个低级库依赖Apache Http Async Client发送http请求。包含了以下依赖

  • org.apache.httpcomponents:httpasyncclient
  • org.apache.httpcomponents:httpcore-nio
  • org.apache.httpcomponents:httpclient
  • org.apache.httpcomponents:httpcore
  • commons-codec:commons-codec
  • commons-logging:commons-logging

ps: 如果遇到打包冲突，可以参考官方maven-shade-plugin解决办法.

3： 初始化吧

直接按照官网来干

1
2
3
RestClient restClient = RestClient.builder(
        new HttpHost("localhost", 9200, "http"),
        new HttpHost("localhost", 9201, "http")).build();

带请求头式

1
2
3
RestClientBuilder builder = RestClient.builder(new HttpHost("localhost", 9200, "http"));
Header[] defaultHeaders = new Header[]{new BasicHeader("header", "value")};
builder.setDefaultHeaders(defaultHeaders);

带超时时间式(毫秒级)

1
2
RestClientBuilder builder = RestClient.builder(new HttpHost("localhost", 9200, "http"));
builder.setMaxRetryTimeoutMillis(10000);

带失败监听式

1
2
3
4
5
6
7
RestClientBuilder builder = RestClient.builder(new HttpHost("localhost", 9200, "http"));
builder.setFailureListener(new RestClient.FailureListener() {
    @Override
    public void onFailure(HttpHost host) {
       // TODO
    }
});

官方文档还提供了修改默认回调的方式 ,修改HttpClient默认配置的方式。

1
2
3
4
5
6
7
8
9
> RestClientBuilder builder = RestClient.builder(new HttpHost("localhost", 9200, "http"));
> builder.setRequestConfigCallback(new RestClientBuilder.RequestConfigCallback() {
>     @Override
>     public RequestConfig.Builder customizeRequestConfig(RequestConfig.Builder requestConfigBuilder) {
> // 请超时
>         return requestConfigBuilder.setSocketTimeout(10000); 
>     }
> });
>

1
2
3
4
5
6
7
8
> RestClientBuilder builder = RestClient.builder(new HttpHost("localhost", 9200, "http"));
> builder.setHttpClientConfigCallback(new RestClientBuilder.HttpClientConfigCallback() {
>     @Override
>     public HttpAsyncClientBuilder customizeHttpClient(HttpAsyncClientBuilder httpClientBuilder) {
>         return httpClientBuilder.setProxy(new HttpHost("proxy", 9000, "http"));
>     }
> });
>

关闭restClient

1
restClient.close();

4： 查询ES

这里写一个Demo级的查询语句，插入和更新与这个类似。只是请求方式换成了POST,PUT 删除换成了DELETE。大家稍微看一下就好，详细的Query语句可以先看看中文版的权威指南。

1
2
3
4
5
6
7
8
9
10
11
12
// Kibana查询语句翻译成 Rest Client低级版
GET code_flow_log*/_search
{
  "size": 20,
  "query": {
   "range": {
     "createTime": {
       "gte": "2018-06-01 00:00:00"
     }
   }
  }
}

java REST CLIENT版本

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
public final RestClient restClient = RestClient
        .builder(new HttpHost("172.18.90.40", 9200, "http"))
        .build();
@Test
public void search() throws IOException {
    Map<String, String> params = Collections.emptyMap();
    String queryString = "{" +
            "  \"size\": 20," +
            "  \"query\": {" +
            "   \"range\": {" +
            "     \"createTime\": {" +
            "       \"gte\": \"2018-06-01 00:00:00\"" +
            "     }" +
            "   }" +
            "  }" +
            "}";
    HttpEntity entity = new NStringEntity(queryString, ContentType.APPLICATION_JSON);
    try {
    Response response = restClient.performRequest("GET", "/some_important_index*/_search", params, entity);
    System.out.println(response.getStatusLine().getStatusCode());
    String responseBody = null;
    responseBody = EntityUtils.toString(response.getEntity());
    System.out.println("******************************************** ");
    JSONObject jsonObject = JSON.parseObject(responseBody);
    System.out.println(jsonObject.get("hits"));
    }catch (ResponseException e){
        e.printStackTrace();
    }
    System.out.println("23333");
    
}

5： 几个注意的点

  • 别用网上封装的工具类，尤其有一个里面全是static方法的，人家是DEOM级别的，不适合放在业务里，早日脱离 TransportClient,但是不用太激进，慢慢来
  • 这个Demo没有针对404错误做处理，需要的可在catch块里处理

  • 这个是同步请求的方式，restClient支持异步并对结果catch IOException和ResponseException ,方式如下

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    >        restClient.performRequestAsync("GET", "/code_flow_log*/_search", params, entity, new ResponseListener() {
    >            @Override
    >            public void onSuccess(Response response) {
    >                try {
    >                    JSONObject jsonObject = JSON.parseObject(EntityUtils.toString(response.getEntity()));
    >                    System.out.println(jsonObject.get("hits"));
    >                } catch (IOException e) {
    >                    e.printStackTrace();
    >                }
    >            }
    >
    >            @Override
    >            public void onFailure(Exception e) {
    >                e.printStackTrace();
    >            }
    >        });
    >

  • 低级版restClient不提供任何JSON处理工具，你自己根据自己喜好来就行，我用的是FastJSON

  • restClient支持设置连接池,默认调度线程是1个，连接线程跟处理器是一样的（取决于Runtime.getRuntime().availableProcessors()），自定义的方式代码如下

    1
    2
    3
    4
    5
    6
    7
    8
    9
    > RestClientBuilder builder = RestClient.builder(new HttpHost("localhost", 9200))
    >         .setHttpClientConfigCallback(new RestClientBuilder.HttpClientConfigCallback() {
    >            @Override
    >           public HttpAsyncClientBuilder customizeHttpClient(HttpAsyncClientBuilder httpClientBuilder) {
    >                 return httpClientBuilder.setDefaultIOReactorConfig(
    >                         IOReactorConfig.custom().setIoThreadCount(1).build());
    >             }
    >         });
    >

  • 对于设置了登录密码，SSL加密处理的ES，请自行查看文档：https://www.elastic.co/guide/en/elasticsearch/client/java-rest/current/_basic_authentication.html

6： Sniffer

用于自动发现Elasticsearch的节点，并设置为restClient实例，初始化后大约5分钟就会从ES中获取最新的节点列表，支持自定义时间，要支持失败时嗅探ES集群的节点，失败嗅探需要在每个RestClient中的Sniffer中添加一个SniffOnFailureListener。

by: MAX Zing blog:https://micorochio.github.io
转载请注明出处：https://micorochio.github.io/2018/07/22/elasticsearch_rest_low_level_client/
文章若有错误，请不吝指正。
下篇 High Level Rest Client的介绍：https://www.jianshu.com/p/d2c8326e8fa3