首页 > 其他分享> > 14 Rest客户端实践（一）D低级客户端入门

14 Rest客户端实践（一）D低级客户端入门

2021-01-08 14:57:45 作者：互联网

文章目录

1 环境准备

1.1 依赖

依赖

<dependencies>
    <dependency>
        <groupId>org.elasticsearch.client</groupId>
        <artifactId>elasticsearch-rest-client</artifactId>
        <version>7.10.1</version>
    </dependency>
    <dependency>
    	<!--高级客户端-->
        <groupId>org.elasticsearch.client</groupId>
        <artifactId>elasticsearch-rest-high-level-client</artifactId>
        <version>7.10.1</version>
    </dependency>
    <dependency>
        <groupId>ch.qos.logback</groupId>
        <artifactId>logback-classic</artifactId>
        <version>1.2.3</version>
    </dependency>
    <dependency>
        <groupId>org.projectlombok</groupId>
        <artifactId>lombok</artifactId>
        <version>1.18.16</version>
    </dependency>
    <dependency>
        <groupId>junit</groupId>
        <artifactId>junit</artifactId>
        <version>4.13.1</version>
    </dependency>
 </dependencies>

1.2 索引库准备

PUT books
{
  "settings": {
    "number_of_shards": 3, 
    "number_of_replicas": 1
  }, 
  "mappings": {
      "properties": {
        "id":{
          "type": "long"
        },
        "title":{
          "type": "text",
          "analyzer": "ik_max_word"
        },
        "language":{
          "type": "keyword"
        },
        "author":{
          "type": "keyword"
        },
        "price":{
          "type": "double"
        },
        "year":{
          "type": "date",
          "format": "yyyy-MM-dd"
        },
        "description":{
          "type": "text",
          "analyzer": "ik_max_word"
        }
      }
    }
  
}

2 客户端初始化

2.1 基本用法

只需要设置es服务的ip和端口，以及通讯协议即可

 @Test
 public void testBuild() throws IOException {
     // 1 构造RestClient
     RestClient client = RestClient.builder(
             new HttpHost("localhost", 9200, "http")
     ).build();
     // 2 关闭
     client.close();
 }

public static RestClientBuilder builder(HttpHost... hosts) {
     if (hosts == null || hosts.length == 0) {
         throw new IllegalArgumentException("hosts must not be null nor empty");
     }
     List<Node> nodes = Arrays.stream(hosts).map(Node::new).collect(Collectors.toList());
     return new RestClientBuilder(nodes);
 }

可见这个方法接受的是一个可变参数，所以如果是集群，可以传HttpHost对象的数组即可

2.2 可配置项目

配置请求头

/**
 *  配置请求头
 */
public void testConfigHead() throws IOException {
    Header[] hears = new Header[]{new BasicHeader("key","value")} ;
    // 接受的是一个Header数组
    RestClient client = restClientBuilder.setDefaultHeaders(hears).build();
    client.close();
}

配置失败监听器

 /****
     * 配置监听器
     */
    @Test
    public void testConfigListener() throws IOException {
        // 设置一个监听器，该监听器在每次节点失败时都会收到通知，在启用嗅探时在内部使用。
        RestClient client = restClientBuilder.setFailureListener(new RestClient.FailureListener() {
            @Override
            public void onFailure(Node node) {
                super.onFailure(node);
            }
        }).build();
        client.close();
    }

配置节点选择器

/****
* 设置节点选择器
*/
@Test
public void testConfigNodeSelector() throws IOException {
   RestClient client = restClientBuilder.setNodeSelector(NodeSelector.SKIP_DEDICATED_MASTERS)
           .build();
   client.close();
}

配置超时时间

/****
  * 配置超时时间
  */
 @Test
 public void testConfigTimeout() throws IOException {
     RestClient client = restClientBuilder.setRequestConfigCallback(new RestClientBuilder.RequestConfigCallback() {
         public RequestConfig.Builder customizeRequestConfig(RequestConfig.Builder requestConfigBuilder) {
             // 设置套接字超时时间
             requestConfigBuilder.setSocketTimeout(10000);
             // 链接超时时间
             requestConfigBuilder.setConnectTimeout(5000);
             return requestConfigBuilder;
         }
     }).build();
     client.close();
 }

