ElasticSearch创建索引，删除索引，判断 index 是否存在，根据 id 删除指定索引中的文档，
根据 id 更新指定索引中的文档，根据 id 更新指定索引中的文档，根据某字段的 k-v 更新索引中的文档，
添加文档 手动指定id，简单模糊匹配 默认分页为 0,10， term 查询 精准匹配，term 查询 精准匹配,返回列表，term 查询 ,返回列表，查询列表并高亮，批量导入，获取分词结果，返回聚合结果。

1、在application.yml配置如下内容

#数据源配置
spring:
  data:
    elasticsearch:
      client:
        reactive:
          endpoints: 127.0.0.1:9200
          connection-timeout: 10s
          username: xxxx
          password: xxxx
      #ES 持久化存储
      repositories:
        enabled: true
  elasticsearch:
    rest:
      # 要连接的ES客户端Rest Uri 多个逗号分隔
      uris: http://127.0.0.1:9200
      connection-timeout: 10s
      read-timeout: 30s
      username: xxxx
      password: xxxx

2、RestHighLevelClientUtils工具类

package cn.itbluebox.itblueboxesjd.utils;

import cn.hutool.core.util.StrUtil;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import org.apache.http.util.EntityUtils;
import org.elasticsearch.action.admin.indices.delete.DeleteIndexRequest;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.delete.DeleteRequest;
import org.elasticsearch.action.delete.DeleteResponse;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.support.master.AcknowledgedResponse;
import org.elasticsearch.action.update.UpdateRequest;
import org.elasticsearch.action.update.UpdateResponse;
import org.elasticsearch.client.Request;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.Response;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.client.indices.CreateIndexRequest;
import org.elasticsearch.client.indices.CreateIndexResponse;
import org.elasticsearch.client.indices.GetIndexRequest;
import org.elasticsearch.common.text.Text;
import org.elasticsearch.core.TimeValue;
import org.elasticsearch.index.query.BoolQueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.index.query.TermQueryBuilder;
import org.elasticsearch.index.reindex.BulkByScrollResponse;
import org.elasticsearch.index.reindex.DeleteByQueryRequest;
import org.elasticsearch.index.reindex.UpdateByQueryRequest;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.SearchHits;
import org.elasticsearch.search.aggregations.AggregationBuilders;
import org.elasticsearch.search.aggregations.bucket.terms.Terms;
import org.elasticsearch.search.aggregations.bucket.terms.TermsAggregationBuilder;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.elasticsearch.search.fetch.subphase.highlight.HighlightField;
import org.elasticsearch.xcontent.XContentType;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.io.IOException;
import java.net.URLEncoder;
import java.text.SimpleDateFormat;
import java.util.*;

//DeleteRequest GetRequest UpdateRequest 都是根据 id 操作文档

/**
 * @author anqi
 */
@Component
public class RestHighLevelClientUtils {
    
    

    @Resource
    private RestHighLevelClient client;

    /**
     * 创建索引
     * @param indexName
     * @param settings
     * @param mapping
     * @return
     * @throws IOException
     */
    public CreateIndexResponse createIndex(String indexName, String settings, String mapping) throws IOException{
    
    
        CreateIndexRequest request = new CreateIndexRequest(indexName);
        if (null != settings && !"".equals(settings)) {
    
    
            request.settings(settings, XContentType.JSON);
        }
        if (null != mapping && !"".equals(mapping)) {
    
    
            request.mapping(mapping, XContentType.JSON);
        }
        return client.indices().create(request, RequestOptions.DEFAULT);
    }

    /**
     * 删除索引
     * @param indexNames
     * @return
     * @throws IOException
     */
    public AcknowledgedResponse deleteIndex(String ... indexNames) throws IOException{
    
    
        DeleteIndexRequest request = new DeleteIndexRequest(indexNames);
        return client.indices().delete(request, RequestOptions.DEFAULT);
    }


    /**
     * 判断 index 是否存在
     * @param indexName
     * @return
     * @throws IOException
     */
    public boolean indexExists(String indexName) throws IOException {
    
    
        GetIndexRequest request = new GetIndexRequest(indexName);
        return client.indices().exists(request, RequestOptions.DEFAULT);
    }

