[Java知识库]spring boot es | spring boot 整合elasticsearch | spring boot整合多数据源es

目录

Spring Boot与ES版本对应

Maven依赖

配置类

使用方式

@Test中注入方式

@Component中注入方式

查询文档

实体类

通过ElasticsearchRestTemplate查询

通过JPA查询

保存文档

参考链接

项目组件版本：

Spring Boot：2.2.13.RELEASE

Elasticsearch：6.8.0

JDK：1.8.0_66

Spring Boot与ES版本对应

Tips: 主要看第3列和第5列，根据ES版本选择对应的Spring Boot版本，如果ES和Spring Boot版本不一致后续会报错。

Maven依赖

<parent>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-parent</artifactId>
    <version>2.2.13.RELEASE</version>
    <relativePath/> <!-- lookup parent from repository -->
</parent>

<!-- 其他无关内容省略 -->

 <dependencies>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-data-elasticsearch</artifactId>
        <version>2.2.13.RELEASE</version>
    </dependency>
</dependencies>

配置类

通过配置类定义两个ES链接的elasticsearchClient，如果是一个连接删除其中一个即可。

import org.apache.commons.lang3.StringUtils;
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.reactor.IOReactorConfig;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestClientBuilder;
import org.elasticsearch.client.RestHighLevelClient;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Primary;
import org.springframework.core.convert.support.DefaultConversionService;
import org.springframework.data.elasticsearch.client.ClientConfiguration;
import org.springframework.data.elasticsearch.client.RestClients;
import org.springframework.data.elasticsearch.config.AbstractElasticsearchConfiguration;
import org.springframework.data.elasticsearch.core.ElasticsearchEntityMapper;
import org.springframework.data.elasticsearch.core.ElasticsearchRestTemplate;
import org.springframework.data.elasticsearch.core.EntityMapper;
import org.springframework.http.HttpHeaders;

/**
 * @author He Changjie on 2022/6/6 14:02
 */
@Configuration
public class ElasticSearchConfig extends AbstractElasticsearchConfiguration{
    /** ES链接一 [host:port] */
    @Value("${spring.data.elasticsearch.client.reactive.endpoints}")
    private String endpoints;
    /** ES链接二 [host:port] */
    @Value("${spring.data.elasticsearch.client.reactive.endpoints.zeek}")
    private String endpointsZeek;
    /** 连接elasticsearch超时时间 */
    @Value("${spring.data.elasticsearch.client.reactive.connection-timeout}")
    private Integer connectTimeout;
    /** 套接字超时时间 */
    @Value("${spring.data.elasticsearch.client.reactive.socket-timeout}")
    private Integer socketTimeout;

    /** 用户名 */
    @Value("${spring.data.elasticsearch.client.reactive.username}")
    private String username;
    /** 密码 */
    @Value("${spring.data.elasticsearch.client.reactive.password}")
    private String password;

    @Bean("elasticsearchRestTemplate")
    @Primary
    public ElasticsearchRestTemplate elasticsearchTemplate() {
        return new ElasticsearchRestTemplate(elasticsearchClient());
    }

    /**
     * 构建方式一
     */
    @Bean("restHighLevelClient")
    @Primary
    @Override
    public RestHighLevelClient elasticsearchClient() {
        // 初始化 RestClient, hostName 和 port 填写集群的内网 IP 地址与端口
        final String host = StringUtils.substringBefore(endpoints, ":");
        final int port = Integer.parseInt(StringUtils.substringAfter(endpoints, ":"));
        RestClientBuilder builder = RestClient.builder(new HttpHost(host, port))
                .setRequestConfigCallback(config -> {
                    config.setConnectTimeout(connectTimeout);
                    config.setSocketTimeout(socketTimeout);
                    return config;
                });
        //保活策略
        builder.setHttpClientConfigCallback(httpClientBuilder -> httpClientBuilder
                .setDefaultIOReactorConfig(IOReactorConfig.custom()
                        .setSoKeepAlive(true)
                        .build()));
        // 设置认证信息
        final CredentialsProvider credentialsProvider = new BasicCredentialsProvider();
        credentialsProvider.setCredentials(AuthScope.ANY, new UsernamePasswordCredentials(username, password));
        builder.setHttpClientConfigCallback(httpAsyncClientBuilder -> {
            httpAsyncClientBuilder.disableAuthCaching();
            return httpAsyncClientBuilder.setDefaultCredentialsProvider(credentialsProvider);
        });

        return new RestHighLevelClient(builder);
    }

