ES查询客户端初始化(RestHighLevelClient)

ElasticSearchProperties  配置类，可以配置在中心中

public class ElasticSearchProperties {

    /**
     * 连接超时时间(毫秒)
     */
    private Integer connectTimeout = 3000;

    /**
     * socket 超时时间
     */
    private Integer socketTimeout = 30000;

    /**
     * 连接请求超时时间
     */
    private Integer connectionRequestTimeout = 500;

    /**
     * keepAlive策略时间
     */
    private Integer keepAliveStrategy = 60 * 1000;
}

EsClientFactory  es客户端工厂

/**
 * @author yangwenxin
 * @Date 2022/5/13
 * @Description
 */
@Slf4j
public class EsClientFactory {

    /**
     * 创建es客户端
     *
     * @param properties 配置信息
     * @param address    地址, 示例：127.0.0.1:9201,127.0.0.1:9202,127.0.0.1:9203
     * @param userName   用户名
     * @param password   密码
     * @return
     */
    public static RestHighLevelClient createRestHighLevelClient(ElasticSearchProperties properties, String address,
                                                                String userName, String password) {
        //ElasticSearch 连接地址地址
        HttpHost[] httpHosts = getElasticSearchHttpHosts(address);

        RestClientBuilder restClientBuilder = RestClient.builder(httpHosts).setRequestConfigCallback(requestConfigBuilder -> {
            //设置连接超时时间
            requestConfigBuilder.setConnectTimeout(properties.getConnectTimeout());
            requestConfigBuilder.setSocketTimeout(properties.getSocketTimeout());
            requestConfigBuilder.setConnectionRequestTimeout(properties.getConnectionRequestTimeout());
            return requestConfigBuilder;
        }).setHttpClientConfigCallback(httpClientBuilder -> {
            httpClientBuilder.disableAuthCaching();
            httpClientBuilder.setKeepAliveStrategy((response, context) -> properties.getKeepAliveStrategy());
            //设置账密
            return getHttpAsyncClientBuilder(httpClientBuilder, userName, password);
        });
        log.info("init esClient success, address : {}", address);
        return new RestHighLevelClient(restClientBuilder);
    }


    /**
     * ElasticSearch 连接地址
     * 多个逗号分隔
     * 示例：127.0.0.1:9201,127.0.0.1:9202,127.0.0.1:9203
     */
    private static HttpHost[] getElasticSearchHttpHosts(String address) {
        String[] hosts = address.split(",");
        HttpHost[] httpHosts = new HttpHost[hosts.length];
        for (int i = 0; i < httpHosts.length; i++) {
            String host = hosts[i];
            host = host.replaceAll("http://", "").replaceAll("https://", "");
            httpHosts[i] = new HttpHost(host.split(":")[0], Integer.parseInt(host.split(":")[1]), "http");
        }
        return httpHosts;
    }

    private static HttpAsyncClientBuilder getHttpAsyncClientBuilder(HttpAsyncClientBuilder httpClientBuilder, String username, String password) {
        if (StringUtils.isBlank(username) || StringUtils.isEmpty(password)) {
            return httpClientBuilder;
        }
        //账密设置
        CredentialsProvider credentialsProvider = new BasicCredentialsProvider();
        //es账号密码（一般使用,用户elastic）
        credentialsProvider.setCredentials(AuthScope.ANY, new UsernamePasswordCredentials(username, password));
        httpClientBuilder.setDefaultCredentialsProvider(credentialsProvider);
        return httpClientBuilder;
    }
}

ElasticSearchClientService  es客户端初始化，从数据库中读取信息初始化客户端。其中DataSourceConfig就是配置了一些es链接的信息,账号密码地址等。

/**
 * @author yangwenxin
 * @Date 2022/5/13
 * @Description
 */
@Service
@Slf4j
public class ElasticSearchClientService {

    @Autowired
    private ElasticSearchProperties elasticSearchProperties;

    private Map<String, EsConfig> esClientMap = new HashMap<>();

    @PostConstruct
    public void init() {
        initEsClientMap();
    }

    /**
     * 初始化es数据源客户端到缓存
     */
    public void initEsClientMap() {
        // TODO 数据库查询数据源信息 自己项目中查询数据库方法，或者将ES信息配置在配置中心
        List<DataSourceConfig> dataSourceConfigs = Lists.newArrayList();
        if (CollectionUtils.isEmpty(dataSourceConfigs)) {
            return;
        }
        // address相同认为是一组  es数据源初始化address包含账号密码
        Map<String, List<DataSourceConfig>> dataSourceConfigMaps = dataSourceConfigs.stream().collect(
                Collectors.groupingBy(
                        DataSourceConfig::getAddress
                ));

        dataSourceConfigMaps.forEach((key, value) -> {
            DataSourceConfig dataSourceConfig = value.get(0);
            RestHighLevelClient restHighLevelClient = EsClientFactory.createRestHighLevelClient(
                    elasticSearchProperties,
                    dataSourceConfig.getAddress(),
                    dataSourceConfig.getUsername(),
                    dataSourceConfig.getPassword()
            );
            value.forEach(v -> {
                EsConfig esConfig = new EsConfig(restHighLevelClient, v.getInstance());
                esClientMap.put(v.getDataSourceNo(), esConfig);
            });
        });
    }

