Springboot+elasticsearch基础整合实例

es,当插入数据的时候，可以自动创建索引，但是mapping却都是默认类型，导致搜索时需要key.keyword方式，不科学。

索引也可以手偶刚创建，指定mapping。

当然还有一种优雅的方案使用template，当自动创建索引的时候，我们的字段类型就可控了。

真实业务中，不能用一个固定的index，索引是需要切分的。

es多用于存储过程数据，指标数据，行为数据，历史数据多数情况下毫无价值，会定期清理。

例如：一天升一个索引index-prefix-yyyy-MM-dd

查询数据可以通过index-prefix*

这里有点尴尬，Springboot内部提供了大量template技术模板，es也有对应的ElasticsearchTemplate，但是并不实用。

我们还是采用RestHighClient发送请求来做es相关操作。

记录一个调试常用shell命令

curl -k --user 'username:password' https://10.110.48.77:9200/_cat   #实用单引号是解决密码中特殊符号问题

1、pom引入依赖

        <dependency>
            <groupId>org.elasticsearch.client</groupId>
            <artifactId>elasticsearch-rest-high-level-client</artifactId>
            <version>7.13.1</version>
        </dependency>
        <dependency>
            <groupId>org.elasticsearch</groupId>
            <artifactId>elasticsearch</artifactId>
            <version>7.13.1</version>
            <exclusions>
                <exclusion>
                    <groupId>org.yaml</groupId>
                    <artifactId>snakeyaml</artifactId>
                </exclusion>
            </exclusions>
        </dependency>

2、properties文件配置es数据源

