ElasticSearch 8.x 版本如何使用 SearchRequestBuilder 检索_elasticsearch8.x bulkrequest

ElasticSearch

1、ElasticSearch学习随笔之基础介绍
2、ElasticSearch学习随笔之简单操作
3、ElasticSearch学习随笔之java api 操作
4、ElasticSearch学习随笔之SpringBoot Starter 操作
5、ElasticSearch学习随笔之嵌套操作
6、ElasticSearch学习随笔之分词算法
7、ElasticSearch学习随笔之高级检索
8、ELK技术栈介绍
9、Logstash部署与使用
10、ElasticSearch 7.x 版本使用 BulkProcessor 实现批量添加数据
11、ElasticSearch 8.x 弃用了 High Level REST Client，移除了 Java Transport Client，推荐使用 Elasticsearch Java API
12、ElasticSearch 8.x 使用 snapshot（快照）进行数据迁移
13、ElasticSearch 8.x 版本如何使用 SearchRequestBuilder 检索
14、ElasticSearch 8.x 使用 High Level Client 以 HTTPS 方式链接，SSL 证书、主机名验证器 各是什么，如何忽略
15、ElasticSearch 8.x 创建父子文档，用Join类型字段以及用has_child、has_parent 检索

ElasticSearch，创始人 Shay Banon（谢巴农）
本文主要讲解ElasticSearch 高级搜索实战，来满足复杂的业务场景，还是用 Kibana 来操作。

---

---

前言

在之前的项目中都是用 SearchRequestBuilder 来构件 ElasticSearch 检索请求的，然后使用了新的 High Level Client 之后新的客户端构建查询是用了 SearchSourceBuilder 来构建检索请求的，若使使用新的检索方式请移步《Elasticsearch Java API 如何使用》查看。
那用 SearchSourceBuilder 进行构建查询时需要使用 RestHighLevelClient 客戶端来发送请求的，简单代码如下：

RestHighLevelClient client = createClient();
SearchRequest searchRequest = new SearchRequest();
SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();
sourceBuilder.query(QueryBuilders.matchAllQuery());
searchRequest.source(sourceBuilder);
sourceBuilder.from(0).size(10);
SearchResponse searchResponse = client.search(searchRequest, RequestOptions.DEFAULT);
1
2
3
4
5
6
7

以上这种写法就是 High Level Client 常见的写法。
那如果特殊情况下，比如需要和老项目融合的时候，可能需要使用到 SearchRequestBuilder 来检索，该如何做呢？
就是用 ElasticsearchClient 来包装 RestHighLevelClient 来执行查询，真正的检索使用 RestHighLevelClient 来执行的，而 ElasticsearchClient 则是构建为了融合并初始化 SearchRequestBuilder 而创建的。

一：pom 依赖

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

二：初始化客户端

2.1 初始化 HighLevelClient 客户端

这里首先要初始化 HighLevelClient 客户端，因为真正的检索需要使用此客户端来执行的。

/**
 * 初始化客户端
 * @return RestHighLevelClient
 */
private static RestHighLevelClient getClient(){
    final CredentialsProvider credentialsProvider = new BasicCredentialsProvider();
    credentialsProvider.setCredentials(AuthScope.ANY, new UsernamePasswordCredentials("Your username", "Your password"));
    RestClientBuilder restClientBuilder = RestClient.builder(new HttpHost("192.168.*.*", 9200))
            .setHttpClientConfigCallback(httpAsyncClientBuilder -> httpAsyncClientBuilder.setDefaultCredentialsProvider(credentialsProvider));
    RestHighLevelClient highLevelClient = new RestHighLevelClient(restClientBuilder);
    return highLevelClient;
}
1
2
3
4
5
6
7
8
9
10
11
12

2.2 初始化 ElasticsearchClient 客户端

这里我们来初始化 ElasticsearchClient 客户端，为什么要初始化此客户端呢？ElasticsearchClient 就像是一个代理客户端，调用了 HighLevelClient 来执行查询。
这里，我们把 RestHighLevelClient 作为参数传入，用于封装。