    /**
     * 根据 id 删除指定索引中的文档
     * @param indexName
     * @param id
     * @return
     * @throws IOException
     */
    public DeleteResponse deleteDoc(String indexName, String id) throws IOException{
    
    
        DeleteRequest request = new DeleteRequest(indexName, id);
        return client.delete(request, RequestOptions.DEFAULT);
    }

    /**
     * 根据 id 更新指定索引中的文档
     * @param indexName
     * @param id
     * @return
     * @throws IOException
     */
    public UpdateResponse updateDoc(String indexName, String id, String updateJson) throws IOException{
    
    
        UpdateRequest request = new UpdateRequest(indexName, id);
        request.doc(XContentType.JSON, updateJson);
        return client.update(request, RequestOptions.DEFAULT);
    }

    /**
     * 根据 id 更新指定索引中的文档
     * @param indexName
     * @param id
     * @return
     * @throws IOException
     */
    public UpdateResponse updateDoc(String indexName, String id, Map<String,Object> updateMap) throws IOException{
    
    
        UpdateRequest request = new UpdateRequest(indexName, id);
        request.doc(updateMap);

        return client.update(request, RequestOptions.DEFAULT);
    }

    private String generateCurrentData() {
    
    

        Long timeStamp = System.currentTimeMillis();  //获取当前时间戳
        SimpleDateFormat sdf=new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
        String sd = sdf.format(new Date(Long.parseLong(String.valueOf(timeStamp))));      // 时间戳转换成时间
        return sd;
    }

    /**
     * 根据某字段的 k-v 更新索引中的文档
     * @param fieldName
     * @param value
     * @param indexName
     * @throws IOException
     */
    public void updateByQuery(String fieldName, String value, String ... indexName) throws IOException {
    
    
        UpdateByQueryRequest request = new UpdateByQueryRequest(indexName);
        //单次处理文档数量
        request.setBatchSize(100)
                .setQuery(new TermQueryBuilder(fieldName, value))
                .setTimeout(TimeValue.timeValueMinutes(2));
        client.updateByQuery(request, RequestOptions.DEFAULT);
    }

    /**
     * 添加文档 手动指定id
     * @param indexName
     * @param id
     * @param source
     * @return
     * @throws IOException
     */
    public IndexResponse addDoc(String indexName, String id, String source) throws IOException{
    
    
        IndexRequest request = new IndexRequest(indexName);
        if (null != id) {
    
    
            request.id(id);
        }
        request.source(source, XContentType.JSON);
        return client.index(request, RequestOptions.DEFAULT);
    }

    /**
     * 添加文档 使用自动id
     * @param indexName
     * @param source
     * @return
     * @throws IOException
     */
    public IndexResponse addDoc(String indexName, String source) throws IOException{
    
    
        return addDoc(indexName, null, source);
    }

    /**
     * 简单模糊匹配 默认分页为 0,10
     * @param field
     * @param key
     * @param page
     * @param size
     * @param indexNames
     * @return
     * @throws IOException
     */
    public SearchResponse search(String field, String key, int page, int size, String ... indexNames) throws IOException{
    
    
        SearchRequest request = new SearchRequest(indexNames);
        SearchSourceBuilder builder = new SearchSourceBuilder();

        BoolQueryBuilder boolQueryBuilder = QueryBuilders.boolQuery();
        if (StrUtil.isEmpty(key)){
    
    
            builder.query(QueryBuilders.matchAllQuery())
                    .from((page-1) * size)
                    .size(size);
        }else{
    
    
            BoolQueryBuilder boolQuery = QueryBuilders.boolQuery();
            boolQuery.should(QueryBuilders.matchQuery(field, key));
            boolQueryBuilder.must(boolQuery);
            builder.query(boolQueryBuilder)
                    .from((page-1) * size)
                    .size(size);
        }

        request.source(builder);
        return client.search(request, RequestOptions.DEFAULT);
    }

    /**
     * term 查询 精准匹配
     * @param field
     * @param key
     * @param page
     * @param size
     * @param indexNames
     * @return
     * @throws IOException
     */
    public SearchResponse termSearch(String field, String key, int page, int size, String ... indexNames) throws IOException{
    
    
        SearchRequest request = new SearchRequest(indexNames);
        SearchSourceBuilder builder = new SearchSourceBuilder();
        builder.query(QueryBuilders.termsQuery(field, key))
                .from(page)
                .size(size);
        request.source(builder);
        return client.search(request, RequestOptions.DEFAULT);
    }