elastic.cluster.server=7.12.235.22:9200,7.12.225.22:9200
elastic.cluster.schema=https
elastic.cluster.user=cssabc
elastic.cluster.password=7ujm*&^(IJN

3、configuration装配bean

这里注意一个核心要适配http和https就要做SSL免证书

package com.wht.test.es;

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.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 javax.net.ssl.SSLContext;
import javax.net.ssl.TrustManager;
import javax.net.ssl.X509TrustManager;
import java.security.KeyManagementException;
import java.security.NoSuchAlgorithmException;
import java.security.cert.CertificateException;
import java.security.cert.X509Certificate;

/**
 * es配置类
 *
 * @Author 红尘过客
 * @DateTime 2023-08-02 18:34:28
 */
@Slf4j
@Configuration
public class ElasticsearchConfig {
    @Value("${elastic.cluster.server}")
    private String elasticServers;

    @Value("${elastic.cluster.user}")
    private String username;

    @Value("${elastic.cluster.schema}")
    private String password;

    @Value("${elastic.cluster.https}")
    private String schema;

    @Primary
    @Bean
    public RestHighLevelClient restHighLevelClient() {
        return new RestHighLevelClient(restClientBuilder());
    }

    public RestClientBuilder restClientBuilder() {
        String[] servers = elasticServers.split(",");
        HttpHost[] httpHosts = new HttpHost[servers.length];
        for (int i = 0; i < servers.length; i++) {
            String server = servers[i];
            String host = server.split(":")[0];
            String port = server.split(":")[1];
            httpHosts[i] = new HttpHost(host, Integer.parseInt(port));
        }

        CredentialsProvider credentialsProvider = new BasicCredentialsProvider();
        credentialsProvider.setCredentials(AuthScope.ANY, new UsernamePasswordCredentials(username, password));
        return RestClient.builder(httpHosts)
                .setRequestConfigCallback(restClientBuilder -> restClientBuilder.setConnectTimeout(5 * 1000)
                        .setSocketTimeout(10 * 1000))
                .setHttpClientConfigCallback(httpClientBuilder -> httpClientBuilder.setSSLContext(getSslContext())
                        .setSSLHostnameVerifier((hostname, session) -> true)
                        .setDefaultCredentialsProvider(credentialsProvider)
                        .setMaxConnTotal(100)
                        .setMaxConnPerRoute(100));

    }

    private SSLContext getSslContext() {
        SSLContext sslContext = null;
        if ("https".equalsIgnoreCase(schema)) {
            try {
                sslContext = createIgnoreVerifySsl();
            } catch (NoSuchAlgorithmException e) {
                e.printStackTrace();
            } catch (KeyManagementException e) {
                e.printStackTrace();
            }
        }
        return sslContext;
    }

    private SSLContext createIgnoreVerifySsl() throws NoSuchAlgorithmException, KeyManagementException {
        SSLContext sslContext = SSLContext.getInstance("TLS");
        X509TrustManager trustManager = new X509TrustManager() {
            @Override
            public void checkClientTrusted(X509Certificate[] x509Certificates, String s) throws CertificateException {

            }

            @Override
            public void checkServerTrusted(X509Certificate[] x509Certificates, String s) throws CertificateException {

            }

            @Override
            public X509Certificate[] getAcceptedIssuers() {
                return new X509Certificate[0];
            }
        };
        sslContext.init(null, new TrustManager[]{trustManager}, null);
        return sslContext;
    }
}

4、通用服务类

ElasticService 很重要，主要包含了常见的索引创建，数据写入和查询，以及数据更新。

package com.wht.test.es;

import com.alibaba.fastjson.JSONObject;
import lombok.extern.slf4j.Slf4j;
import org.elasticsearch.action.bulk.BulkProcessor;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.update.UpdateRequest;
import org.elasticsearch.action.update.UpdateResponse;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.client.indices.CreateIndexRequest;
import org.elasticsearch.client.indices.GetIndexRequest;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.unit.ByteSizeUnit;
import org.elasticsearch.common.unit.ByteSizeValue;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.common.xcontent.XContentBuilder;
import org.elasticsearch.common.xcontent.XContentFactory;
import org.elasticsearch.common.xcontent.XContentType;
import org.elasticsearch.index.query.BoolQueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.search.SearchHits;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.elasticsearch.search.sort.FieldSortBuilder;
import org.elasticsearch.search.sort.SortOrder;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.io.IOException;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.Locale;
import java.util.concurrent.TimeUnit;

/**
 * desc
 *
 * @Author 红尘过客
 * @DateTime 2023-08-02 18:52:33
 */
@Service
@Slf4j
public class ElasticService {
    private final SimpleDateFormat dateFormat = new SimpleDateFormat("yyyy-MM-dd", Locale.ROOT);

    @Autowired
    private RestHighLevelClient restHighLevelClient;

    private String currentIndex = "test_es_index";

    public boolean existsIndex(String index) {
        GetIndexRequest request = new GetIndexRequest(index);
        boolean exists = false;
        try {
            exists = restHighLevelClient.indices().exists(request, RequestOptions.DEFAULT);
        } catch (IOException ioException) {
            ioException.printStackTrace();
        }
        return exists;
    }

    public BulkProcessor bulkInit() {
        BulkProcessor.Listener listener = new ElasticListener();
        return BulkProcessor.builder((request, bulkListener) -> restHighLevelClient.bulkAsync(request, RequestOptions.DEFAULT, bulkListener), listener)
                .setBulkActions(900)
                .setBulkSize(new ByteSizeValue(5, ByteSizeUnit.MB))
                .setFlushInterval(TimeValue.timeValueSeconds(5))
                .setConcurrentRequests(2)
                .build();
    }

    public BulkProcessor writeEs(String index, JSONObject jsonObject) {
        try (BulkProcessor bulkProcessor = bulkInit()) {
            IndexRequest request = new IndexRequest(index);
            request.source(jsonObject, XContentType.JSON);
            bulkProcessor.add(request);
            return bulkProcessor;
        } catch (Exception exception) {
            log.error("");
        }
        return null;
    }

    public SearchHits findEs(String key, String value, String orderKey, String index) throws IOException {
        BoolQueryBuilder boolQueryBuilder = new BoolQueryBuilder();
        SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();
        sourceBuilder.timeout(new TimeValue(2, TimeUnit.SECONDS));
        boolQueryBuilder.must(QueryBuilders.termQuery(key, value));
        sourceBuilder.query(boolQueryBuilder);
        sourceBuilder.sort(new FieldSortBuilder(orderKey).order(SortOrder.DESC));
        SearchRequest searchRequest = new SearchRequest(index);
        searchRequest.source(sourceBuilder);
        SearchResponse searchResponse = restHighLevelClient.search(searchRequest, RequestOptions.DEFAULT);
        return searchResponse.getHits();

    }

    public UpdateResponse updateEs(String id, String index, JSONObject jsonObject) {
        UpdateRequest updateRequest = new UpdateRequest(index, id);
        updateRequest.retryOnConflict(3);
        updateRequest.doc(jsonObject.toJSONString(), XContentType.JSON);
        UpdateResponse updateResponse = null;
        log.info(">>>>>>>>>>>>>");
        try {
            updateResponse = restHighLevelClient.update(updateRequest, RequestOptions.DEFAULT);
        } catch (IOException e) {
            e.printStackTrace();
        }
        return updateResponse;
    }

    public String initCurrentIndex(String index) {
        synchronized (this) {
            String todayIndex = index + "-" + dateFormat.format(new Date());
            if (todayIndex.equals(currentIndex)) {
                return currentIndex;
            }

            currentIndex = todayIndex;
            try {
                if (!existsIndex(currentIndex)) {
                    CreateIndexRequest createIndexRequest = new CreateIndexRequest(currentIndex);
                    createIndexRequest.settings(Settings.builder().put("index.number_of_shards", 3).
                            put("index.number_of_replicas", 2));

                    XContentBuilder builder = XContentFactory.jsonBuilder();
                    builder.startObject();
                    {
                        builder.startObject("properties");
                        {
                            builder.startObject("testName").field("type", "keyword").endObject();
                            builder.startObject("log").field("type", "text").endObject();
                        }
                        builder.endObject();
                    }
                    builder.endObject();
                    createIndexRequest.mapping(builder);
                    restHighLevelClient.indices().create(createIndexRequest, RequestOptions.DEFAULT);

                }
            } catch (IOException e) {
                e.printStackTrace();
            }
        }

        return currentIndex;
    }
    
}

package com.wht.test.es;

import lombok.extern.slf4j.Slf4j;
import org.elasticsearch.action.bulk.BulkProcessor;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.bulk.BulkResponse;

/**
 * desc
 *
 * @Author 红尘过客
 * @DateTime 2023-08-02 18:58:16
 */
@Slf4j
public class ElasticListener implements BulkProcessor.Listener {
    @Override
    public void beforeBulk(long l, BulkRequest bulkRequest) {
        log.info(">>>>>>>>>>>>>>>");
    }

    @Override
    public void afterBulk(long l, BulkRequest bulkRequest, BulkResponse bulkResponse) {
        boolean hasFailures = bulkResponse.hasFailures();
        if (hasFailures) {
            String buildFailureMessage = bulkResponse.buildFailureMessage();
            log.error("--try insert {} no error --", bulkRequest.numberOfActions());
            log.error("-- error message --{}", buildFailureMessage);
        }
    }

    @Override
    public void afterBulk(long l, BulkRequest bulkRequest, Throwable throwable) {
        log.error("--try insert {} no error --", throwable);
    }
}

5、验证示例--client

package com.wht.test.es;

import com.alibaba.fastjson.JSONObject;
import lombok.extern.slf4j.Slf4j;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.SearchHits;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.util.Map;

/**
 * desc
 *
 * @Author 红尘过客
 * @DateTime 2023-08-02 16:27:05
 */
@Slf4j
@Component
public class ElasticsearchClient {

    private static final String TEST_INDEX = "test_es_index";

    @Autowired
    private ElasticService elasticService;

    @Scheduled(cron = "*/10 * * * * * ")
    public void bizHandler() {
        JSONObject jsonObject = new JSONObject();
        long time = System.currentTimeMillis();
        String testName = "test_name_" + time;
        jsonObject.put("testName", testName);
        jsonObject.put("time", time);
        jsonObject.put("log", time + "-样例日志。。。。。。。。。。。。。。。。。。。。。。。。。");
        String initIndex = elasticService.initCurrentIndex(TEST_INDEX);
        elasticService.writeEs(initIndex, jsonObject);

        try {
            Thread.sleep(1000);
            SearchHits hits = elasticService.findEs("testName", testName, "time", initIndex);
            if (hits.getTotalHits().value > 0) {
                for (SearchHit hit : hits.getHits()) {
                    Map<String, Object> result = hit.getSourceAsMap();
                    Object logString = result.get("log");
                    if (logString != null) {
                        log.info(">>>>>>>>>>>>>>{}", logString.toString());
                    }
                }
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (IOException ioException) {
            ioException.printStackTrace();
        }
    }
}