配置线程数

/****
  * 配置线程数量
  * @throws IOException
  */
 @Test
 public void testConfigThread() throws IOException {
     RestClient client= restClientBuilder.setHttpClientConfigCallback(new RestClientBuilder.HttpClientConfigCallback() {
         @Override
         public HttpAsyncClientBuilder customizeHttpClient(HttpAsyncClientBuilder httpClientBuilder) {
            httpClientBuilder.setDefaultIOReactorConfig(IOReactorConfig.custom().setIoThreadCount(2).build());
             return httpClientBuilder;
         }
     }).build();
     client.close();
 }

完整的测试样例

package study.wyy.es.client.base;

import lombok.extern.slf4j.Slf4j;
import org.apache.http.Header;
import org.apache.http.HttpHost;
import org.apache.http.client.config.RequestConfig;
import org.apache.http.impl.nio.client.HttpAsyncClientBuilder;
import org.apache.http.impl.nio.reactor.IOReactorConfig;
import org.apache.http.message.BasicHeader;
import org.elasticsearch.client.Node;
import org.elasticsearch.client.NodeSelector;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestClientBuilder;
import org.junit.Before;
import org.junit.Test;

import java.io.IOException;

/**
 * @author by wyaoyao
 * @Description
 * @Date 2021/1/3 5:20 下午
 */
@Slf4j
public class RestClientBuilderTest {

    RestClientBuilder restClientBuilder;

    @Before
    public void initHost(){
        restClientBuilder = RestClient.builder(
                new HttpHost("localhost", 9200, "http")
        );

    }
    @Test
    public void hello(){
        log.info("hello world");
    }

    /**
     *  构造简单的es RestClient
     */
    @Test
    public void testBuild() throws IOException {
        // 1 构造RestClient
        RestClient client = RestClient.builder(
                new HttpHost("localhost", 9200, "http")
        ).build();
        // 2 关闭
        client.close();
    }

    /**
     *  配置请求头
     */
    public void testConfigHead() throws IOException {
        Header[] hears = new Header[]{new BasicHeader("key","value")} ;
        // 接受的是一个Header数组
        RestClient client = restClientBuilder.setDefaultHeaders(hears).build();
        client.close();
    }

    /****
     * 配置监听器
     */
    @Test
    public void testConfigListener() throws IOException {
        // 设置一个监听器，该监听器在每次节点失败时都会收到通知，在启用嗅探时在内部使用。
        RestClient client = restClientBuilder.setFailureListener(new RestClient.FailureListener() {
            @Override
            public void onFailure(Node node) {
                super.onFailure(node);
            }
        }).build();
        client.close();
    }

    /****
     * 设置节点选择器
     */
    @Test
    public void testConfigNodeSelector() throws IOException {
        RestClient client = restClientBuilder.setNodeSelector(NodeSelector.SKIP_DEDICATED_MASTERS)
                .build();
        client.close();
    }

    /****
     * 配置超时时间
     */
    @Test
    public void testConfigTimeout() throws IOException {
        RestClient client = restClientBuilder.setRequestConfigCallback(new RestClientBuilder.RequestConfigCallback() {
            public RequestConfig.Builder customizeRequestConfig(RequestConfig.Builder requestConfigBuilder) {
                // 设置套接字超时时间
                requestConfigBuilder.setSocketTimeout(10000);
                // 链接超时时间
                requestConfigBuilder.setConnectTimeout(5000);
                return requestConfigBuilder;
            }
        }).build();
        client.close();
    }

    /****
     * 配置线程数量
     * @throws IOException
     */
    @Test
    public void testConfigThread() throws IOException {
        RestClient client= restClientBuilder.setHttpClientConfigCallback(new RestClientBuilder.HttpClientConfigCallback() {
            @Override
            public HttpAsyncClientBuilder customizeHttpClient(HttpAsyncClientBuilder httpClientBuilder) {
               httpClientBuilder.setDefaultIOReactorConfig(IOReactorConfig.custom().setIoThreadCount(2).build());
                return httpClientBuilder;
            }
        }).build();
        client.close();
    }
}

3 构建请求