    /**
     * term 查询 精准匹配,返回列表
     * @param field
     * @param key
     * @param page
     * @param size
     * @param indexNames
     * @return
     * @throws IOException
     */
    public List<Map<String, Object>> termSearchList(String field, String key, int page, int size, String ... indexNames){
    
    
        try {
    
    

            SearchRequest request = new SearchRequest(indexNames);
            SearchSourceBuilder builder = new SearchSourceBuilder();
            builder.query(QueryBuilders.termsQuery(field, key))
                    .from(page)
                    .size(size);
            request.source(builder);
            SearchResponse searchResponse = client.search(request, RequestOptions.DEFAULT);
            SearchHits hits = searchResponse.getHits();
            // 获取检索结果总数
            List<Map<String, Object>> resultList=new ArrayList<>();
            for (SearchHit hit : hits.getHits()) {
    
    
                Map<String, Object> sourceAsMap = hit.getSourceAsMap();
                resultList.add(sourceAsMap);
            }
            return resultList;
        }catch (Exception e){
    
    
            e.printStackTrace();
            throw new RuntimeException("查询失败");
        }

    }

    /**
     * term 查询 ,返回列表
     * @param indexNames
     * @return
     * @throws IOException
     */
    public List<Map<String, Object>> SearchList(SearchSourceBuilder builder, String ...indexNames){
    
    
        try {
    
    
            SearchRequest request = new SearchRequest(indexNames);
            request.source(builder);
            SearchResponse searchResponse = client.search(request, RequestOptions.DEFAULT);
            SearchHits hits = searchResponse.getHits();
            // 获取检索结果总数
            List<Map<String, Object>> resultList=new ArrayList<>();
            for (SearchHit hit : hits.getHits()) {
    
    
                Map<String, Object> sourceAsMap = hit.getSourceAsMap();
                sourceAsMap.put("index",hit.getIndex());
                resultList.add(sourceAsMap);
            }
            return resultList;
        }catch (Exception e){
    
    
            e.printStackTrace();
            throw new RuntimeException("查询失败");
        }

    }

    /**
     * 查询列表并高亮
     * @param builder
     * @param indexNames
     * @return
     */
    public List<Map<String, Object>> SearchListWithHigh(SearchSourceBuilder builder, String ...indexNames){
    
    
        try {
    
    
            SearchRequest request = new SearchRequest(indexNames);
            request.source(builder);
            SearchResponse searchResponse = client.search(request, RequestOptions.DEFAULT);
            SearchHits hits = searchResponse.getHits();
            // 获取检索结果总数
            List<Map<String, Object>> resultList=new ArrayList<>();
            for (SearchHit hit : hits.getHits()) {
    
    
                Map<String, Object> sourceAsMap = hit.getSourceAsMap();
                Map<String, HighlightField> highlightFieldMap = hit.getHighlightFields();
                // 获取高亮
                for (Map.Entry<String, HighlightField> entry : highlightFieldMap.entrySet()) {
    
    
                    Map<String, String> map = new HashMap<>();
                    Text[] texts = highlightFieldMap.get(entry.getKey()).getFragments();
                    String title = "";
                    for (Text text : texts) {
    
    
                        title += text;
                    }
                    map.put(entry.getKey(), title);
                    sourceAsMap.put(entry.getKey(), title);
                }
                sourceAsMap.put("index",hit.getIndex());
                resultList.add(sourceAsMap);
            }
            return resultList;
        }catch (Exception e){
    
    
            e.printStackTrace();
            throw new RuntimeException("查询失败");
        }

    }