    @PreDestroy
    public void closeEsClient() {
        if (CollectionUtils.isEmpty(esClientMap)) {
            return;
        }
        esClientMap.forEach((key, value) -> {
            try {
                RestHighLevelClient client = value.getRestHighLevelClient();
                if (client != null) {
                    client.close();
                }
            } catch (Exception e) {
                log.error("close restHighLevelClient error! key: {},errMsg : {}", key, e.getMessage());
            }
        });
    }

    public EsConfig getEsConfigByNo(String dateSourceNo) {
        return esClientMap.get(dateSourceNo);
    }

    public synchronized void refreshEsDataSource(){
        initEsClientMap();
        log.info("refresh esDataSource success!");
    }


    @Data
    @AllArgsConstructor
    @NoArgsConstructor
    public static class EsConfig {
        /**
         * es客户端
         */
        private RestHighLevelClient restHighLevelClient;
        /**
         * 索引名称
         */
        private String dataSourceName;
    }
}

EsClientUtil 查询工具类

@Slf4j
public class EsClientUtil {

    private static final String DEFAULT_TYPE = "_doc";

    /**
     * query string 查询， queryString语法。语法自行百度
     * @param request
     * @return
     * @throws IOException
     */
    public static LogDataResp queryStringQuery(RestHighLevelClient client, QueryLogDataReq request) {
        String indexName = request.getQueryIndex();
        try {
            SearchRequest searchRequest = new SearchRequest(indexName);
            SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();

            // 查询语句
            BoolQueryBuilder must =
                    QueryBuilders.boolQuery().must(
                            QueryBuilders.queryStringQuery(request.getQueryScript())
                    );

            // 时间范围  TODO 时区
            if(StringUtils.isNotBlank(request.getTimeFieldName()) && request.getTimeFrameMinutes() != null){
                RangeQueryBuilder rangeQueryBuilder = QueryBuilders.rangeQuery(request.getTimeFieldName())
                        .gte("now-" + request.getTimeFrameMinutes() + "m")
                        .lte("now")
                        // es时区默认是UTC
                        .timeZone("UTC");
                must.must(rangeQueryBuilder);
                // 根据时间降序排列
                searchSourceBuilder.sort(request.getTimeFieldName(), SortOrder.DESC);
            }
            searchSourceBuilder.query(must).from(0).size(request.getQuerySize());
            searchRequest.source(searchSourceBuilder);
            SearchResponse searchResponse = client.search(searchRequest, RequestOptions.DEFAULT);
            // 返回值解析
            return buildLogDataResponse(searchResponse,null);

        } catch (Exception e) {
            String errMsg = MessageFormat.format("EsClientUtil#queryStringQuery failed, params={0}", request);
            log.error(errMsg, e);
            return LogDataResp.buildFailed(e.getMessage());
        }
    }

    /**
     * 查询所有的索引
     * @param client
     * @return
     * @throws IOException
     */
    public static List<String> queryAllIndexAlias(RestHighLevelClient client) {
        try {
            GetAliasesRequest getAliasesRequest = new GetAliasesRequest();
            GetAliasesResponse getAliasesResponse = client.indices().getAlias(getAliasesRequest, RequestOptions.DEFAULT);
            Map<String, Set<AliasMetaData>> aliases = getAliasesResponse.getAliases();
            if(!aliases.isEmpty()){
                return new ArrayList<>(aliases.keySet());
            }
            return Arrays.asList();
        } catch (Exception e) {
            String errMsg = "EsClientUtil#queryAllIndexAlias error";
            log.error("errMsg" , e);
            throw new RuntimeException(errMsg, e);
        }
    }

    /**
     * 向索引插入单条记录
     * @param client
     * @param indexName    索引
     * @param docId        文档ID
     * @param docSource    文档 JSON格式
     * @return
     * @throws IOException
     */
    public static boolean insertDoc(RestHighLevelClient client, String indexName, String docId, Object docSource) {
        try {
            IndexRequest request = new IndexRequest(indexName,DEFAULT_TYPE);
            if (null != docId) {
                request.id(docId);
            }
            request.source(JSON.toJSONString(docSource), XContentType.JSON);
            IndexResponse indexResponse = client.index(request, RequestOptions.DEFAULT);
            if ((indexResponse.status() == RestStatus.CREATED) || (indexResponse.status() == RestStatus.OK)) {
                return true;
            }
            log.error("insertDoc fail status:{} => {}", indexResponse.status(), indexResponse);
            return false;
        } catch (Exception e) {
            String errMsg =
                    MessageFormat.format("EsClientUtil#insertDoc failed, indexName={0},docId={1}, docSource={2}",
                            indexName,docId,docSource);
            log.error(errMsg, e);
            throw new RuntimeException(errMsg, e);
        }
    }

