Elasticsearch整合及基本操作示例

@Configuration
public class ElasticsearchConfiguration {

    @Value("${elasticsearch.host}")
    private String host;

    @Value("${elasticsearch.port}")
    private int port;

    @Value("${elasticsearch.username}")
    private String username;

    @Value("${elasticsearch.password}")
    private String password;

    @Bean(destroyMethod = "close", name = "client")
    public RestHighLevelClient initRestClient() {
        // 用户认证对象
        final CredentialsProvider credentialsProvider = new BasicCredentialsProvider();
        // 设置账号密码
        credentialsProvider.setCredentials(AuthScope.ANY, new UsernamePasswordCredentials(username, password));
        // 创建rest client对象
        RestClientBuilder builder = RestClient.builder(
                new HttpHost(host, port))
                .setHttpClientConfigCallback(new RestClientBuilder.HttpClientConfigCallback() {
                    @Override
                    public HttpAsyncClientBuilder customizeHttpClient(
                            HttpAsyncClientBuilder httpClientBuilder) {
                        return httpClientBuilder
                                .setDefaultCredentialsProvider(credentialsProvider);
                    }
                });
        return new RestHighLevelClient(builder);
    }
}

@Override
public boolean createIndex(String index) throws IOException {
    CreateIndexRequest createIndexRequest = new CreateIndexRequest(index);
    CreateIndexResponse createIndexResponse = client.indices().create(createIndexRequest, RequestOptions.DEFAULT);
    return createIndexResponse.isAcknowledged();
}

@Override
public String[] queryIndex(String index) throws IOException {
    GetIndexRequest queryIndexRequest = new GetIndexRequest(index);
    GetIndexResponse getIndexResponse = client.indices().get(queryIndexRequest, RequestOptions.DEFAULT);
    return getIndexResponse.getIndices();
}

@Override
public boolean deleteIndex(String index) throws IOException {
    DeleteIndexRequest deleteIndexRequest = new DeleteIndexRequest(index);
    AcknowledgedResponse deleteIndexResponse = client.indices().delete(deleteIndexRequest, RequestOptions.DEFAULT);
    return deleteIndexResponse.isAcknowledged();
}

GetIndexRequest getIndexRequest = new GetIndexRequest (Constant.INDEX);
boolean exists = client.indices().exists(getIndexRequest, RequestOptions.DEFAULT);

public interface Constant {
    String INDEX = "user";
}

@Data
public class UserDocument {
    private String id;
    private String name;
    private String sex;
    private Integer age;
    private String city;
}

@Override
public Boolean createDocument(UserDocument document) throws Exception {
    String id = document.getId();
    IndexRequest indexRequest = new IndexRequest(Constant.INDEX)
            .id(id)
            .source(JSON.toJSONString(document), XContentType.JSON);
    IndexResponse indexResponse = client.index(indexRequest, RequestOptions.DEFAULT);
    return indexResponse.status().equals(RestStatus.CREATED);
}    

@Override
   public UserDocument queryDocument(String id) throws IOException {
       GetRequest getRequest = new GetRequest(Constant.INDEX, id);
       GetResponse getResponse = client.get(getRequest, RequestOptions.DEFAULT);
       UserDocument result = new UserDocument();
       if (getResponse.isExists()) {
           String sourceAsString = getResponse.getSourceAsString();
           result = JSON.parseObject(sourceAsString, UserDocument.class);
       } else {
           logger.error("没有找到该 id 的文档");
       }
       return result;
   }

@Override
 public Boolean updateDocument(UserDocument document) throws Exception {
     String id = document.getId();
     UpdateRequest updateRequest = new UpdateRequest();
     updateRequest.index(Constant.INDEX).id(id);
     updateRequest.doc(JSON.toJSONString(document), XContentType.JSON);
     UpdateResponse updateResponse = client.update(updateRequest, RequestOptions.DEFAULT);
     return updateResponse.status().equals(RestStatus.OK);
 }

@Override
public String deleteDocument(String id) throws Exception {
    DeleteRequest deleteRequest = new DeleteRequest(Constant.INDEX, id);
    DeleteResponse response = client.delete(deleteRequest, RequestOptions.DEFAULT);
    return response.getResult().name();
}