    @Bean("zeekElasticsearchTemplate")
    public ElasticsearchRestTemplate ZeekElasticsearchTemplate() {
        return new ElasticsearchRestTemplate(zeekRestHighLevelClient());
    }

    /**
     * 构建方式二
     */
    @Bean("zeekRestHighLevelClient")
    public RestHighLevelClient zeekRestHighLevelClient() {
        HttpHeaders defaultHeaders = new HttpHeaders();
        defaultHeaders.setBasicAuth(username, password);
        ClientConfiguration clientConfiguration = ClientConfiguration.builder()
                .connectedTo(endpointsZeek)
                .withConnectTimeout(connectTimeout)
                .withSocketTimeout(socketTimeout)
                .withDefaultHeaders(defaultHeaders)
                .withBasicAuth(username, password)
                .build();
        return RestClients.create(clientConfiguration).rest();
    }

    @Bean
    @Override
    public EntityMapper entityMapper() {
        ElasticsearchEntityMapper entityMapper = new ElasticsearchEntityMapper(elasticsearchMappingContext(),
                new DefaultConversionService());
        entityMapper.setConversions(elasticsearchCustomConversions());
        return entityMapper;
    }
}

Tips:

1. 配置类中定义RestHighLevelClient使用了两个种方式，任选其中一种都可以
2. 必须选择一个ES链接打上@Primary注解
3. 示例中两个连接都需要密码，且密码相同

使用方式

elasticsearchRestTemplate的使用在@Test中和其他@Component中注入方式不同（亲测），在@Component中直接使用@Resource注入ElasticsearchRestTemplate会报找不到对应的Bean。

@Test中注入方式

@Resource(name = "elasticsearchRestTemplate")
private ElasticsearchRestTemplate elasticsearchRestTemplate;
@Resource(name = "zeekElasticsearchTemplate")
private ElasticsearchRestTemplate zeekElasticsearchTemplate;

@Component中注入方式

@Service
public class DemoServiceImpl implements DemoService {
    private final ElasticsearchRestTemplate elasticsearchRestTemplate;
    private final ElasticsearchRestTemplate zeekElasticsearchRestTemplate;

    @Autowired
    public DemoServiceImpl(RestHighLevelClient restHighLevelClient,
    					@Qualifier(value = "zeekRestHighLevelClient") RestHighLevelClient zeekRestHighLevelClient) {
        this.elasticsearchRestTemplate = new ElasticsearchRestTemplate(restHighLevelClient);
        this.zeekElasticsearchRestTemplate = new ElasticsearchRestTemplate(zeekRestHighLevelClient);
    }
}

查询文档

实体类

import lombok.Data;
import org.springframework.data.annotation.Id;
import org.springframework.data.elasticsearch.annotations.Document;
import org.springframework.data.elasticsearch.annotations.Field;
import org.springframework.data.elasticsearch.annotations.FieldType;
import java.io.Serializable;

@Data
@Document(indexName = "demo-index-20220609", type = "log")
public class QdnsLogs implements Serializable {
    @Id
    private String _id;
    @Field(type = FieldType.Keyword)
    private String name;
    @Field(type = FieldType.Keyword)
    private String address;
    // ........
    @Field(type = FieldType.Date, name = "timestamp")
    private Long timestamp;
}

Tips:

1. 该类具体内容进行了脱敏
2. 需要特别注意@Document的type一定要和es中的_type一致，否则查询结果为是空
3. 如果不需要保存文档，可以不要@Field注解

通过ElasticsearchRestTemplate查询

import com.xxx.entity.es.Eth0Logs;
import com.xxx.entity.es.QdnsLogs;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.SearchHits;
import org.elasticsearch.search.aggregations.AggregationBuilders;
import org.elasticsearch.search.aggregations.bucket.terms.ParsedStringTerms;
import org.elasticsearch.search.aggregations.bucket.terms.Terms;
import org.elasticsearch.search.aggregations.metrics.tophits.TopHits;
import org.elasticsearch.search.sort.SortOrder;
import org.junit.jupiter.api.Test;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.data.elasticsearch.core.ElasticsearchRestTemplate;
import org.springframework.data.elasticsearch.core.aggregation.AggregatedPage;
import org.springframework.data.elasticsearch.core.query.*;
import javax.annotation.Resource;
import java.util.List;

/**
 * 实现描述：
 *
 * @author Hecj
 * @version v 1.0.0
 * @since 2022/06/17
 */