    /**
     * 向索引插入批量记录  不指定id
     * @param client
     * @param indexName   索引名称
     * @param docSources  数据 JSON格式
     * @return
     */
    public static boolean insertDocBulk(RestHighLevelClient client,String indexName, List<Object> docSources) {
        try {
            if (CollectionUtils.isEmpty(docSources)) {
                return false;
            }
            BulkRequest bulkRequest = new BulkRequest();
            docSources.forEach(docSource -> {
                IndexRequest request = new IndexRequest(indexName,DEFAULT_TYPE);
                request.source(JSON.toJSONString(docSource), XContentType.JSON);
                bulkRequest.add(request);
            });
            BulkResponse bulkResponse = client.bulk(bulkRequest, RequestOptions.DEFAULT);
            if (!bulkResponse.hasFailures()) {
                return true;
            }
            log.error("insertDocBulk fail status:{}, errorMsg:{}", bulkResponse.status(),
                    bulkResponse.buildFailureMessage());
            return false;
        } catch (Exception e) {
            String errMsg =
                    MessageFormat.format("EsClientUtil#insertDocBulk failed, indexName={0}",
                            indexName);
            log.error(errMsg, e);
            throw new RuntimeException(errMsg, e);
        }
    }

    /**
     * 向索引插入批量记录  指定id
     * @param client
     * @param indexName      索引名称
     * @param docSourceMap   key->Id value->docSource JSON格式
     * @return
     */
    public static boolean insertDocBulk(RestHighLevelClient client,String indexName, Map<String,Object> docSourceMap) {
        try{
            if (CollectionUtils.isEmpty(docSourceMap)) {
                return false;
            }
            BulkRequest bulkRequest = new BulkRequest();
            docSourceMap.forEach((key,value) -> {
                IndexRequest request = new IndexRequest(indexName,DEFAULT_TYPE);
                request.id(key);
                request.source(JSON.toJSONString(value), XContentType.JSON);
                bulkRequest.add(request);
            });
            BulkResponse bulkResponse = client.bulk(bulkRequest, RequestOptions.DEFAULT);
            if (!bulkResponse.hasFailures()) {
                return true;
            }
            log.error("insertDocBulk fail status:{}, errorMsg:{}", bulkResponse.status(),
                    bulkResponse.buildFailureMessage());
            return false;
        } catch (Exception e) {
            String errMsg =
                    MessageFormat.format("EsClientUtil#insertDocBulk failed, indexName={0}",
                            indexName);
            log.error(errMsg, e);
            throw new RuntimeException(errMsg, e);
        }
    }

    /**
     * 根据ID 删除文档
     * @param client
     * @param indexName  索引名称
     * @param docId
     * @return
     */
    public static boolean deleteDocById(RestHighLevelClient client, String indexName, String docId) {
        try{
            DeleteRequest deleteRequest = new DeleteRequest(indexName,DEFAULT_TYPE,docId);
            DeleteResponse delete = client.delete(deleteRequest, RequestOptions.DEFAULT);
            int total = delete.getShardInfo().getTotal();
            return total > 0;
        } catch (Exception e) {
            String errMsg =
                    MessageFormat.format("EsClientUtil#deleteDocById failed, indexName={0},docId={1} ",
                            indexName, docId);
            log.error(errMsg, e);
            throw new RuntimeException(errMsg, e);
        }
    }

    /**
     * 根据ID更新文档
     * @param client
     * @param indexName   索引名称
     * @param docId       文档ID
     * @param docSource   文档数据 JSON格式
     * @return
     */
    public static boolean updateDocById(RestHighLevelClient client, String indexName, String docId, Object docSource) {
        try{
            UpdateRequest updateRequest = new UpdateRequest(indexName,DEFAULT_TYPE,docId);
            updateRequest.doc(JSON.toJSONString(docSource),XContentType.JSON);
            // 执行更新
            UpdateResponse update = client.update(updateRequest, RequestOptions.DEFAULT);
            int total = update.getShardInfo().getTotal();
            return total > 0;
        } catch (Exception e) {
            String errMsg =
                    MessageFormat.format("EsClientUtil#updateDoc failed, indexName={0},docSource={1} ",
                            indexName, docSource);
            log.error(errMsg, e);
            throw new RuntimeException(errMsg, e);
        }
    }