@Test
public void testBuildRequest() throws IOException {
    // 构建一个请求：localhost:9200/
    Request request = new Request("GET", "/");
    Response response = restClient.performRequest(request); 
    log.info("response: {}", EntityUtils.toString(response.getEntity()));
    restClient.close();
}

通过org.elasticsearch.client.Request进行构建请求参数的
- 参数一：就是请求方法，GET，POST等
- 参数二：就是请求的endPoint

关于响应结果的解析，后面在介绍。这里就是简单看一下日志的输出

{
  "name" : "01DR11620116651",
  "cluster_name" : "elasticsearch",
  "cluster_uuid" : "tWcocvlyQxWfNFhIshS5Dw",
  "version" : {
    "number" : "7.10.1",
    "build_flavor" : "default",
    "build_type" : "zip",
    "build_hash" : "1c34507e66d7db1211f66f3513706fdf548736aa",
    "build_date" : "2020-12-05T01:00:33.671820Z",
    "build_snapshot" : false,
    "lucene_version" : "8.7.0",
    "minimum_wire_compatibility_version" : "6.8.0",
    "minimum_index_compatibility_version" : "6.0.0-beta1"
  },
  "tagline" : "You Know, for Search"
}

再比如查询我们上面准备的索引中_id=1的文档，注意下，这时索引库中一条数据还没有

@Test
public void testBuildRequest3() throws IOException {
    // 构建一个请求
    Request request = new Request("GET", "books/_doc/1");
    Response response = restClient.performRequest(request);
    log.info("response: {}", EntityUtils.toString(response.getEntity()));
    restClient.close();
}

执行发现404：

