ES升级--04--SpringBoot整合Elasticsearch

提示：文章写完后，目录可以自动生成，如何生成可参考右边的帮助文档

SpringBoot整合Elasticsearch

1.建立项目

2.Maven 依赖

进入到 ES 官方网站

ES 官方网站：https://www.elastic.co/guide/en/elasticsearch/client/java-rest/6.8/index.html

• 可以看到有低级和 高级的 Rest Client

3. pom配置

基于 springboot 2.1.7.RELEASE

<dependencies>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>

        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter</artifactId>
        </dependency>

        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
            <exclusions>
                <exclusion>
                    <groupId>org.junit.vintage</groupId>
                    <artifactId>junit-vintage-engine</artifactId>
                </exclusion>
            </exclusions>
        </dependency>
 <!-- ES -->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-data-elasticsearch</artifactId>
        </dependency>
         <dependency>
            <groupId>org.springframework.data</groupId>
            <artifactId>spring-data-elasticsearch</artifactId>
            <version>3.2.9.RELEASE</version>
        </dependency>
        <dependency>
            <groupId>org.elasticsearch.client</groupId>
            <artifactId>elasticsearch-rest-high-level-client</artifactId>
            <version>6.8.23</version>
            <exclusions>
                <exclusion>
                    <groupId>org.elasticsearch.client</groupId>
                    <artifactId>transport</artifactId>
                </exclusion>
                <exclusion>
                    <groupId>org.elasticsearch.client</groupId>
                    <artifactId>elasticsearch-rest-client</artifactId>
                </exclusion>
                <exclusion>
                    <groupId>org.elasticsearch</groupId>
                    <artifactId>elasticsearch</artifactId>
                </exclusion>
            </exclusions>
        </dependency>
        <dependency>
            <groupId>org.elasticsearch.client</groupId>
            <artifactId>transport</artifactId>
            <version>6.8.23</version>
        </dependency>
        <dependency>
            <groupId>org.elasticsearch.client</groupId>
            <artifactId>elasticsearch-rest-client</artifactId>
            <version>6.8.23</version>
        </dependency>
        <dependency>
            <groupId>org.elasticsearch</groupId>
            <artifactId>elasticsearch</artifactId>
            <version>6.8.23</version>
        </dependency>
        <dependency>
            <groupId>org.elasticsearch.plugin</groupId>
            <artifactId>transport-netty4-client</artifactId>
            <version>6.8.23</version>
        </dependency>
        <!-- Elasticsearch客户端依赖版本升级到6.8.32 新增x-pack依赖-->
        <dependency>
            <groupId>org.elasticsearch.client</groupId>
            <artifactId>x-pack-transport</artifactId>
            <version>6.8.23</version>
        </dependency>

4.证书文件elastic-certificates.p12 拷贝

证书文件elastic-certificates.p12需拷贝到所有ES节点对应的目录下

• 注意：只需创建一次证书并将其复制到所有节点。

5.配置类 ElasticsearchConfig

支持x-pack 密码验证

/**
 * ES 配置 -----Elasticsearch 6.8.23
 * 通过实现配置配，初始化安全Elasticsearch客户端对象，包括ElasticsearchTemplate和RestHighLevelClient两者客户端类
 * 支持x-pack 密码验证
 */

@Slf4j
@Configuration
public class SecureElasticsearchConfig {


    //用户名 elastic
    @Value("${elasticsearch.xpack.username}")
    private String xpackUsername = "elastic";
    //用户密码
    @Value("${elasticsearch.xpack.password}")
    private String xpackrPassword;

    //证书路径  "/home/data/es"
    @Value("${elasticsearch.xpack.kspath}")
    private String certPath;
    //证书密码  ""
    @Value("${elasticsearch.xpack.kspwd}")
    private String certPassword;

    //集群名
    @Value("${elasticsearch.master.cluster-name}")
    private String masterClusterName;
    //节点名
    @Value("${elasticsearch.master.clusterNodes}")
    private String clusterNodes;
    //ip
    @Value("${elasticsearch.master.address}")
    private String masterAddress;
    //端口
    @Value("${elasticsearch.master.port}")
    private Integer masterPort;

//    // es 连接超时时间
//    private int connectTimeOut;
//    // es socket 连接超时时间
//    private int socketTimeOut;
//    // es 请求超时时间
//    private int connectionRequestTimeOut;
//    // es 最大连接数
//    private int maxConnectNum;
//    // es 每个路由的最大连接数
//    private int maxConnectNumPerRoute;



