Elasticsearch - Java API 操作 ES7.16.0+ES8.x 索引，文档；高级搜索

Posted 2022-09-19 MinggeQingchun

阅读本文前可先参考

Elasticsearch - Java API 操作 ES7.15.0、ES7.x 索引，文档；高级搜索（六）_MinggeQingchun的博客-CSDN博客

Elasticsearch - Java API（Lambda 表达式） 操作 ES7.16.0+、ES8.x 索引，文档；高级搜索（七）_MinggeQingchun的博客-CSDN博客

Elasticsearch 软件是由 Java 语言开发的，所以也可以通过 Java API 的方式对 Elasticsearch服务进行访问

以 Elasticsearch7.16.0版本为例 

1、创建空工程elasticsearch-demo，创建普通Java的maven模块elasticsearch-test

2、添加依赖

查看Elasticsearch官网文档

 Installation | Elasticsearch Java API Client [7.16] | Elastic

<dependencies>
 
    <!-- elasticsearch -->
<!--    <dependency>-->
<!--      <groupId>org.elasticsearch</groupId>-->
<!--      <artifactId>elasticsearch</artifactId>-->
<!--      <version>7.16.0</version>-->
<!--    </dependency>-->
 
    <!-- elasticsearch 的客户端 -->
    <dependency>
      <groupId>co.elastic.clients</groupId>
      <artifactId>elasticsearch-java</artifactId>
      <version>7.16.0</version>
    </dependency>
 
    <!-- elasticsearch 依赖 2.x 的 log4j -->
    <dependency>
      <groupId>org.apache.logging.log4j</groupId>
      <artifactId>log4j-api</artifactId>
      <version>2.17.0</version>
    </dependency>
    <dependency>
      <groupId>org.apache.logging.log4j</groupId>
      <artifactId>log4j-core</artifactId>
      <version>2.17.0</version>
    </dependency>
 
    <dependency>
      <groupId>com.fasterxml.jackson.core</groupId>
      <artifactId>jackson-databind</artifactId>
      <version>2.12.3</version>
    </dependency>
 
    <!-- junit 单元测试 -->
    <dependency>
      <groupId>junit</groupId>
      <artifactId>junit</artifactId>
      <version>4.12</version>
    </dependency>
  </dependencies>

一、Elasticsearch 客户端对象

在Elasticsearch7.15版本之后，Elasticsearch官方将它的高级客户端 RestHighLevelClient 标记为弃用状态

同时推出了全新的Java API客户端Elasticsearch Java API Client，该客户端也将在Elasticsearch8.0及以后版本中成为官方推荐使用的客户端

就像连接 mysql 数据库一样，Java 通过客户端操作 Elasticsearch 也要获取到连接后才可以使用

可参考

https://blog.csdn.net/MinggeQingchun/article/details/126830831

在第一台服务器节点 es-node-1 设置集群多节点 HTTP 证书 

基于https 安全的 Elasticsearch 服务，所以首先我们需要将之前的证书进行一个转换

cd /opt/software/elasticsearch-7.16.0/config/certs/
openssl pkcs12 -in elastic-stack-ca.p12 -clcerts -nokeys -out java-ca.crt

将其下载下来并导入到Java项目中

创建客户端

import co.elastic.clients.elasticsearch.*;
import co.elastic.clients.json.jackson.JacksonJsonpMapper;
import co.elastic.clients.transport.ElasticsearchTransport;
import co.elastic.clients.transport.rest_client.RestClientTransport;
import org.apache.http.HttpHost;
import org.apache.http.auth.*;
import org.apache.http.client.*;
import org.apache.http.conn.ssl.NoopHostnameVerifier;
import org.apache.http.impl.client.*;
import org.apache.http.impl.nio.client.HttpAsyncClientBuilder;
import org.apache.http.ssl.*;
import org.elasticsearch.client.*;
import javax.net.ssl.SSLContext;
import java.io.IOException;
import java.io.InputStream;
import java.nio.file.*;
import java.security.KeyManagementException;
import java.security.KeyStore;
import java.security.KeyStoreException;
import java.security.NoSuchAlgorithmException;
import java.security.cert.*;

/**
 * 在Elasticsearch7.15版本之后，Elasticsearch官方将它的高级客户端 RestHighLevelClient 标记为弃用状态
 *   同时推出了全新的Java API客户端Elasticsearch Java API Client，该客户端也将在Elasticsearch8.0及以后版本中成为官方推荐使用的客户端
 * Elasticsearch Java API Client 支持除 Vector tile search API 和 Find structure API 之外的所有 Elasticsearch API。
 *   且支持所有API数据类型，并且不再有原始JsonValue属性。它是针对Elasticsearch8.0及之后版本的客户端
 */