private static ElasticsearchClient getEsClient(RestHighLevelClient highLevelClient){
    return new ElasticsearchClient() {
        @Override
        public <Request extends ActionRequest, Response extends ActionResponse> ActionFuture<Response> execute(ActionType<Response> action, Request request) {
            if (action.equals(SearchAction.INSTANCE) && request instanceof SearchRequest) {
                return new ActionFuture<Response>() {
                    @Override
                    public Response actionGet() {
                        try {
                        	// 执行检索
                            return (Response) highLevelClient.search((SearchRequest) request, RequestOptions.DEFAULT);
                        } catch (IOException e) {
                            e.printStackTrace();
                        }
                        return null;
                    }

                    @Override
                    public Response actionGet(String timeout) {
                        try {
                            // 将字符串形式的时间转换为 TimeValue
                            TimeValue timeValue = TimeValue.parseTimeValue(timeout, "timeout");
                            SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();
                            sourceBuilder.timeout(timeValue);
                            SearchRequest searchRequest = new SearchRequest();
                            searchRequest.source(sourceBuilder);
                            // 执行检索
                            return (Response) highLevelClient.search((SearchRequest) request, RequestOptions.DEFAULT);
                        } catch (IOException e) {
                            throw new RuntimeException(e);
                        }
                    }

                    @Override
                    public Response actionGet(long timeoutMillis) {
                        try {
                            SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();
                            sourceBuilder.timeout(TimeValue.timeValueMillis(timeoutMillis));
                            SearchRequest searchRequest = new SearchRequest();
                            searchRequest.source(sourceBuilder);
                            // 执行检索
                            return (Response) highLevelClient.search((SearchRequest) request, RequestOptions.DEFAULT);
                        } catch (IOException e) {
                            throw new RuntimeException(e);
                        }
                    }

                    @Override
                    public Response actionGet(long timeout, TimeUnit unit) {
                        return null;
                    }

                    @Override
                    public Response actionGet(TimeValue timeout) {
                        return null;
                    }

                    @Override
                    public boolean cancel(boolean mayInterruptIfRunning) {
                        return false;
                    }

                    @Override
                    public boolean isCancelled() {
                        return false;
                    }

                    @Override
                    public boolean isDone() {
                        return false;
                    }

                    @Override
                    public Response get() throws InterruptedException, ExecutionException {
                        return null;
                    }

                    @Override
                    public Response get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException {
                        return null;
                    }
                };
            } else {
                throw new UnsupportedOperationException("Unsupported action: " + action.name());
            }
        }

        @Override
        public <Request extends ActionRequest, Response extends ActionResponse> void execute(ActionType<Response> action, Request request, ActionListener<Response> listener) {
            if (action.equals(SearchAction.INSTANCE) && request instanceof SearchRequest) {
                highLevelClient.searchAsync((SearchRequest) request, RequestOptions.DEFAULT, (ActionListener<SearchResponse>) listener);
            } else {
                throw new UnsupportedOperationException("Unsupported action: " + action.name());
            }
        }

        @Override
        public ThreadPool threadPool() {
            return null;
        }
    };
}
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
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102

以上初始化的 ElasticsearchClient 客户端中，只实现了常用的部分 actionGet 方法，其他方法在使用时再实现，实现方法基本上都差不多。

三：执行查询

ElasticsearchClient 则是可以通过 SearchRequestBuilder 来调用执行检索，在 new SearchRequestBuilder() 时，ElasticsearchClient 作为参数传递，则直接可以通过 execute().actionGet() 来执行检索并得到结果了。

public static void main(String[] args) {
    RestHighLevelClient highLevelClient = getClient();
    ElasticsearchClient esClient = getEsClient(highLevelClient);
    SearchRequestBuilder requestBuilder = new SearchRequestBuilder(esClient, SearchAction.INSTANCE);
    requestBuilder.setIndices("product_info");
    requestBuilder.setFrom(0);
    requestBuilder.setSize(10);
    requestBuilder.setQuery(QueryBuilders.matchAllQuery());
    requestBuilder.addAggregation(AggregationBuilders.terms("productName").field("productName.keyword"));
    // 执行检索并获取检索结果
    String s = requestBuilder.execute().actionGet().toString();
    System.out.println("检索结果：" + s);
    JSONObject jsonObject = JSONObject.parseObject(s);
    // 获取到 agg 统计结果
    JSONObject aggregations = jsonObject.getJSONObject("aggregations");
    System.out.println(aggregations);
    try {
        highLevelClient.close();
    } catch (IOException e) {
        e.printStackTrace();
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

执行结果：

检索结果：{"took":0,"timed_out":false,"_shards":{"total":1,"successful":1,"skipped":0,"failed":0},"hits":{"total":{"value":2,"relation":"eq"},"max_score":1.0,"hits":[{"_index":"product_info","_id":"1001","_score":1.0,"_source":{"productName":"小米汽车","productDescription":"相当于保时捷特斯拉的小米超跑","color":"海蓝色","price":19}},{"_index":"product_info","_id":"1002","_score":1.0,"_source":{"productName":"小米手机","productDescription":"小米，智能手机，价格实惠便宜","color":"黑色","price":2999}}]},"aggregations":{"sterms#productName":{"doc_count_error_upper_bound":0,"sum_other_doc_count":0,"buckets":[{"key":"小米手机","doc_count":1},{"key":"小米汽车","doc_count":1}]}}}
统计结果：{"sterms#productName":{"doc_count_error_upper_bound":0,"sum_other_doc_count":0,"buckets":[{"doc_count":1,"key":"小米手机"},{"doc_count":1,"key":"小米汽车"}]}}
1
2