    /**
     *集群配置
     */
    private Settings settings() {
        Settings.Builder builder = Settings.builder();
        //基础配置
        builder.put("cluster.name", masterClusterName);
        builder.put("xpack.security.user", xpackUsername+ ":" + xpackrPassword);
        // Keystore 配置
        builder.put("xpack.security.transport.ssl.keystore.path", certPath);
        builder.put("xpack.security.transport.ssl.keystore.password", certPassword);
        // Truststore 配置
        builder.put("xpack.security.transport.ssl.truststore.path", certPath);
        builder.put("xpack.security.transport.ssl.truststore.password", certPassword);
        // 验证模式配置
        builder.put("xpack.security.transport.ssl.verification_mode", "certificate");
        // 启用 X-Pack 安全功能
        builder.put("xpack.security.enabled", true);
        builder.put("xpack.security.transport.ssl.enabled", true);
        return builder.build();
    }

    /**
     * 初始化安全TransportClient类
     */
    @Bean
    public TransportClient transportClient() throws Exception {

        //本地测试用
       // certPath="D:\\cdms\\es\\elastic-certificates.p12";
        log.info(">>>>>>>>>>> SecureElasticsearchConfig TransportClient 开始初始化");
        Settings settings = settings();
        PreBuiltXPackTransportClient client = new PreBuiltXPackTransportClient(settings);
        client.addTransportAddress(new TransportAddress(InetAddress.getByName(masterAddress), masterPort));
        return client;
    }

    /**
     *初始化安全ElasticsearchTemplate类
     * 基于 spring-boot-starter-data
     */
    @Bean
    public ElasticsearchTemplate elasticsearchTemplate(@Autowired TransportClient transportClient) throws Exception {
        log.info(">>>>>>>>>>> SecureElasticsearchConfig ElasticsearchTemplate 开始初始化");
        ElasticsearchTemplate secureElasticsearchTemplate;
        try {
            secureElasticsearchTemplate = new ElasticsearchTemplate(transportClient);
            return secureElasticsearchTemplate;
        } catch (Exception e) {
            log.error("SecureElasticsearchConfig  初始化ElasticsearchTemplate报错： ", e.getMessage());
            throw e;
        }
    }



    /**
     * 初始化安全RestHighLevelClient类
     * 只支持http  端口： 9200
     */
    @Bean
    public RestHighLevelClient restHighLevelClient() {
        log.info(">>>>>>>>>>> SecureElasticsearchConfig RestHighLevelClient 开始初始化");
        final CredentialsProvider credentialsProvider = new BasicCredentialsProvider();

        credentialsProvider.setCredentials(AuthScope.ANY,new UsernamePasswordCredentials(xpackUsername, xpackrPassword));

        RestClientBuilder builder = RestClient.builder(new HttpHost(masterAddress,9200))
                .setHttpClientConfigCallback(new RestClientBuilder.HttpClientConfigCallback() {
                    @Override
                    public HttpAsyncClientBuilder customizeHttpClient(HttpAsyncClientBuilder httpAsyncClientBuilder) {
                        return httpAsyncClientBuilder.setDefaultCredentialsProvider(credentialsProvider);
                    }
                });
        RestHighLevelClient client = new RestHighLevelClient(builder);

//        // 连接延时配置
//        builder.setRequestConfigCallback(requestConfigBuilder -> {
//            requestConfigBuilder.setConnectTimeout(connectTimeOut);
//            requestConfigBuilder.setSocketTimeout(socketTimeOut);
//            requestConfigBuilder.setConnectionRequestTimeout(connectionRequestTimeOut);
//            return requestConfigBuilder;
//        });
//        // 连接数配置
//        builder.setHttpClientConfigCallback(httpClientBuilder -> {
//                    httpClientBuilder.setMaxConnTotal(maxConnectNum);
//                    httpClientBuilder.setMaxConnPerRoute(maxConnectNumPerRoute);
//                    httpClientBuilder.setDefaultCredentialsProvider(credentialsProvider);
//                    return httpClientBuilder;
//        });


        return client;
    }

    /**
     *初始化安全ElasticsearchRestTemplate类
     * 基于 spring-boot-starter-data
     */
    @Bean
    ElasticsearchRestTemplate elasticsearchRestTemplate(@Autowired RestHighLevelClient restHighLevelClient) {
        log.info(">>>>>>>>>>> SecureElasticsearchConfig ElasticsearchRestTemplate 开始初始化");
        return new ElasticsearchRestTemplate(restHighLevelClient);
    }

}