@SpringBootTest(classes = Application.class)
public class EsTest {
    @Resource(name = "elasticsearchRestTemplate")
    private ElasticsearchRestTemplate elasticsearchRestTemplate;
    @Resource(name = "zeekElasticsearchTemplate")
    private ElasticsearchRestTemplate zeekElasticsearchTemplate;

    /**
     * 通过时间范围和是否存在某一字段查询
     */
    @Test
    void test1(){
        SearchQuery searchQuery  = new NativeSearchQueryBuilder()//查询数据，构造出一个查询
                .withQuery(QueryBuilders.boolQuery().must(QueryBuilders.rangeQuery("timestamp").from(1654506000000L).to(1654507340785L)).must(QueryBuilders.existsQuery("name")))
                .build();//构造一个SearchQuery
        List<QdnsLogs> list = elasticsearchRestTemplate.queryForList(searchQuery, QdnsLogs.class);
        System.out.println(list.size());
    }

    /**
     * 通过name值等于特定值
     */
    @Test
    void test2(){
        SearchQuery searchQuery  = new NativeSearchQueryBuilder()
                .withQuery(QueryBuilders.termQuery("name", "zhangsan"))
                .build();//构造一个SearchQuery
        List<Eth0Logs> eth0Logs = zeekElasticsearchTemplate.queryForList(searchQuery, Eth0Logs.class);
        System.out.println(eth0Logs.size());
        for (Eth0Logs log : eth0Logs) {
            System.out.println(log.getTs());
        }
    }

    /**
     * aggs查询
     * 
     * 查询指定时间范围内存在name值的记录，进行通过name聚合，按照时间倒序排序取最新一条记录
     */
    @Test
    void test3() {
        SearchQuery searchQuery  = new NativeSearchQueryBuilder()
                .withQuery(QueryBuilders.boolQuery()
                        .must(
                                QueryBuilders.rangeQuery("timestamp")
                                        .from(1654506000000L)
                                        .to(1654507340785L)
                        )
                        .must(QueryBuilders.existsQuery("name")))
                .addAggregation(AggregationBuilders.terms("name")
                        .field("name")
                        .size(10000)
                        .subAggregation(
                                AggregationBuilders.topHits("top")
                                        .sort("timestamp", SortOrder.DESC)
                                        .size(1)
                        )
                )
                .build();
        AggregatedPage<QdnsLogs> logs = elasticsearchRestTemplate.queryForPage(searchQuery, QdnsLogs.class);
        ParsedStringTerms fqdn = (ParsedStringTerms)logs.getAggregation("name");
        List<? extends Terms.Bucket> buckets = fqdn.getBuckets();
        for (Terms.Bucket entry : buckets) {
            String key = entry.getKeyAsString();
            TopHits topHits= entry.getAggregations().get("top");
            SearchHits hits = topHits.getHits();
            SearchHit at = hits.getAt(0);
            System.out.println(key + "-" + at);
        }
    }
}

Tips: 部分包完整名称进行了脱敏

通过JPA查询

这里的接口不需要添加@Service，通过JPA方式需要特别注意书写规范，字段名称的正确性。

interface BookRepository extends Repository<Book, String> {
  List<Book> findByNameAndPrice(String name, Integer price);
}

相当于：

{
    "query": {
        "bool" : {
            "must" : [
                { "query_string" : { "query" : "?", "fields" : [ "name" ] } },
                { "query_string" : { "query" : "?", "fields" : [ "price" ] } }
            ]
        }
    }
}

Table 2. Supported keywords inside method names

保存文档

import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.core.util.IdUtil;
import org.springframework.data.elasticsearch.core.query.IndexQuery;
import java.util.List;
import java.util.stream.Collectors;
import lombok.extern.slf4j.Slf4j;

// 这里的dataList是需要保存到ES的bean集合，各位自行替换
List<IndexQuery> queries = dataList.stream().map(e -> {
        IndexQuery query = new IndexQuery();
        // 这个自行替换，也可以省略
        query.setId(IdUtil.simpleUUID());
        // 具体的数据
        query.setObject(e);
        // 索引名称
        query.setIndexName("demo-index-20220609");
        // 索引类型
        query.setType("log");
        return query;
    }).collect(Collectors.toList());
    
    if(CollectionUtil.isNotEmpty(queries)){
        zeekElasticsearchTemplate.bulkIndex(queries);
        log.info("#~ 写入日志成功，写入条数：{}", queries.size());
    }

参考链接

Spring Data版本依赖矩阵

elasticsearch官方手册