@Override
public Boolean bulkCreateDocument(List<UserDocument> documents) throws IOException {
    BulkRequest bulkRequest = new BulkRequest();
    for (UserDocument document : documents) {
        String id = document.getId();
        IndexRequest indexRequest = new IndexRequest(Constant.INDEX)
                .id(id)
                .source(JSON.toJSONString(document), XContentType.JSON);
        bulkRequest.add(indexRequest);
    }
    BulkResponse bulkResponse = client.bulk(bulkRequest, RequestOptions.DEFAULT);
    return bulkResponse.status().equals(RestStatus.OK);
}

@Override
public Boolean bulkDeleteDocument(List<UserDocument> documents) throws Exception {
    BulkRequest bulkRequest = new BulkRequest();
    for (UserDocument document : documents) {
        String id = document.getId();
        bulkRequest.add(new DeleteRequest().index(Constant.INDEX).id(id));
    }
    BulkResponse bulkResponse = client.bulk(bulkRequest, RequestOptions.DEFAULT);
    return bulkResponse.status().equals(RestStatus.OK);
}

@Override
public List<UserDocument> queryAllDocument() throws IOException {
    SearchRequest getAllRequest = new SearchRequest();
    getAllRequest.indices(Constant.INDEX);
    getAllRequest.source(new SearchSourceBuilder().query(QueryBuilders.matchAllQuery()));
    SearchResponse getAllResponse = client.search(getAllRequest, RequestOptions.DEFAULT);
    SearchHits hits = getAllResponse.getHits();
    List<UserDocument> result = new ArrayList<>();
    for ( SearchHit hit : hits ) {
        result.add(JSON.parseObject(hit.getSourceAsString(), UserDocument.class));
    }
    return result;
}

@Override
public List<UserDocument> queryFilterDocument() throws IOException {
    SearchRequest request = new SearchRequest();
    request.indices(Constant.INDEX);
    SearchSourceBuilder builder = new SearchSourceBuilder().query(QueryBuilders.matchAllQuery());
    builder.sort("age", SortOrder.DESC);
    String[] excludes = {"id","city"};
    String[] includes = {};
    builder.fetchSource(includes, excludes);
    request.source(builder);
    SearchResponse response = client.search(request, RequestOptions.DEFAULT);
    SearchHits hits = response.getHits();
    List<UserDocument> result = new ArrayList<>();
    for ( SearchHit hit : hits ) {
        result.add(JSON.parseObject(hit.getSourceAsString(), UserDocument.class));
    }
    return result;
}

@Override
public List<UserDocument> queryPageDocument(int from, int size) throws IOException {
    SearchRequest getPartRequest = new SearchRequest();
    getPartRequest.indices(Constant.INDEX);
    SearchSourceBuilder builder = new SearchSourceBuilder().query(QueryBuilders.matchAllQuery());
    builder.from(from); // 分页起始位置,(当前页码-1)*每页显示数据条数
    builder.size(size); // 每页展示条数
    getPartRequest.source(builder);
    SearchResponse response = client.search(getPartRequest, RequestOptions.DEFAULT);
    SearchHits hits = response.getHits();
    List<UserDocument> result = new ArrayList<>();
    for ( SearchHit hit : hits ) {
        result.add(JSON.parseObject(hit.getSourceAsString(), UserDocument.class));
    }
    return result;
}

@Override
public List<UserDocument> querySingleConditionDocument(String name) throws IOException {
    SearchRequest request = new SearchRequest();
    request.indices(Constant.INDEX);
    request.source(new SearchSourceBuilder().query(QueryBuilders.termQuery("name", name)));
    SearchResponse response = client.search(request, RequestOptions.DEFAULT);
    SearchHits hits = response.getHits();
    List<UserDocument> result = new ArrayList<>();
    for ( SearchHit hit : hits ) {
        result.add(JSON.parseObject(hit.getSourceAsString(), UserDocument.class));
    }
    return result;
}

@Override
public List<UserDocument> queryCombinationConditionDocument(String name,String city) throws IOException {
    SearchRequest request = new SearchRequest();
    request.indices(Constant.INDEX);
    SearchSourceBuilder builder = new SearchSourceBuilder();
    BoolQueryBuilder boolQueryBuilder = QueryBuilders.boolQuery();
    boolQueryBuilder.must(QueryBuilders.matchQuery("name", name));
    boolQueryBuilder.mustNot(QueryBuilders.matchQuery("city", city));
    builder.query(boolQueryBuilder);
    request.source(builder);
    SearchResponse response = client.search(request, RequestOptions.DEFAULT);
    SearchHits hits = response.getHits();
    List<UserDocument> result = new ArrayList<>();
    for ( SearchHit hit : hits ) {
        result.add(JSON.parseObject(hit.getSourceAsString(), UserDocument.class));
    }
    return result;
}