SecureRestClientConfig

import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import org.apache.http.HttpHost;
import org.apache.http.auth.AuthScope;
import org.apache.http.auth.UsernamePasswordCredentials;
import org.apache.http.client.CredentialsProvider;
import org.apache.http.impl.client.BasicCredentialsProvider;
import org.apache.http.impl.nio.client.HttpAsyncClientBuilder;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestClientBuilder;
import org.elasticsearch.client.RestHighLevelClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.elasticsearch.core.ElasticsearchRestTemplate;

import java.util.Arrays;

/**
 * ES 配置 -----Elasticsearch 6.8.23
 * 通过实现配置配，初始化安全 RestHighLevelClient,ElasticsearchRestTemplate客户端类
 * 支持x-pack 密码验证
 */
@Slf4j
@Configuration
@ConfigurationProperties(prefix = "spring.elasticsearch.rest")
public class SecureRestClientConfig {

    //端口 ip
    @Setter
    private String[] hosts = new String[]{};

    //用户名 elastic
    @Setter
    private String xpackusername;
    //用户密码
    @Setter
    private String xpackpassword;


//    // es 连接超时时间
//    private int connectTimeOut;
//    // es socket 连接超时时间
//    private int socketTimeOut;
//    // es 请求超时时间
//    private int connectionRequestTimeOut;
//    // es 最大连接数
//    private int maxConnectNum;
//    // es 每个路由的最大连接数
//    private int maxConnectNumPerRoute;



    /**
     * 初始化安全RestHighLevelClient类
     * 只支持http  端口： 9200
     */
    @Bean
    public RestHighLevelClient restHighLevelClient() {
        log.info(">>>>>>>>>>> RestClientConfig RestHighLevelClient 开始初始化");

        HttpHost[] httpHosts = Arrays.stream(hosts).map(x -> {
            String[] hostInfo = x.split(":");
            return new HttpHost(hostInfo[0], Integer.parseInt(hostInfo[1]));
        }).toArray(HttpHost[]::new);

        log.info("elasticsearch hosts: ", Arrays.toString(httpHosts));

        final CredentialsProvider credentialsProvider = new BasicCredentialsProvider();
        credentialsProvider.setCredentials(AuthScope.ANY, new UsernamePasswordCredentials(xpackusername, xpackpassword));
        RestClientBuilder builder = null;
        try {
            builder = RestClient.builder(httpHosts)
                    .setHttpClientConfigCallback(new RestClientBuilder.HttpClientConfigCallback() {
                        @Override
                        public HttpAsyncClientBuilder customizeHttpClient(HttpAsyncClientBuilder httpAsyncClientBuilder) {
                            return httpAsyncClientBuilder.setDefaultCredentialsProvider(credentialsProvider);
                        }
                    });
        } catch (Exception e) {
            log.error("RestClientConfig  初始化RestHighLevelClient报错： ", e.getMessage());
            throw new RuntimeException(e);
        }

//        // 连接延时配置
//        builder.setRequestConfigCallback(requestConfigBuilder -> {
//            requestConfigBuilder.setConnectTimeout(connectTimeOut);
//            requestConfigBuilder.setSocketTimeout(socketTimeOut);
//            requestConfigBuilder.setConnectionRequestTimeout(connectionRequestTimeOut);
//            return requestConfigBuilder;
//        });
//        // 连接数配置
//        builder.setHttpClientConfigCallback(httpClientBuilder -> {
//                    httpClientBuilder.setMaxConnTotal(maxConnectNum);
//                    httpClientBuilder.setMaxConnPerRoute(maxConnectNumPerRoute);
//                    httpClientBuilder.setDefaultCredentialsProvider(credentialsProvider);
//                    return httpClientBuilder;
//        });

        RestHighLevelClient client = new RestHighLevelClient(builder);
        return client;
    }


    @Bean(name = {"elasticsearchOperations", "elasticsearchTemplate"})
    ElasticsearchRestTemplate elasticsearchRestTemplate(@Autowired RestHighLevelClient restHighLevelClient) {
        return new ElasticsearchRestTemplate(restHighLevelClient);
    }
}

6.nacos配置参数

