17、全文检索 -- Elasticsearch -- 使用 反应式 RestClient （ReactiveElasticsearchClient）操作 Es 服务器（增、删、查 ：索引库和文档）

使用反应式RestClient （ReactiveElasticsearchClient） 操作 Elasticsearch 服务器（增、删、查 索引库和文档）

反应式 RestClient

Elasticsearch 所提供 RestHighLevelClient 本身提供了 【同步编程】 和 【异步编程】两种模型。

Elasticsearch 官方并未提供反应式的 RestClient :

因此 Spring Data Elasticsearch 额外补充了一个 ReactiveElasticsearchClient，用于提供反应式API支持，

ReactiveElasticsearchClient 相当于 RestHighLevelClient 的反应式版本，因此它们二者的功能基本相似。

ReactiveElasticsearchClient 是基于 WebFlux 的 WebClient 的 ，因此如果要使用反应式的 RestClient，还需要添加 Spring WebFlux 依赖。

反应式 RestClient 的方法

ReactiveElasticsearchClient 的方法与 RestHighLevelClient 的方法大同小异，区别只是 ReactiveElasticsearchClient 的方法所返回的是Mono 或 Flux。

处理配置信息及对 反应式 RestClient 进行定制

所有以 spring.data.elasticsearch.client.reactive.* 开头的属性由 ReactiveElasticsearchRestClientProperties 类负责加载，并由容器中自动配置的 ClientConfiguration 负责处理。

如果觉得上面配置属性还不够，或希望完全控制 ReactiveElasticsearchClient 的配置，则可在容器中配置一个自定义的 ClientConfiguration ， 这样 Spring Boot 将不再提供自动配置的 ClientConfiguration。