public class ESClient 
    public static void main(String[] args) throws CertificateException, IOException, KeyStoreException, NoSuchAlgorithmException, KeyManagementException 

        final CredentialsProvider credentialsProvider = new BasicCredentialsProvider();
        credentialsProvider.setCredentials(AuthScope.ANY,
                new UsernamePasswordCredentials("elastic", "O3x0hfu7i=ZbQvlktCnd"));
        Path caCertificatePath = Paths.get("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);
        SSLContextBuilder sslContextBuilder = SSLContexts.custom()
                .loadTrustMaterial(trustStore, null);

        final SSLContext sslContext = sslContextBuilder.build();
        RestClientBuilder builder = RestClient.builder(
                new HttpHost("linux1", 9200, "https"))
                .setHttpClientConfigCallback(new RestClientBuilder.HttpClientConfigCallback() 
                    @Override
                    public HttpAsyncClientBuilder customizeHttpClient(
                            HttpAsyncClientBuilder httpClientBuilder) 
                        return httpClientBuilder.setSSLContext(sslContext)
                                .setSSLHostnameVerifier(NoopHostnameVerifier.INSTANCE)
                                .setDefaultCredentialsProvider(credentialsProvider);
                    
                );

        // 创建低级客户端
        RestClient restClient = RestClient.builder(new HttpHost("localhost",9200)).build();

        // 使用Jackson映射器创建传输层
        ElasticsearchTransport transport = new RestClientTransport(restClient,new JacksonJsonpMapper());

        // 创建API 同步客户端
        ElasticsearchClient esClient = new ElasticsearchClient(transport);
        // 创建API 异步客户端
        ElasticsearchAsyncClient asyncClient = new ElasticsearchAsyncClient(transport);

        // 关闭ES客户端
        try 
            transport.close();
            restClient.close();
         catch (IOException e) 
            e.printStackTrace();
        
    

二、索引操作

// 创建索引
CreateIndexRequest request = new 
CreateIndexRequest.Builder().index("myindex").build();
final CreateIndexResponse createIndexResponse = 
client.indices().create(request);
System.out.println("创建索引成功：" + createIndexResponse.acknowledged());


// 查询索引
GetIndexRequest getIndexRequest = new 
GetIndexRequest.Builder().index("myindex").build();
final GetIndexResponse getIndexResponse = 
client.indices().get(getIndexRequest);
System.out.println("索引查询成功：" + getIndexResponse.result());


// 删除索引
DeleteIndexRequest deleteIndexRequest = new 
DeleteIndexRequest.Builder().index("myindex").build();
final DeleteIndexResponse delete = client.indices().delete(deleteIndexRequest);
final boolean acknowledged = delete.acknowledged();
System.out.println("删除索引成功：" + acknowledged);

三、文档操作

// 创建文档
IndexRequest indexRequest = new IndexRequest.Builder()
 .index("myindex")
 .id(user.getId().toString())
 .document(user)
 .build();
final IndexResponse index = client.index(indexRequest);
System.out.println("文档操作结果:" + index.result());


// 批量创建文档
final List<BulkOperation> operations = new ArrayList<BulkOperation>();
for ( int i= 1;i <= 5; i++ ) 
 final CreateOperation.Builder builder = new CreateOperation.Builder();
 builder.index("myindex");
 builder.id("200" + i);
 builder.document(new User(2000 + i, 30 + i * 10, "zhangsan" + i, "beijing", 
1000 + i*1000));
 final CreateOperation<Object> objectCreateOperation = builder.build();
 final BulkOperation bulk = new 
BulkOperation.Builder().create(objectCreateOperation).build();
 operations.add(bulk);

BulkRequest bulkRequest = new 
BulkRequest.Builder().operations(operations).build();
final BulkResponse bulkResponse = client.bulk(bulkRequest);
System.out.println("数据操作成功：" + bulkResponse);


// 删除文档
DeleteRequest deleteRequest = new 
DeleteRequest.Builder().index("myindex").id("1001").build();
client.delete(deleteRequest);

四、文档高级查询

final SearchRequest.Builder searchRequestBuilder = new SearchRequest.Builder().index("myindex1");
        MatchQuery matchQuery = new
                MatchQuery.Builder().field("city").query(FieldValue.of("beijing")).build();
        Query query = new Query.Builder().match(matchQuery).build();
        searchRequestBuilder.query(query);
        SearchRequest searchRequest = searchRequestBuilder.build();
        final SearchResponse<Object> search = esClient.search(searchRequest,
                Object.class);
        System.out.println(search);

五、客户端异步操作

// 创建索引
        asyncClient.indices().create(
                req -> 
                    req.index("newindex");
                    return req;
                
        ).whenComplete(
                (resp, error) -> 
                    System.out.println("回调函数");
                    if ( resp != null ) 
                        System.out.println(resp.acknowledged());
                     else 
                        error.printStackTrace();
                    
                
        );
        
        System.out.println("主线程操作...");
        
        asyncClient.indices().create(
                req -> 
                    req.index("newindex");
                    return req;
                 )
                .thenApply(
                        resp -> 
                            return resp.acknowledged();
                         )
                .whenComplete(
                        (resp, error) -> 
                            System.out.println("回调函数");
                            if ( !resp ) 
                                System.out.println();
                             else 
                                error.printStackTrace();
                            
                        
                );