elasticsearch:
  xpack:
    username: elastic
    password: escdms
    kspath: /home/data/es
    kspwd: 
  master:
    cluster-name: gz-java-test-laas
    clusterNodes: master-test-laas
    address: 192.168.2.89
    port: 9300

7.测试

1.TransportClient

 @Autowired
    private  TransportClient transportClient;


    @Test
    public void createIndex_transportClient() {
        String indexName="lass_test_transportclient";

        try {
            CreateIndexRequest request = new CreateIndexRequest(indexName);
            // 可以在此处添加更多设置，例如映射 (mapping) 和设置 (settings)

            CreateIndexResponse response = transportClient.admin().indices().create(request).actionGet();
            if (response.isAcknowledged()) {
                System.out.println("Index created successfully: " + indexName);
            } else {
                System.out.println("Index creation failed: " + indexName);
            }
        } catch (Exception e) {
            System.err.println("Error creating index: " + e.getMessage());
        }

    }



    @Test
    public void addDocuments_transportClient() {
        String indexName = "lass_test_transportclient";

        try {


            String json1 = "{" +
                    "\"user\":\"kimchy\"," +
                    "\"postDate\":\"2013-01-30\"," +
                    "\"message\":\"trying out Elasticsearch\"" +
                    "}";

            IndexResponse response1 = transportClient.prepareIndex(indexName, "_doc")
                    .setSource(json1, XContentType.JSON)
                    .get();


//            if (response1.status() == RestStatus.CREATED) {
//                System.out.println("Document 1 indexed successfully.");
//            } else {
//                System.out.println("Failed to index Document 1.");
//            }

            String json2 = "{" +
                    "\"user\":\"Tom\"," +
                    "\"postDate\":\"2024-01-30\"," +
                    "\"message\":\"lass升级 transportClient \"" +
                    "}";

            transportClient.prepareIndex(indexName, "_doc")
                    .setSource(json2, XContentType.JSON)
                    .get();

        } catch (Exception e) {
            System.err.println("Error adding documents: " + e.getMessage());
        }
    }





    @Test
    public void deleteIndex_transportClient() {
        String indexName = "lass_test_transportclient";

        try {
            DeleteIndexRequest request = new DeleteIndexRequest(indexName);
            AcknowledgedResponse response = transportClient.admin().indices().delete(request).actionGet();
            if (response.isAcknowledged()) {
                System.out.println("Index deleted successfully: " + indexName);
            } else {
                System.out.println("Failed to delete index: " + indexName);
            }
        } catch (Exception e) {
            System.err.println("Error deleting index: " + e.getMessage());
        }
    }

GET lass_test_transportclient/_search
{
  "query":{
    "match_all" : {}
  }
}

2.ElasticsearchTemplate

@Data
@AllArgsConstructor
@NoArgsConstructor
@Accessors(chain = true)
@Document(indexName = "lass_test_people",type = "_doc",shards = 1,replicas = 1)
public class People {
    @Id
    private String id;
    // 整个name不被分词，切不创建索引
    // Keyword表示不被分词
    @Field(type= FieldType.Keyword,index = false)
    private String name;
    // address被ik分词
    // Text类型的属性才能被分词
    @Field(type = FieldType.Text)
    private String address;

    @Field(type = FieldType.Long,index = false)
    private int age;

}

 @Autowired
    private ElasticsearchTemplate elasticsearchTemplate;

    @Test
    public void createIndex_elasticsearchTemplate() {

        //根据实体类创建索引，
        boolean result1 = elasticsearchTemplate.createIndex(People.class);
        System.out.println(result1);
        //将索引放到软件里面
        boolean results = elasticsearchTemplate.putMapping(People.class);

    }




    @Test
    public void addDocuments_elasticsearchTemplate() {

        People peo = new People();
        peo.setId("123");
        peo.setName("张三");
        peo.setAddress("北京市海淀区回龙观东大街");
        peo.setAge(18);

        IndexQuery query = new IndexQuery();
        query.setObject(peo);

        String result = elasticsearchTemplate.index(query);
        System.out.println(result);
    }

    @Test
    public void bulk(){
        List<IndexQuery> list = new ArrayList<>();
        // IndexQuery多行写法
        IndexQuery indexQuery = new IndexQuery();
        indexQuery.setObject(new People("1", "王五", "北京东城", 12));
        list.add(indexQuery);
        // IndexQuery 连缀写法
        list.add(new IndexQueryBuilder().withObject(new People("2", "赵六", "北京西城", 13)).build());
        list.add(new IndexQueryBuilder().withObject(new People("3", "吴七", "北京昌平", 14)).build());
        elasticsearchTemplate.bulkIndex(list);
    }


    @Test
    public void deletee_elasticsearchTemplate() {
        boolean result = elasticsearchTemplate.deleteIndex(People.class);
        System.out.println(result);
    }