这样一来，Spring Boot 在自动配置 ReactiveElasticsearchClient 时，就会改为使用自定义的 ClientConfiguration 所提供的配置信息，完全忽略以 spring.data.elasticsearch.client.reactive.* 开头的配置信息。

 ReactiveElasticsearchClient
          ↑
   ClientConfiguration （该Bean可以被替换，这意味着使用自己的 ClientConfiguration 来管理与 Elasticsearch 的反应式API连接）
          ↓ 
 处理 spring.data.elasticsearch.client.reactive.*属性
 ( 如果用自己自定义的ClientConfiguration ，那么就可以自己决定要不要读取这个配置文件的这个开头的属性）
 ( 如果用 springboot 自身的 ClientConfiguration ，则会读取这个配置文件的这个开头的属性）
1
2
3
4
5
6
7

代码演示：

1、创建项目

2、添加依赖

        <!--  为 Elasticsearch 添加 反应式API  -->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-data-elasticsearch</artifactId>
        </dependency>

        <!-- 反应式的依赖 -->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-webflux</artifactId>
        </dependency>
1
2
3
4
5
6
7
8
9
10
11

3、配置文件

上面配置中 spring.data.elasticsearch.client.reactive.use-ssl 属性为 false，这说明这次演示不使用SSL，因此需要关闭 Elasticsearch 的SSL。

配置反应式 RestClient 所用的配置属性与配置 RestHighLevelClient 的配置属性完全不同的，且属性处理类也不相同。
此处用的属性处理类是：ReactiveElasticsearchRestClientProperties

4、测试类 演示 反应式 操作Elasticsearch服务器

这篇是正常的web项目：使用 RESTful 客户端（RestHighLevelClient ）操作 Elasticsearch，测试方法是一样的，只不过现在改成了反应式编程，可以根据这篇文章做对比。

1、添加索引库

代码

测试结果

查看所有索引库：http://localhost:9200/_cat/indices

2、删除索引库

代码

测试结果

查看所有索引库：http://localhost:9200/_cat/indices
“items”, “users” 索引库已经被删除了

3、查询所有索引库

代码

测试结果

查看所有索引库：http://localhost:9200/_cat/indices
结果跟 postman 一样，表示代码查询索引库的数据正确

4、往索引库添加文档

代码

测试结果

查询指定 index 的全部文档：http://localhost:9200/books/_search

5、根据文档的 id 获取文档

代码

测试结果

成功

6、根据关键字和通配符查询文档

代码

测试结果

成功

7、根据文档的 id 删除文档

代码

测试结果

查询指定 index 的全部文档：http://localhost:9200/books/_search

如图：id 为 3 和 4 的文档被成功删除了

完整代码

application.properties 配置文件

# 配置Elasticsearch服务器的地址
spring.data.elasticsearch.client.reactive.endpoints=127.0.0.1:9200
# 配置不使用SSL
spring.data.elasticsearch.client.reactive.use-ssl=false
# 连接超时时间
spring.data.elasticsearch.client.reactive.socket-timeout=10s
# 配置连接Elasticsearch服务器的用户名、密码
spring.data.elasticsearch.client.reactive.username=elastic
spring.data.elasticsearch.client.reactive.password=123456

1
2
3
4
5
6
7
8
9
10
11

ReactiveclientElasticsearchTest 测试类

package cn.ljh.reactiveclient;

import lombok.SneakyThrows;
import org.elasticsearch.action.admin.indices.delete.DeleteIndexRequest;
import org.elasticsearch.action.delete.DeleteRequest;
import org.elasticsearch.action.delete.DeleteResponse;
import org.elasticsearch.action.get.GetRequest;
import org.elasticsearch.action.get.GetResponse;
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.action.support.master.AcknowledgedResponse;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.indices.CreateIndexRequest;
import org.elasticsearch.client.indices.CreateIndexResponse;
import org.elasticsearch.client.indices.GetIndexRequest;
import org.elasticsearch.client.indices.GetIndexResponse;
import org.elasticsearch.common.xcontent.XContentType;
import org.elasticsearch.index.get.GetResult;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.CsvSource;
import org.junit.jupiter.params.provider.ValueSource;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.data.elasticsearch.client.reactive.ReactiveElasticsearchClient;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import java.util.Arrays;

@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.NONE)
class ReactiveclientElasticsearchTest
{

    @Autowired
    private ReactiveElasticsearchClient reactiveEsClient;


    //以下方法皆是以反应式编程测试


    //反应式创建三个索引库
    @ParameterizedTest //参数测试
    @SneakyThrows
    @ValueSource(strings = {"books", "items", "users"})
    public void testCreateIndex(String indexName)
    {

        //指定分词器创建索引库的json格式的数据，每一行用双引号包起来，然后里面的每个双引号前面用反斜杠\转义
        String json = "{" +
                "\"settings\": {" +
                "    \"analysis\": {" +
                "      \"analyzer\": {" +
                "        \"default\": {" +
                "           \"tokenizer\": \"ik_max_word\"" +
                "        }" +
                "      }" +
                "    }" +
                "  }" +
                "}";

        CreateIndexRequest request = new CreateIndexRequest(indexName)
                //参数1：指定创建索引库时要传入的参数  ； 参数2：指定传入内容的类型
                .source(json, XContentType.JSON);

        //调用反应式方法，返回 Mono
        Mono<Boolean> resp = reactiveEsClient.indices().createIndex(request);

        //以同步的方式直接获取结果
        resp.blockOptional().ifPresent(System.err::println);

    }


    //删除索引库
    @SneakyThrows
    @ParameterizedTest
    @ValueSource(strings = {"items", "users"})
    public void testDeleteIndex(String indexName)
    {
        //删除索引的请求数据
        DeleteIndexRequest request = new DeleteIndexRequest(indexName);

        //客户端调用操作索引的方法，然后再调用删除的方法
        Mono<Boolean> resp = reactiveEsClient.indices().deleteIndex(request);

        //以同步的方式直接获取结果
        resp.blockOptional().ifPresent(System.err::println);
    }


    //查询所有的索引库
    @SneakyThrows
    @Test //这个测试不需要参数，直接用这个@Test注解即可
    public void testGetIndex()
    {
        //参数 "*" ： 表示匹配所有的索引库
        GetIndexRequest request = new GetIndexRequest("*");

        //用反应式的rest客户端的方法来查询
        Mono<GetIndexResponse> monoResp = reactiveEsClient.indices().getIndex(request);

        //以同步的方式直接获取结果
        monoResp.blockOptional().ifPresent(resp ->
        {
            //返回的索引库是一个String类型的数组
            String[] indices = resp.getIndices();
            //把数组转成字符串
            String s = Arrays.toString(indices);
            System.err.println(s);
        });

    }










    //往索引库添加文档

    @ParameterizedTest
    @SneakyThrows
    //测试参数有多个值 ，用这个注解
    @CsvSource({
            "1,火影忍者,旋涡鸣人成长为第七代火影的故事,150",
            "2,家庭教师,废材纲成长为十代首领的热血事迹,200",
            "3,七龙珠,赛亚人孙悟空来到地球后的热血升级之旅,300",
            "4,七龙珠Z,超级赛亚人贝吉塔来到地球后的热闹景象,400"
    })
    public void testSaveDocument(Integer id, String title, String description, Double price)
    {
        //表明向 books 索引库添加文档
        IndexRequest request = new IndexRequest("books")
                .id(id + "")
                .source(
                        "title", title,
                        "description", description,
                        "price", price
                );

        Mono<IndexResponse> resp = reactiveEsClient.index(request);

        //以同步的方式直接获取结果
        resp.blockOptional().ifPresent(System.err::println);

    }










    //根据文档的id获取文档
    @SneakyThrows
    @ParameterizedTest
    @ValueSource(ints = {1, 3})
    public void testGetDocumentById(Integer id)
    {
        //表明从 books 索引库获取文档
        GetRequest request = new GetRequest("books")
                //表明根据指定的文档的id获取文档
                .id(id + "");

        Mono<GetResult> resp = reactiveEsClient.get(request);

        //以同步的方式直接获取结果
        resp.blockOptional().ifPresent(System.err::println);

    }






    //根据条件查询文档（普通关键字查询和通配符查询）

    @SneakyThrows
    @ParameterizedTest
    @CsvSource({
            "description,热*",
            "description,成长"
    })
    public void testSearchDocument(String field, String term)
    {
        // 构建查询条件的类
        SearchSourceBuilder builder = new SearchSourceBuilder();

        // 通过 SearchSourceBuilder 可以用面向对象的方式来构建查询的 JSON 字符串
        // SearchSourceBuilder 需要传入 QueryBuilders，而 QueryBuilders 用于构建 QueryBuilder
        if (term != null && term.contains("*"))
        {
            //根据字段和通配符关键字查询
            builder.query(QueryBuilders.wildcardQuery(field, term));
        } else
        {
            //根据字段和普通关键字查询
            builder.query(QueryBuilders.matchQuery(field,term));
        }

        //表明从 books 索引库查询文档
        SearchRequest request = new SearchRequest("books")
                // 此处的 builder 参数用于构建查询语法
                .source(builder);

        //客户端调用查询的方法 ， 参数1：查询条件语法
        //Flux 表示返回了多个结果
        Flux<SearchHit> resp = reactiveEsClient.search(request);

        //以同步的方式获取Flux中的数据
        resp.toIterable().forEach(System.err::println);

    }








    //根据w文档的 id 删除文档

    @ParameterizedTest
    @SneakyThrows
    @ValueSource(ints = {3,4})
    public void testDeleteDocumentById(Integer id)
    {
        //表明从 books 索引库删除文档
        DeleteRequest request = new DeleteRequest("books")
                //获取指定id的文档
                .id(id+"");

        //rest客户端调用删除文档的方法
        Mono<DeleteResponse> resp = reactiveEsClient.delete(request);

        //以同步的方式直接获取结果
        resp.blockOptional().ifPresent(System.err::println);

    }


}

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
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258

pom.xml 依赖文件

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>
    <parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>2.5.3</version>
    </parent>

    <groupId>cn.ljh</groupId>
    <artifactId>reactiveclient</artifactId>
    <version>1.0.0</version>
    <name>reactiveclient</name>

    <properties>
        <java.version>11</java.version>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
    </properties>
    <dependencies>

        <!--  为 Elasticsearch 添加反应式API  -->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-data-elasticsearch</artifactId>
        </dependency>

        <!-- 反应式的依赖 -->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-webflux</artifactId>
        </dependency>


        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-devtools</artifactId>
            <scope>runtime</scope>
            <optional>true</optional>
        </dependency>
        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
            <optional>true</optional>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
        </dependency>
    </dependencies>

    <build>
        <plugins>
            <plugin>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-maven-plugin</artifactId>
                <configuration>
                    <excludes>
                        <exclude>
                            <groupId>org.projectlombok</groupId>
                            <artifactId>lombok</artifactId>
                        </exclude>
                    </excludes>
                </configuration>
            </plugin>
        </plugins>
    </build>

</project>

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