    /**
     * 批量导入
     * @param indexName
     * @param isAutoId 使用自动id 还是使用传入对象的id
     * @param source
     * @return
     * @throws IOException
     */
    public BulkResponse importAll(String indexName, boolean isAutoId, String  source) throws IOException{
    
    
        if (0 == source.length()){
    
    
            //todo 抛出异常 导入数据为空
        }
        BulkRequest request = new BulkRequest();

        JSONArray array = JSON.parseArray(source);

        //todo 识别json数组
        if (isAutoId) {
    
    
            for (Object s : array) {
    
    
                request.add(new IndexRequest(indexName).source(s, XContentType.JSON));
            }
        } else {
    
    
            for (Object s : array) {
    
    
                request.add(new IndexRequest(indexName).id(JSONObject.parseObject(s.toString()).getString("id")).source(s, XContentType.JSON));
            }
        }
        return client.bulk(request, RequestOptions.DEFAULT);
    }

    /**
     * 获取分词结果
     * @param text
     * @return
     */
    public List<String> getAnalyze(String text){
    
    
        try{
    
    
            List<String> list = new ArrayList<>();
            Request request = new Request("GET", URLEncoder.encode("_analyze"));
            JSONObject entity = new JSONObject();
            entity.put("analyzer", "hanlp");
            entity.put("text", text);
            request.setJsonEntity(entity.toJSONString());
            Response response = client.getLowLevelClient().performRequest(request);
            JSONObject tokens = JSONObject.parseObject(EntityUtils.toString(response.getEntity()));
            JSONArray arrays = tokens.getJSONArray("tokens");
            for (int i = 0; i < arrays.size(); i++)
            {
    
    
                JSONObject obj = JSON.parseObject(arrays.getString(i));
                list.add(obj.getString("token"));
            }

            return list;
        }catch (Exception e){
    
    
            e.printStackTrace();
            throw new RuntimeException("获取分词结果失败");
        }
    }

    /**
     * 返回聚合结果
     * @param groupfield
     * @param indexNames
     * @return
     */
	public List<String> searchForGroupOne(String groupfield, String ... indexNames){
    
    
		try {
    
    
			SearchRequest request = new SearchRequest(indexNames);
			SearchSourceBuilder builder = new SearchSourceBuilder();
			TermsAggregationBuilder aggregation = AggregationBuilders.terms(groupfield).field(groupfield).size(100);
			builder.aggregation(aggregation);
			request.source(builder);
			SearchResponse searchResponse = client.search(request, RequestOptions.DEFAULT);
			//获取政策的聚合结果
			List<String> agglist=new ArrayList<>();
			//获取主分类-年份
			Terms terms = searchResponse.getAggregations().get(groupfield);
			if (terms != null) {
    
    
				for (Terms.Bucket bucket : terms.getBuckets()) {
    
    
					String the = (String) bucket.getKey();
					if (StrUtil.isNotBlank(the)) {
    
    
						agglist.add(the);
					}
				}
			}
			return agglist;
		}catch (Exception e){
    
    
			e.printStackTrace();
			throw new RuntimeException("查询失败");
		}
	}

	public List<String> searchForGroupOneAll(String groupfield, String ... indexNames){
    
    
		try {
    
    
			SearchRequest request = new SearchRequest(indexNames);
			SearchSourceBuilder builder = new SearchSourceBuilder();
			TermsAggregationBuilder aggregation = AggregationBuilders.terms(groupfield).field(groupfield);
			builder.aggregation(aggregation);
			request.source(builder);
			SearchResponse searchResponse = client.search(request, RequestOptions.DEFAULT);
			//获取政策的聚合结果
			List<String> agglist=new ArrayList<>();
			//获取主分类-年份
			Terms terms = searchResponse.getAggregations().get(groupfield);
			if (terms != null) {
    
    
				for (Terms.Bucket bucket : terms.getBuckets()) {
    
    
					String the = (String) bucket.getKey();
					if (StrUtil.isNotBlank(the)) {
    
    
						agglist.add(the);
					}
				}
			}
			return agglist;
		}catch (Exception e){
    
    
			e.printStackTrace();
			throw new RuntimeException("查询失败");
		}
	}
	public BulkByScrollResponse deleteByQuery(DeleteByQueryRequest request, RequestOptions aDefault)  {
    
    
		BulkByScrollResponse bulkByScrollResponse = null;
		try {
    
    
			bulkByScrollResponse = client.deleteByQuery(request, aDefault);
		}catch (Exception e){
    
    
			e.printStackTrace();
		}
		return bulkByScrollResponse;
	}
}