3.RestHighLevelClient

@Data
@AllArgsConstructor
@NoArgsConstructor
@Accessors(chain = true)
@Document(indexName = "lass_test_student",type = AudienceEsConst.DOC, createIndex = false, useServerConfiguration = true)
public class Student {
    @Id
    private String id;
    // 整个name不被分词，切不创建索引
    // Keyword表示不被分词
    @Field(type= FieldType.Keyword,index = false)
    private String name;
    // address被ik分词
    // Text类型的属性才能被分词
    @Field(type = FieldType.Text)
    private String address;

    @Field(type = FieldType.Long,index = false)
    private int age;

}

 @Autowired
    private RestHighLevelClient restHighLevelClient;

    @Test
    public void createIndex_restHighLevelClient() throws IOException {
        String indexName = "lass_test_resthighlevelclient";

        XContentBuilder builder = XContentFactory.jsonBuilder();
        builder.startObject();
        {
            builder.field("user", "zhangSan");
            builder.timeField("postDate", new Date());
            builder.field("message", "laas 升级 RestHighLevelClient ");
        }

        builder.endObject();
        IndexRequest request = new IndexRequest(indexName, "doc").source(builder);

        IndexResponse indexResponse = restHighLevelClient.index(request, RequestOptions.DEFAULT);

            if (indexResponse.status() == RestStatus.CREATED) {
                System.out.println("Document 1 indexed successfully.");
            } else {
                System.out.println("Failed to index Document 1.");
            }

    }



    @Test
    public void addDocuments_restHighLevelClient() {
        String indexName = "lass_test_resthighlevelclient";

        try {
            Map<String, Object> jsonMap = new HashMap<>();
            jsonMap.put("user", "李四");
            jsonMap.put("postDate", new Date());
            jsonMap.put("message", "laas 升级 RestHighLevelClient ");
            IndexRequest indexRequest = new IndexRequest(indexName,"doc").source(jsonMap);
            IndexResponse indexResponse = restHighLevelClient.index(indexRequest, RequestOptions.DEFAULT);

            if (indexResponse.status() == RestStatus.CREATED) {
                System.out.println("Document 1 indexed successfully.");
            } else {
                System.out.println("Failed to index Document 1.");
            }

        } catch (Exception e) {
            System.err.println("Error adding documents: " + e.getMessage());
        }
    }



    @Test
    public void deletee_restHighLevelClient() {
        String indexName = "lass_test_resthighlevelclient";
        boolean result = elasticsearchTemplate.deleteIndex(indexName);
        System.out.println(result);
    }

4.ElasticsearchRestTemplate

 @Autowired
    private  ElasticsearchRestTemplate  restTemplate;

    @Test
    public void createIndex_restTemplate() {

        //根据实体类创建索引，
        boolean result1 = restTemplate.createIndex(Student.class);
        System.out.println(result1);
        //将索引放到软件里面
        boolean results = restTemplate.putMapping(Student.class);

    }

    @Test
    public void addDocuments_restTemplate() {

        Student student = new Student();
        student.setId("123");
        student.setName("张三");
        student.setAddress("北京市海淀区回龙观东大街");
        student.setAge(18);

        IndexQuery query = new IndexQuery();
        query.setObject(student);

        String result = restTemplate.index(query);
        System.out.println(result);
    }

    @Test
    public void bulk_restTemplate(){
        List<IndexQuery> list = new ArrayList<>();
        // IndexQuery多行写法
        IndexQuery indexQuery = new IndexQuery();
        indexQuery.setObject(new Student("1", "王五", "北京东城", 12));
        list.add(indexQuery);
        // IndexQuery 连缀写法
        list.add(new IndexQueryBuilder().withObject(new Student("2", "赵六", "北京西城", 13)).build());
        list.add(new IndexQueryBuilder().withObject(new Student("3", "吴七", "北京昌平", 14)).build());
        restTemplate.bulkIndex(list);
    }


    @Test
    public void deletee_restTemplate() {
        boolean result = restTemplate.deleteIndex(Student.class);
        System.out.println(result);
    }