org.elasticsearch.client.ResponseException: method [GET], host [http://localhost:9200], URI [books/_doc/1], status line [HTTP/1.1 404 Not Found]
{"_index":"books","_type":"_doc","_id":"1","found":false}

但是是有结果的，日志里面已经看到：这不就是kibana上执行查询的结果


可点击key和value值进行编辑
{
    "_index":"books",
    "_type":"_doc",
    "_id":"1",
    "found":false
}

那么去kibana新增一条数据之后，在测试呢：

PUT books/_doc/1
{
    "id":"1",
    "title":"Java编程思想",
    "language":"java",
    "author":"Bruce Eckel",
    "price":70.2,
    "publish_time":"2007-10-01",
    "description":"Java学习必读经典,殿堂级著作！赢得了全球程序员的广泛赞誉。"
}

此时就会发现不会出现404，测试不会报错了。

完整的测试代码

package study.wyy.es.client.base;

import lombok.extern.slf4j.Slf4j;
import org.apache.http.HttpHost;
import org.apache.http.util.EntityUtils;
import org.elasticsearch.client.Request;
import org.elasticsearch.client.Response;
import org.elasticsearch.client.RestClient;
import org.junit.Before;
import org.junit.Test;

import java.io.IOException;

/**
 * @author wyaoyao
 * @description
 * @date 2021/1/8 10:47
 */
@Slf4j
public class BuildRequestTest {

    RestClient restClient;

    @Before
    public void buildRestClient(){
        restClient = RestClient.builder(new HttpHost("localhost", 9200, "http"))
                .build();
    }

    /***
     * 构建请求对象
     */
    @Test
    public void testBuildRequest() throws IOException {
        // 构建一个请求
        Request request = new Request("GET", "/");
        Response response = restClient.performRequest(request);
        log.info("response: {}", EntityUtils.toString(response.getEntity()));
        restClient.close();
    }

    @Test
    public void testBuildRequest2() throws IOException {
        // 构建一个请求
        Request request = new Request("GET", "_cat/indices");
        Response response = restClient.performRequest(request);
        log.info("response: {}", EntityUtils.toString(response.getEntity()));
        restClient.close();
    }

    @Test
    public void testBuildRequest3() throws IOException {
        // 构建一个请求
        Request request = new Request("GET", "books/_doc/1");
        Response response = restClient.performRequest(request);
        log.info("response: {}", EntityUtils.toString(response.getEntity()));
        restClient.close();
    }
}

4 执行请求

es的客户端提供了两种执行方式

同步执行
异步执行

4.1 同步执行

同步执行：restClient.performRequest()
前面的代码已经演示如何使用了，这里不多赘述

4.2 异步执行

当异步执行，客户端不需要等es服务返回响应。为了能够异步处理，所以需要一个监听器，来监听es服务的响应，进行相应的处理

API介绍

调用的方法的就是restClient.performRequestAsync()，接收两个参数
- 请求对象Request
- 监听器”：ResponseListener
org.elasticsearch.client.ResponseListener

public interface ResponseListener {

    /**
     * Method invoked if the request yielded a successful response
     * // 执行成功的回调函数
     */
    void onSuccess(Response response);

    /**
     *  // 执行异常的回调函数
     */
    void onFailure(Exception exception);
}

示例

package study.wyy.es.client.base;

import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.apache.http.HttpHost;
import org.apache.http.util.EntityUtils;
import org.elasticsearch.client.Request;
import org.elasticsearch.client.Response;
import org.elasticsearch.client.ResponseListener;
import org.elasticsearch.client.RestClient;
import org.junit.Before;
import org.junit.Test;

import java.io.IOException;

/**
 * @author wyaoyao
 * @description
 * @date 2021/1/8 13:49
 */
@Slf4j
public class PerformRequestTest {

    /***
     * 测试异步
     * @throws IOException
     */
    public static void main(String[] args) {
        RestClient restClient = RestClient.builder(new HttpHost("localhost", 9200, "http"))
                .build();
        // 构建一个请求
        Request request = new Request("GET", "/");
        restClient.performRequestAsync(request, new ResponseListener() {
            // 执行成功的回调函数
            public void onSuccess(Response response) {
                log.info("执行成功");
            }
            // 执行异常的回调函数
            public void onFailure(Exception exception) {
                log.error("执行异常:{}",exception.getMessage(),exception);
            }
        });
        // 不能在这关闭了，异步请求，可能还没有拿到结果，客户端关闭了
        //restClient.close();
    }
}

5 响应解析

package study.wyy.es.client.base;

import lombok.extern.slf4j.Slf4j;
import org.apache.http.Header;
import org.apache.http.HttpHost;
import org.apache.http.RequestLine;
import org.apache.http.util.EntityUtils;
import org.elasticsearch.client.Request;
import org.elasticsearch.client.Response;
import org.elasticsearch.client.RestClient;
import org.junit.Before;
import org.junit.Test;

import java.io.IOException;

/**
 * @author wyaoyao
 * @description
 * @date 2021/1/8 14:29
 */
@Slf4j
public class ParseResponseTest {
    RestClient restClient;

    @Before
    public void buildRestClient(){
        restClient = RestClient.builder(new HttpHost("localhost", 9200, "http"))
                .build();
    }


    @Test
    public void parseResponseTest() throws IOException {
        // 构建一个请求
        Request request = new Request("GET", "books/_doc/1");
        Response response = restClient.performRequest(request);
        // 拿到请求信息
        RequestLine requestLine = response.getRequestLine();
        log.info("请求信息: {}",requestLine);
        // 拿到host地址
        HttpHost host = response.getHost();
        log.info("es地址：{}",host);
        // 响应头
        Header[] headers = response.getHeaders();
        log.info("响应头：{}",headers);
        String contentType = response.getHeader("content-type");
        log.info("contentType：{}",contentType);
        // 状态码
        int statusCode = response.getStatusLine().getStatusCode();
        log.info("状态码：{}",statusCode);
        // 响应体
        String body = EntityUtils.toString(response.getEntity());
        log.info("响应体：{}",body);

    }

}

请求信息: GET books/_doc/1 HTTP/1.1
es地址：http://localhost:9200
响应头：content-type: application/json; charset=UTF-8
contentType：application/json; charset=UTF-8
 状态码：200
 响应体：{"_index":"books","_type":"_doc","_id":"1","_version":1,"_seq_no":4,"_primary_term":1,"found":true,"_source":{
    "id":"1",
    "title":"Java编程思想",
    "language":"java",
    "author":"Bruce Eckel",
    "price":70.2,
    "publish_time":"2007-10-01",
    "description":"Java学习必读经典,殿堂级著作！赢得了全球程序员的广泛赞誉。"
}
}

标签：RestClient,14,void,Rest,client,org,import,public,客户端
来源： https://blog.csdn.net/wyaoyao93/article/details/112345838