[elastic 8.x]java客户端连接elasticsearch与操作索引与文档

初始化客户端

引入相关依赖

<dependency>
    <groupId>co.elastic.clientsgroupId>
    <artifactId>elasticsearch-javaartifactId>
    <version>8.10.2version>
dependency>
1
2
3
4
5

初始化客户端

为了方便演示，我关闭了elasticsearch的安全验证，带安全验证的初始化方式将在最后专门介绍

String serverUrl="http://127.0.0.1:9200";
RestClient restClient=RestClient
                .builder(HttpHost
                .create(serverUrl))
                .build();
ElasticsearchTransport transport=new RestClientTransport(restClient,new JacksonJsonpMapper());
ElasticsearchClient esClient=new ElasticsearchClient(transport);
1
2
3
4
5
6
7

索引

创建索引

    void createIndex(){

        String mappings = "{\n" +
                "  \"properties\" : {\n" +
                "    \"id\" : {\n" +
                "      \"type\" : \"keyword\" \n" +
                "    },\n"+
                "    \"name\" : {\n" +
                "      \"type\" : \"text\",\n" +
                "      \"fields\" : {\n" +
                "        \"keyword\" : {\n" +
                "          \"type\" : \"keyword\",\n" +
                "          \"ignore_above\" : 256 \n" +
                "        }\n" +
                "      } \n" +
                "    }, \n" +
                "    \"price\" : { \n" +
                "      \"type\" : \"long\" \n" +
                "     } \n" +
                "  }\n" +
                "}\n";
        JsonpMapper mapper=esClient._transport().jsonpMapper();
        JsonParser parser= Json.createParser(new StringReader(mappings));
        CreateIndexRequest createIndexRequest=new CreateIndexRequest.Builder().index("test")
                .mappings(TypeMapping._DESERIALIZER.deserialize(parser,mapper)).build();
        try {
            esClient.indices().create(createIndexRequest);
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }
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

删除索引

    void deleteMapping(){
        try {
            DeleteIndexResponse response;
            response= esClient.indices().delete(deleteIndexRequest->deleteIndexRequest.index("test"));
            System.out.println(response.toString());
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }
1
2
3
4
5
6
7
8
9

判断索引是否存在

    
    void existsIndex(){
        try {
            BooleanResponse hotel = esClient.indices().exists(existsIndexRequest -> existsIndexRequest.index("hotel"));
            System.out.println(hotel.value());
        } catch (IOException e) {
            throw new RuntimeException(e);
        }

    }
1
2
3
4
5
6
7
8
9
10

文档

新增文档

    void insertDoc(){
        Hotel hotel=hotelService.getById(61083L);
        HotelDoc hotelDoc=new HotelDoc(hotel);
        IndexRequest<HotelDoc> request=new IndexRequest.Builder<HotelDoc>()
                .id("11")
                .index("hotel")
                .document(hotelDoc)
                .build();
        try {
            IndexResponse index = esClient.index(request);
            System.out.println(index.id());
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

其中，HotelDoc是一个实体类

删除文档

    void deleteDoc(){
        try {
            esClient.delete(deleteRequest->deleteRequest.index("hotel").id("11"));
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }
1
2
3
4
5
6
7

查询文档

    void searchDoc(){
        TermQuery termQuery= QueryBuilders.term()
                .field("_id")
                .value("11")
                .build();
        SearchRequest request=new SearchRequest.Builder()
                        .index("hotel")
                        .query(termQuery._toQuery()).build();
        try {
            SearchResponse<HotelDoc> response=esClient.search(request,HotelDoc.class);
            
            //输出结果
            for(Hit<HotelDoc> hit:response.hits().hits()){
                System.out.println(hit.source());
            }
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

更新文档

    void updateDoc(){
        HotelDoc hotelDoc=new HotelDoc();
        //需要更新哪个字段就赋值哪个字段
        hotelDoc.setCity("xx");
        try {
            esClient.update(updateRequest->updateRequest
                            .index("hotel")
                            .id("11")
                            .doc(hotelDoc)
                    ,HotelDoc.class);
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14

批量导入文档

    void insertMany(){
        List<Hotel> hotels=hotelService.list();
        List<HotelDoc> hotelDocs=hotels.stream().map(HotelDoc::new).collect(Collectors.toList());
        BulkRequest.Builder bl=new BulkRequest.Builder();
        for(HotelDoc hotelDoc:hotelDocs){
            bl.operations(op->op.index(
                    idx->idx.index("hotel")
                            .id(hotelDoc.getId().toString())
                            .document(hotelDoc)
            ));
        }
        try {
            esClient.bulk(bl.refresh(Refresh.WaitFor).build());
        } catch (IOException e) {
            throw new RuntimeException(e);
        }

    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

连接Https集群

带安全验证的连接有点复杂，将下列代码中CA证书的位置改为实际所在的位置就行了。

通过用户名密码连接

password为elastic的密码，可以在我的另一篇文章中查看密码的重置方式
Docker安装部署[8.x]版本Elasticsearch+Kibana+IK分词器

    void makeConnection_https() throws CertificateException, IOException,
            NoSuchAlgorithmException, KeyStoreException, KeyManagementException {
        // 创建凭据提供器
        final CredentialsProvider credentialsProvider =
                new BasicCredentialsProvider();
        credentialsProvider.setCredentials(AuthScope.ANY,
                new UsernamePasswordCredentials("elastic", password));

        // 设置CA证书路径
        Path caCertificatePath = Paths.get("E:\\tools\\elasticsearch-8.10.2\\config\\certs\\http_ca.crt");
        // 创建证书工厂
        CertificateFactory factory =
                CertificateFactory.getInstance("X.509");
        Certificate trustedCa;
        try (InputStream is = Files.newInputStream(caCertificatePath)) {
            // 从输入流中生成证书
            trustedCa = factory.generateCertificate(is);
        }
        // 创建密钥库
        KeyStore trustStore = KeyStore.getInstance("pkcs12");
        trustStore.load(null, null);
        trustStore.setCertificateEntry("ca", trustedCa);
        // 创建SSL上下文构建器
        SSLContextBuilder sslContextBuilder = SSLContexts.custom()
                .loadTrustMaterial(trustStore, null);
        final SSLContext sslContext = sslContextBuilder.build();

        // 构建Rest客户端构建器
        RestClientBuilder builder = RestClient.builder(
                        new HttpHost("localhost", 9200, "https"))
                .setHttpClientConfigCallback(new RestClientBuilder.HttpClientConfigCallback() {
                    @Override
                    public HttpAsyncClientBuilder customizeHttpClient(
                            HttpAsyncClientBuilder httpClientBuilder) {
                        return httpClientBuilder.setSSLContext(sslContext)
                                .setDefaultCredentialsProvider(credentialsProvider);
                    }
                });

        // 构建Rest客户端
        RestClient restClient = builder.build();

        // 创建Rest客户端传输
        ElasticsearchTransport transport = new RestClientTransport(
                restClient, new JacksonJsonpMapper());
        esClient = new ElasticsearchClient(transport);
//        asyncClient = new ElasticsearchAsyncClient(transport);
    }
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

通过ApiKey连接

ApiKey在Kibana的Security下生成

     void makeConnection_token() throws CertificateException, IOException,
            NoSuchAlgorithmException, KeyStoreException, KeyManagementException {
        // 定义CA证书路径
        Path caCertificatePath = Paths.get("E:\\tools\\elasticsearch-8.10.2\\config\\certs\\http_ca.crt");
        // 创建X.509证书工厂
        CertificateFactory factory =
                CertificateFactory.getInstance("X.509");
        Certificate trustedCa;
        try (InputStream is = Files.newInputStream(caCertificatePath)) {
            // 从输入流中生成X.509证书
            trustedCa = factory.generateCertificate(is);
        }
        // 创建PKCS12密钥库
        KeyStore trustStore = KeyStore.getInstance("pkcs12");
        trustStore.load(null, null);
        // 将CA证书添加到密钥库
        trustStore.setCertificateEntry("ca", trustedCa);
        // 创建SSL上下文构建器，并设置信任材料
        SSLContextBuilder sslContextBuilder = SSLContexts.custom()
                .loadTrustMaterial(trustStore, null);
        final SSLContext sslContext = sslContextBuilder.build();

        // 创建Rest客户端构建器
        RestClientBuilder builder = RestClient.builder(
                        new HttpHost("localhost", 9200, "https"))
                .setHttpClientConfigCallback(new RestClientBuilder.HttpClientConfigCallback() {
                    @Override
                    public HttpAsyncClientBuilder customizeHttpClient(
                            HttpAsyncClientBuilder httpClientBuilder) {
                        return httpClientBuilder.setSSLContext(sslContext);
                    }
                });

        // 设置默认请求头
        Header[] defaultHeaders =
                new Header[]{new BasicHeader("Authorization",
                        "ApiKey yourApiKey")};
        builder.setDefaultHeaders(defaultHeaders);

        // 构建Rest客户端
        RestClient restClient = builder.build();

        // 创建基于RestClient的传输方式
        ElasticsearchTransport transport = new RestClientTransport(
                restClient, new JacksonJsonpMapper());

        // 创建Elasticsearch客户端
        esClient = new ElasticsearchClient(transport);
    }
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