@Override
public List<UserDocument> queryRangeDocument(int minAge, int maxAge) throws IOException {
    SearchRequest request = new SearchRequest();
    request.indices(Constant.INDEX);
    SearchSourceBuilder builder = new SearchSourceBuilder();
    RangeQueryBuilder rangeQuery = QueryBuilders.rangeQuery("age");
    // gt大于，gte大于等于，lt小于，lte小于等于
    rangeQuery.gte(minAge);
    rangeQuery.lt(maxAge);
    builder.query(rangeQuery);
    request.source(builder);
    SearchResponse response = client.search(request, RequestOptions.DEFAULT);
    SearchHits hits = response.getHits();
    List<UserDocument> result = new ArrayList<>();
    for ( SearchHit hit : hits ) {
        result.add(JSON.parseObject(hit.getSourceAsString(), UserDocument.class));
    }
    return result;
}

@Override
public List<UserDocument> queryFuzzyDocument(String name) throws IOException {
    SearchRequest request = new SearchRequest();
    request.indices(Constant.INDEX);
    SearchSourceBuilder builder = new SearchSourceBuilder();
    builder.query(QueryBuilders.fuzzyQuery("name", name).fuzziness(Fuzziness.ONE)); // 模糊字段偏移量
    request.source(builder);
    SearchResponse response = client.search(request, RequestOptions.DEFAULT);
    SearchHits hits = response.getHits();
    List<UserDocument> result = new ArrayList<>();
    for ( SearchHit hit : hits ) {
        result.add(JSON.parseObject(hit.getSourceAsString(), UserDocument.class));
    }
    return result;
}

import cn.hutool.core.util.ObjectUtil;
import cn.hutool.core.util.StrUtil;
import cn.hutool.json.JSONUtil;
import lombok.extern.slf4j.Slf4j;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.delete.DeleteRequest;
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.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.client.core.CountRequest;
import org.elasticsearch.index.query.BoolQueryBuilder;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.aggregations.AggregationBuilder;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.elasticsearch.search.sort.SortOrder;
import org.elasticsearch.xcontent.XContentType;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.io.IOException;
import java.util.*;

/**
 * 通用ElasticSearch工具接口
 */
@Slf4j
@Service
public class ElasticSearchService  {

    @Autowired
    private RestHighLevelClient restHighLevelClient;

    /**
     * 共用方法：ElasticSearch通用查询
     * @param searchSourceBuilder
     * @param index
     * @return
     */
    private SearchResponse pageQuerySearchResponse(SearchSourceBuilder searchSourceBuilder, String index) {
        SearchRequest searchRequest = new SearchRequest()
                .source(searchSourceBuilder)
                .indices(index);
        SearchResponse searchResponse = null;
        try {
            searchResponse = restHighLevelClient.search(searchRequest, RequestOptions.DEFAULT);
        } catch (IOException e) {
            e.printStackTrace();
        }
        return searchResponse;
    }

    /**
     * 共用方法：ElasticSearch批量操作（新增、删除）数据
     *
     * @param bulkRequest
     */
    private void esBatch(BulkRequest bulkRequest) {
        try {
            BulkResponse bulkResponse = restHighLevelClient.bulk(bulkRequest, RequestOptions.DEFAULT);
            if (bulkResponse.hasFailures()) {
                log.error("bulk错误信息：{}", bulkResponse.buildFailureMessage());
            }
        } catch (Exception e) {
            log.error("批量操作es数据错误", e);
        }
    }

    /**
     * 查询指定条件的数据
     * @param indexName 索引名称
     * @param sortField 排序字段
     * @param sortType 排序类别
     * @param page 页码
     * @param rows 每页大小
     * @param boolQueryBuilder 查询条件
     * @param includeFields 返回字段
     * @param excludeFields 排除字段
     * @return
     */
    public SearchResponse search(String indexName, String sortField, String sortType, Integer page, Integer rows, BoolQueryBuilder boolQueryBuilder,
                                 String[] includeFields, String[] excludeFields) {

        SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder()
                .from((page - 1) * rows)
                .size(rows)
                .fetchSource(includeFields, excludeFields)
                .trackTotalHits(true);

        SortOrder so = SortOrder.DESC;
        if (ObjectUtil.isNotEmpty(sortType) && "asc".equals(sortType)) {
            so = SortOrder.ASC;
        }
        searchSourceBuilder.sort(sortField, so);
        searchSourceBuilder.query(boolQueryBuilder);

        return pageQuerySearchResponse(searchSourceBuilder, indexName);
    }