    /**
     * 根据ID查询
     * @param client
     * @param indexName
     * @param docId
     * @return
     */
    public static <T> T queryDocById(RestHighLevelClient client, String indexName, String docId, Class<T> clazz) {
        try {
            GetRequest request = new GetRequest(indexName,DEFAULT_TYPE,docId);
            GetResponse response = client.get(request, RequestOptions.DEFAULT);
            if(response.isSourceEmpty()){
                return null;
            }
            return JSON.parseObject(response.getSourceAsString(), clazz);
        } catch (Exception e) {
            String errMsg =
                    MessageFormat.format("EsClientUtil#queryDocById failed, indexName={0},docId={1} ",
                            indexName, docId);
            log.error(errMsg, e);
            throw new RuntimeException(errMsg, e);
        }
    }

    /**
     * 根据条件查询分页
     * 注：from-size分页 可以随机跳转页面，在深度分页的情况下，效率是非常低的。深度分页参考search_after 不支持随机跳转，只支持
     * @param index    索引
     * @param pageNo   页码（第几页）
     * @param pageSize 页容量- Elasticsearch默认配置单次最大限制10000
     */
    public static <T> List<T> searchPageByIndex(RestHighLevelClient client, String index, SearchSourceBuilder searchSourceBuilder, Integer pageNo, Integer pageSize, Class<T> clazz) {
        if(searchSourceBuilder == null){
            searchSourceBuilder = new SearchSourceBuilder();
        }
        searchSourceBuilder.from((pageNo-1)*pageSize);
        searchSourceBuilder.size(pageSize);
        return searchByQuery(client, index, searchSourceBuilder, clazz);
    }

    /**
     * 条件查询
     *
     * @param indexName         索引
     * @param sourceBuilder 条件查询构建起
     * @param <T>           数据类型
     * @return T 类型的集合
     */
    public static <T> List<T> searchByQuery(RestHighLevelClient client, String indexName, SearchSourceBuilder sourceBuilder, Class<T> clazz) {
        try {
            // 构建查询请求
            SearchRequest searchRequest = new SearchRequest(indexName).source(sourceBuilder);
            // 获取返回值
            SearchResponse response = client.search(searchRequest, RequestOptions.DEFAULT);
            SearchHit[] hits = response.getHits().getHits();
            if (null == hits || hits.length == 0) {
                return new ArrayList<>(0);
            }
            List<T> resultList = new ArrayList<>(hits.length);
            for (SearchHit hit : hits) {
                resultList.add(JSON.parseObject(hit.getSourceAsString(), clazz));
            }
            return resultList;
        } catch (Exception e) {
            String errMsg =
                    MessageFormat.format("EsClientUtil#searchByQuery failed, indexName={0}, sourceBuilder={1} ",
                            sourceBuilder);
            log.error(errMsg, e);
            throw new RuntimeException(errMsg, e);
        }
    }


    /**
     * 结果解析构建返回结果
     * @param searchResponse
     * @param aggName
     * @return
     */
    private static LogDataResp buildLogDataResponse(SearchResponse searchResponse, String aggName){
        LogDataResp logDataResponse = new LogDataResp();
        if(searchResponse.status().getStatus() != RestStatus.OK.getStatus()){
            logDataResponse.setSucceeded(Boolean.FALSE);
            return logDataResponse;
        }

        long totalHits = searchResponse.getHits().getTotalHits();
        logDataResponse.setTotalHit(totalHits);
        SearchHit[] hits = searchResponse.getHits().getHits();
        if(hits.length > 0){
            List<Map> hitSourceList = new ArrayList<>();
            for(SearchHit hit : hits){
                hitSourceList.add(hit.getSourceAsMap());
            }
            logDataResponse.setHitSourceList(hitSourceList);
        }

        if(aggName != null){
            Aggregations aggregations = searchResponse.getAggregations();
            Terms aggregation = (Terms)aggregations.get(aggName);
            List<? extends Terms.Bucket> buckets = aggregation.getBuckets();
            Map map = Maps.newHashMap();
            List<Map> aggregationBucketList = new ArrayList<>();
            for (Terms.Bucket bucket : buckets) {
                map.put(bucket.getKeyAsString(),bucket.getDocCount());
            }
            aggregationBucketList.add(map);
            logDataResponse.setAggregationBucketList(aggregationBucketList);
        }
        return logDataResponse;
    }


}

另外，如果说把es当成一个数据库使用，可以看下开源项目easy-es，操作更方便。文档地址:快速开始 | Easy-Es文章来源地址https://www.toymoban.com/news/detail-505683.html

到了这里，关于ES查询客户端初始化(RestHighLevelClient)的文章就介绍完了。如果您还想了解更多内容，请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章，希望大家以后多多支持TOY模板网！