    /**
     * 查询单条数据
     * @param indexName 索引名称
     * @param boolQueryBuilder 查询条件
     * @return
     */
    public Map<String, Object> searchOne(String indexName, BoolQueryBuilder boolQueryBuilder) {
        Map<String, Object> resultMap = new HashMap();
        SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder().from(0).size(1);
        searchSourceBuilder.query(boolQueryBuilder);
        SearchResponse searchResponse = pageQuerySearchResponse(searchSourceBuilder, indexName);
        if (searchResponse.getHits().getTotalHits().value > 0) {
            SearchHit searchHit = searchResponse.getHits().getHits()[0];
            resultMap = searchHit.getSourceAsMap();
        }
        return resultMap;
    }

    /**
     * 获取指定索引下的数据量
     *
     * @param indexName
     * @return
     */
    public Long getCountByIndex(String indexName) {
        Long totalHites = 0L;
        if (!StrUtil.isEmpty(indexName)) {
            try {
                CountRequest countRequest = new CountRequest(indexName);
                totalHites = restHighLevelClient.count(countRequest, RequestOptions.DEFAULT).getCount();
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
        return totalHites;
    }

    /**
     * 查询聚合数据
     * @param indexName 索引名称
     * @param boolQueryBuilder 查询条件
     * @param aggregationBuilder 聚合条件
     * @param includeFields 返回字段
     * @param excludeFields 排除字段
     * @return
     */
    public SearchResponse aggsSearch(String indexName, BoolQueryBuilder boolQueryBuilder, AggregationBuilder aggregationBuilder,
                                     String[] includeFields, String[] excludeFields) {
        SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder()
                .size(0)
                .fetchSource(includeFields, excludeFields)
                .trackTotalHits(true);
        searchSourceBuilder.aggregation(aggregationBuilder);
        searchSourceBuilder.query(boolQueryBuilder);
        return pageQuerySearchResponse(searchSourceBuilder, indexName);
    }

    /**
     * 批量添加数据（自动生成id）
     * @param indexName
     * @param dataList
     */
    public void addBatchDataAutoId(String indexName, List<Map> dataList) {
        BulkRequest request = new BulkRequest();
        try {
            int count = 0;
            for (Map map : dataList) {
                request.add(new IndexRequest(indexName).source(map));
                count++;
                if (count == 1000) {
                    esBatch(request);
                    count = 0;
                }
            }
            if (count != 0) {
                esBatch(request);
            }
        } catch (Exception e) {
            log.error("生成BulkRequest错误", e);
        }
    }

    /**
     * 批量添加数据（主动指定id）
     * @param indexName
     * @param dataList
     */
    public void addBatchData(String indexName, List<Map> dataList) {
        BulkRequest request = new BulkRequest();
        try {
            int count = 0;
            for (Map map : dataList) {
                String id = map.get("id").toString();
                request.add(new IndexRequest(indexName).id(id).source(map));
                count++;
                if (count == 1000) {
                    esBatch(request);
                    count = 0;
                }
            }
            if (count != 0) {
                esBatch(request);
            }
        } catch (Exception e) {
            log.error("生成BulkRequest错误", e);
        }
    }

    /**
     * 批量删除数据
     * @param indexName
     * @param idList
     */
    public void deleteBatchData(String indexName, List<String> idList) {
        BulkRequest request = new BulkRequest();
        try {
            int count = 0;
            for (String id : idList) {
                request.add(new DeleteRequest().index(indexName).id(id));
                count++;
                if (count == 1000) {
                    esBatch(request);
                    count = 0;
                }
            }
            if (count != 0) {
                esBatch(request);
            }
        } catch (Exception e) {
            log.error("生成BulkRequest错误", e);
        }
    }

    /**
     * 修改数据
     * @param indexName
     * @param dataMap
     */
    public void update(String indexName, Map dataMap) {
        String contentId = dataMap.get("id").toString();
        UpdateRequest updateRequest = new UpdateRequest(indexName, "_doc", contentId);
        updateRequest.doc(JSONUtil.parse(dataMap.get("data")).toString(), XContentType.JSON);
        try {
            restHighLevelClient.update(updateRequest, RequestOptions.DEFAULT);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

}