1准备工作
es底层支持浅分页、游标分页、searchafter分页。
浅分页跳页数太大，就太费资源了
游标分页不实时
searchAfter不支持跳页
工具类
public class CommonResult<T> { private long code; private String message; private T data; protected CommonResult() { } protected CommonResult(long code, String message, T data) { this.code = code; this.message = message; this.data = data; } /** * 成功返回结果 * * @param data 获取的数据 */ public static <T> CommonResult<T> success(T data) { return new CommonResult<T>(ResultCode.SUCCESS.getCode(), ResultCode.SUCCESS.getMessage(), data); } /** * 成功返回结果 * * @param data 获取的数据 * @param message 提示信息 */ public static <T> CommonResult<T> success(T data, String message) { return new CommonResult<T>(ResultCode.SUCCESS.getCode(), message, data); } /** * 失败返回结果 * @param errorCode 错误码 */ public static <T> CommonResult<T> failed(IErrorCode errorCode) { return new CommonResult<T>(errorCode.getCode(), errorCode.getMessage(), null); } /** * 失败返回结果 * @param message 提示信息 */ public static <T> CommonResult<T> failed(String message) { return new CommonResult<T>(ResultCode.FAILED.getCode(), message, null); } /** * 失败返回结果 */ public static <T> CommonResult<T> failed() { return failed(ResultCode.FAILED); } /** * 参数验证失败返回结果 */ public static <T> CommonResult<T> validateFailed() { return failed(ResultCode.VALIDATE_FAILED); } /** * 参数验证失败返回结果 * @param message 提示信息 */ public static <T> CommonResult<T> validateFailed(String message) { return new CommonResult<T>(ResultCode.VALIDATE_FAILED.getCode(), message, null); } /** * 未登录返回结果 */ public static <T> CommonResult<T> unauthorized(T data) { return new CommonResult<T>(ResultCode.UNAUTHORIZED.getCode(), ResultCode.UNAUTHORIZED.getMessage(), data); } /** * 未授权返回结果 */ public static <T> CommonResult<T> forbidden(T data) { return new CommonResult<T>(ResultCode.FORBIDDEN.getCode(), ResultCode.FORBIDDEN.getMessage(), data); } public long getCode() { return code; } public void setCode(long code) { this.code = code; } public String getMessage() { return message; } public void setMessage(String message) { this.message = message; } public T getData() { return data; } public void setData(T data) { this.data = data; } } package com.xxx.entity.resp; import com.github.pagehelper.PageInfo; import org.springframework.data.domain.Page; import java.util.List; public class CommonPage<T> { private Integer pageNum; private Integer pageSize; private Integer totalPage; private Long total; private List<T> list; /** * 将PageHelper分页后的list转为分页信息 */ public static <T> CommonPage<T> restPage(List<T> list) { CommonPage<T> result = new CommonPage<T>(); PageInfo<T> pageInfo = new PageInfo<T>(list); result.setTotalPage(pageInfo.getPages()); result.setPageNum(pageInfo.getPageNum()); result.setPageSize(pageInfo.getPageSize()); result.setTotal(pageInfo.getTotal()); result.setList(pageInfo.getList()); return result; } /** * 将SpringData分页后的list转为分页信息 */ public static <T> CommonPage<T> restPage(Page<T> pageInfo) { CommonPage<T> result = new CommonPage<T>(); result.setTotalPage(pageInfo.getTotalPages()); result.setPageNum(pageInfo.getNumber()); result.setPageSize(pageInfo.getSize()); result.setTotal(pageInfo.getTotalElements()); result.setList(pageInfo.getContent()); return result; } public Integer getPageNum() { return pageNum; } public void setPageNum(Integer pageNum) { this.pageNum = pageNum; } public Integer getPageSize() { return pageSize; } public void setPageSize(Integer pageSize) { this.pageSize = pageSize; } public Integer getTotalPage() { return totalPage; } public void setTotalPage(Integer totalPage) { this.totalPage = totalPage; } public List<T> getList() { return list; } public void setList(List<T> list) { this.list = list; } public Long getTotal() { return total; } public void setTotal(Long total) { this.total = total; } } package com.xxx.entity.resp; import lombok.Data; import lombok.NoArgsConstructor; import lombok.experimental.Accessors; import org.springframework.data.annotation.Id; import org.springframework.data.elasticsearch.annotations.DateFormat; 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 @NoArgsConstructor @Accessors(chain = true) @Document(indexName = "index", type = "_doc", shards = 1, replicas = 0) public class LogInfoPage implements Serializable { private static final long serialVersionUID = -1L; @Id private String id;//主键 @Field(type = FieldType.Integer) private Integer logName;//name @Field(type = FieldType.Integer) private Integer logType;//日志类型 （1：异常日志、2：操作日志） @Field(type = FieldType.Integer) private Integer operationType;//操作类型（1：增加操作、2：删除操作） @Field(type = FieldType.Text) private String operationDesc;//操作描述 @Field(type = FieldType.Keyword) private String operationUserId;//操作用户id @Field(type = FieldType.Date, format = DateFormat.basic_date) private String operationTime;//操作时间 @Field(type = FieldType.Text, analyzer = "ik_max_word", searchAnalyzer = "ik_smart") private String logCont;//日志内容 //以下属性为系统自动获取 @Field(type = FieldType.Keyword) private String requestIp;//请求ip @Field(type = FieldType.Date, format = DateFormat.basic_date) private String collectTime;//日志收集时间 }
代码流程
pom: <dependency> <groupId>org.springframework.data</groupId> <artifactId>spring-data-elasticsearch</artifactId> <version>4.0.3.RELEASE</version> </dependency> <dependency> <groupId>com.github.pagehelper</groupId> <artifactId>pagehelper-spring-boot-starter</artifactId> <version>1.2.10</version> </dependency> controller: @ApiOperation(value = "分页查询日志数据", notes = "分页查询日志数据") @RequestMapping(value = "/search/list", method = RequestMethod.POST) @ResponseBody public CommonResult<CommonPage<LogInfoPage>> search(@RequestBody LogQueryDTO logInfo) { log.info("logInfo======" + logInfo); Page<LogInfoPage> esProductPage = logQueryService.search(logInfo); return CommonResult.success(CommonPage.restPage(esProductPage)); } service: public interface LogQueryService { Page<LogInfoPage> search(LogQueryDTO logInfo); } service.impl: public Page<LogInfoPage> search(LogQueryDTO logInfo) { Pageable pageable = PageRequest.of(logInfo.getPageNum() - 1, logInfo.getPageSize(), Sort.by(Sort.Direction.DESC, "operationTime")); return logInfoRepository.getdatas(logInfo, pageable); } 接口： public interface LogInfoRepositoryCustom<T, ID extends Serializable> { Page<LogInfoPage> getdatas(LogQueryDTO logInfo, Pageable pageable); } 接口-实现： package com.xxxxx.serivce.impl; import com.xxxxx.entity.dto.LogQueryDTO; import com.xxxxx.entity.resp.LogInfoPage; import com.xxxxx.serivce.LogInfoRepository; import com.xxxxx.serivce.LogInfoRepositoryCustom; import lombok.extern.slf4j.Slf4j; import org.apache.commons.lang3.StringUtils; import org.elasticsearch.index.query.BoolQueryBuilder; import org.elasticsearch.index.query.QueryBuilders; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.data.domain.Page; import org.springframework.data.domain.Pageable; import org.springframework.data.elasticsearch.core.query.NativeSearchQuery; import org.springframework.data.elasticsearch.core.query.NativeSearchQueryBuilder; import org.springframework.stereotype.Component; @Slf4j @Component public class LogInfoRepositoryImpl implements LogInfoRepositoryCustom<LogInfoPage, String> { @Autowired private LogInfoRepository logInfoRepository; @Override public Page<LogInfoPage> getdatas(LogQueryDTO logInfo, Pageable pageable) { NativeSearchQueryBuilder nativeSearchQueryBuilder = new NativeSearchQueryBuilder(); //分页 nativeSearchQueryBuilder.withPageable(pageable); //过滤 if (logInfo.getOperationType() != null || logInfo.getSystemSign() != null || logInfo.getSoftwareSign() != null || StringUtils.isNotEmpty(logInfo.getOperationTime()) || StringUtils.isNotEmpty(logInfo.getOperationDesc()) || StringUtils.isNotEmpty(logInfo.getRequestIp()) ) { BoolQueryBuilder boolQueryBuilder = QueryBuilders.boolQuery(); if (logInfo.getOperationType() != null) { boolQueryBuilder.must(QueryBuilders.matchPhraseQuery("operationType", logInfo.getOperationType())); } ......... nativeSearchQueryBuilder.withFilter(boolQueryBuilder); } NativeSearchQuery searchQuery = nativeSearchQueryBuilder.build(); Page<LogInfoPage> search = logInfoRepository.search(searchQuery); return search; } } Repository: package com.xxxx.serivce; import com.xxxx.entity.resp.LogInfoPage; import org.springframework.data.elasticsearch.repository.ElasticsearchRepository; public interface LogInfoRepository extends ElasticsearchRepository<LogInfoPage, String>, LogInfoRepositoryCustom<LogInfoPage, String> { }
es7简单操作
package com.xxxxx.es; import com.alibaba.fastjson.JSON; import lombok.extern.slf4j.Slf4j; import org.apache.http.HttpHost; import org.elasticsearch.action.admin.indices.get.GetIndexRequest; 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.update.UpdateRequest; import org.elasticsearch.action.update.UpdateResponse; import org.elasticsearch.client.IndicesClient; import org.elasticsearch.client.RequestOptions; import org.elasticsearch.client.RestClient; import org.elasticsearch.client.RestHighLevelClient; import org.elasticsearch.common.text.Text; import org.elasticsearch.common.xcontent.XContentType; import org.elasticsearch.index.query.*; import org.elasticsearch.search.SearchHit; import org.elasticsearch.search.SearchHits; import org.elasticsearch.search.aggregations.Aggregation; import org.elasticsearch.search.aggregations.AggregationBuilders; import org.elasticsearch.search.aggregations.Aggregations; import org.elasticsearch.search.aggregations.bucket.terms.ParsedStringTerms; import org.elasticsearch.search.aggregations.bucket.terms.TermsAggregationBuilder; import org.elasticsearch.search.builder.SearchSourceBuilder; import org.elasticsearch.search.fetch.subphase.highlight.HighlightBuilder; import org.elasticsearch.search.fetch.subphase.highlight.HighlightField; import org.elasticsearch.search.sort.SortOrder; import java.io.IOException; import java.util.*; import java.util.stream.Collectors; @Slf4j public class EsCrudTests { private static final String ES_INDEX = "ecommerce"; private static final String ES_TYPE = "product"; public static void main(String[] args) throws Exception { RestHighLevelClient restHighLevelClient = null; try { restHighLevelClient = new RestHighLevelClient( // RestClient.builder(new HttpHost("ip", 9200, "http"))); RestClient.builder(new HttpHost("ip", 9900, "http"))); createInfo(restHighLevelClient); // createInfoBulk(restHighLevelClient); // getInfo(restHighLevelClient); // getInfoAggs(restHighLevelClient); // getInfoBool(restHighLevelClient); // getInfoHight(restHighLevelClient); // getInfoMatchPhrase(restHighLevelClient); // updateInfo(restHighLevelClient); // deleteInfo(restHighLevelClient); // existIndex(restHighLevelClient); } finally { restHighLevelClient.close(); } } /** * 插入 * * @param restHighLevelClient * @throws Exception */ private static void createInfo(RestHighLevelClient restHighLevelClient) throws Exception { Map<String, Object> requestMap = new HashMap<>(); requestMap.put("name", "gaolujie4"); requestMap.put("desc", "gaolujie4 meibai"); requestMap.put("price", 33); requestMap.put("producer", "gaolujie3 producer"); requestMap.put("tags", Arrays.asList("meibai", "fangzhu")); IndexRequest indexRequest = new IndexRequest(ES_INDEX, ES_TYPE); String source = JSON.toJSONString(requestMap); indexRequest.source(source, XContentType.JSON); // 操作ES IndexResponse indexResponse = null; try { indexResponse = restHighLevelClient.index(indexRequest, RequestOptions.DEFAULT); } catch (IOException e) { e.printStackTrace(); } log.info(indexResponse.toString()); restHighLevelClient.close(); } /** * 批量插入 * * @param restHighLevelClient * @throws Exception */ private static void createInfoBulk(RestHighLevelClient restHighLevelClient) throws Exception { Map<String, Object> requestMap = new HashMap<>(); requestMap.put("name", "gaolujie0"); requestMap.put("desc", "gaolujie0 meibai"); requestMap.put("price", 10); requestMap.put("producer", "gaolujie0 producer"); requestMap.put("tags", Arrays.asList("meibai", "fangzhu")); BulkRequest bulkRequestBuilder = new BulkRequest(); IndexRequest indexRequest = new IndexRequest(ES_INDEX, ES_TYPE); String source = JSON.toJSONString(requestMap); indexRequest.source(source, XContentType.JSON); bulkRequestBuilder.add(indexRequest); bulkRequestBuilder.add(indexRequest); bulkRequestBuilder.add(indexRequest); bulkRequestBuilder.add(indexRequest); // 操作ES BulkResponse bulk = null; try { bulk = restHighLevelClient.bulk(bulkRequestBuilder, RequestOptions.DEFAULT); log.info("bulk====" + bulk); } catch (IOException e) { e.printStackTrace(); } log.info(bulk.toString()); restHighLevelClient.close(); } /** * 查询排序返回bean * * @param restHighLevelClient * @throws Exception */ private static void getInfo(RestHighLevelClient restHighLevelClient) throws Exception { SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder(); searchSourceBuilder.sort("price", SortOrder.DESC); QueryBuilder queryBuilder = QueryBuilders.matchQuery("producer", "producer"); searchSourceBuilder.query(queryBuilder); SearchRequest searchRequest = new SearchRequest(); searchRequest.source(searchSourceBuilder); SearchResponse searchResponse = restHighLevelClient.search(searchRequest, RequestOptions.DEFAULT); SearchHits hits = searchResponse.getHits(); for (SearchHit hit : hits) { Map<String, Object> sourceAsMap = hit.getSourceAsMap(); Iterator<Map.Entry<String, Object>> entries = sourceAsMap.entrySet().iterator(); while (entries.hasNext()) { Map.Entry<String, Object> entry = entries.next(); String key = entry.getKey(); Object value = entry.getValue(); if (key.contains("producer")) { log.info(key + ":" + value); } } log.info("id===" + hit.getId()); } List<ProductBean> bookDtoList = new ArrayList<>(); SearchHit[] searchHits = hits.getHits(); for (SearchHit searchHit : searchHits) { bookDtoList.add(JSON.parseObject(searchHit.getSourceAsString(), ProductBean.class)); } // 封装Map参数返回 Map<String, Object> result = new HashMap<>(16); result.put("count", 1); result.put("data", bookDtoList); log.info("result====" + result.toString()); } /** * 简单聚合 * * @param restHighLevelClient * @throws Exception */ private static void getInfoAggs(RestHighLevelClient restHighLevelClient) throws Exception { SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder(); TermsAggregationBuilder field1 = AggregationBuilders.terms("group_by_tags").field("tags"); searchSourceBuilder.aggregation(field1); SearchRequest searchRequest = new SearchRequest(); searchRequest.source(searchSourceBuilder); SearchResponse searchResponse = restHighLevelClient.search(searchRequest, RequestOptions.DEFAULT); Aggregations aggregations = searchResponse.getAggregations(); for (Aggregation aggregation : aggregations) { String type = aggregation.getType(); log.info("type===" + type); Map<String, Object> map = new HashMap(); if (aggregation instanceof ParsedStringTerms) { String name = aggregation.getName(); log.info("name===" + name); List<ParsedStringTerms.ParsedBucket> collect = (List<ParsedStringTerms.ParsedBucket>) ((ParsedStringTerms) aggregation).getBuckets(); collect.stream().map(a -> { String value = a.getKeyAsString() + "|" + a.getDocCount(); map.put(a.getKeyAsString(), value); return map; }).collect(Collectors.toList()); log.info("map===" + map); } } } /** * bool查询价钱指定范围 * * @param restHighLevelClient * @throws Exception */ private static void getInfoBool(RestHighLevelClient restHighLevelClient) throws Exception { SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder(); BoolQueryBuilder boolQueryBuilder = QueryBuilders.boolQuery(); QueryBuilder queryBuilder = QueryBuilders.termQuery("name", "gaolujie2"); boolQueryBuilder.must(queryBuilder); RangeQueryBuilder price = QueryBuilders.rangeQuery("price").gt(25); boolQueryBuilder.filter(price); searchSourceBuilder.query(boolQueryBuilder); SearchRequest searchRequest = new SearchRequest(); searchRequest.source(searchSourceBuilder); SearchResponse searchResponse = restHighLevelClient.search(searchRequest, RequestOptions.DEFAULT); SearchHits hits = searchResponse.getHits(); List<ProductBean> bookDtoList = new ArrayList<>(); SearchHit[] searchHits = hits.getHits(); for (SearchHit searchHit : searchHits) { bookDtoList.add(JSON.parseObject(searchHit.getSourceAsString(), ProductBean.class)); } // 封装Map参数返回 Map<String, Object> result = new HashMap<>(16); result.put("count", 1); result.put("data", bookDtoList); log.info("result====" + result.toString()); } /** * 模糊匹配 * * @param restHighLevelClient * @throws Exception */ private static void getInfoMatchPhrase(RestHighLevelClient restHighLevelClient) throws Exception { SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder(); QueryBuilder queryBuilder = QueryBuilders.matchPhrasePrefixQuery("producer", "gaolujie"); ((MatchPhrasePrefixQueryBuilder) queryBuilder).slop(1); searchSourceBuilder.query(queryBuilder); SearchRequest searchRequest = new SearchRequest(); searchRequest.source(searchSourceBuilder); SearchResponse searchResponse = restHighLevelClient.search(searchRequest, RequestOptions.DEFAULT); SearchHits hits = searchResponse.getHits(); List<ProductBean> bookDtoList = new ArrayList<>(); SearchHit[] searchHits = hits.getHits(); for (SearchHit searchHit : searchHits) { bookDtoList.add(JSON.parseObject(searchHit.getSourceAsString(), ProductBean.class)); } // 封装Map参数返回 Map<String, Object> result = new HashMap<>(16); result.put("count", 1); result.put("data", bookDtoList); log.info("result====" + result.toString()); } //配置标题高亮显示 /* HighlightBuilder highlightBuilder = new HighlightBuilder(); //生成高亮查询器 highlightBuilder.field(title); //高亮查询字段 highlightBuilder.field(content); //高亮查询字段 highlightBuilder.requireFieldMatch(false); //如果要多个字段高亮,这项要为false highlightBuilder.preTags("<span style="color:red">"); //高亮设置 highlightBuilder.postTags("</span>"); //下面这两项,如果你要高亮如文字内容等有很多字的字段,必须配置,不然会导致高亮不全,文章内容缺失等 highlightBuilder.fragmentSize(800000); //最大高亮分片数 highlightBuilder.numOfFragments(0); //从第一个分片获取高亮片段 */ /** * 高亮 * * @param restHighLevelClient * @throws Exception */ private static void getInfoHight(RestHighLevelClient restHighLevelClient) throws Exception { SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder(); QueryBuilder queryBuilder = QueryBuilders.matchQuery("producer", "producer"); searchSourceBuilder.query(queryBuilder); HighlightBuilder highlightBuilder = new HighlightBuilder(); highlightBuilder.preTags("<highlight>"); highlightBuilder.postTags("</highlight>"); highlightBuilder.fields().add(new HighlightBuilder.Field("producer")); searchSourceBuilder.highlighter(highlightBuilder); SearchRequest searchRequest = new SearchRequest(); searchRequest.source(searchSourceBuilder); SearchResponse searchResponse = restHighLevelClient.search(searchRequest, RequestOptions.DEFAULT); SearchHits hits = searchResponse.getHits(); SearchHit[] searchHits = hits.getHits(); List<Map<String, Object>> list = new ArrayList<>(); for (SearchHit searchHit : searchHits) { Map<String, HighlightField> highlightFields = searchHit.getHighlightFields(); HighlightField titleField = highlightFields.get("producer"); // HighlightField contentField = highlightFields.get(""); Map<String, Object> source = searchHit.getSourceAsMap(); //千万记得要记得判断是不是为空,不然你匹配的第一个结果没有高亮内容,那么就会报空指针异常,这个错误一开始真的搞了很久 if (titleField != null) { Text[] fragments = titleField.fragments(); String name = ""; for (Text text : fragments) { name += text; } source.put("producer", name); //高亮字段替换掉原本的内容 } list.add(source); } // 封装Map参数返回 Map<String, Object> result = new HashMap<>(16); result.put("count", 1); result.put("data", list); log.info("result==high==" + result.toString()); } /** * 更新 * * @param client * @throws Exception */ private static void updateInfo(RestHighLevelClient client) throws Exception { UpdateRequest updateRequest = new UpdateRequest(ES_INDEX, ES_TYPE, "I00VC3QBIpl3hAPcNY_1"); Map<String, Object> requestMap = new HashMap<>(); requestMap.put("price", 11.5); updateRequest.doc(requestMap); UpdateResponse update = client.update(updateRequest, RequestOptions.DEFAULT); log.info("update=====" + update.getShardInfo().getSuccessful()); } /** * 删除 * * @param client * @throws Exception */ private static void deleteInfo(RestHighLevelClient client) throws Exception { DeleteRequest deleteRequest = new DeleteRequest(ES_INDEX, ES_TYPE, "JE0dC3QBIpl3hAPcno8g"); DeleteResponse delete = client.delete(deleteRequest, RequestOptions.DEFAULT); log.info("delete====" + delete.getResult().name()); } /** * 索引是否存在 * * @param client * @throws Exception */ public static void existIndex(RestHighLevelClient client) throws Exception { GetIndexRequest indexRequest = new GetIndexRequest().indices("ecommerce333"); IndicesClient indices = client.indices(); boolean exists = indices.exists(indexRequest, RequestOptions.DEFAULT); log.info("exists===" + exists); } }
批量初始化数据
package com.xxxxx.es; import com.alibaba.fastjson.JSON; import com.xxxxx.config.base.BaseConfig; import com.xxxxx.entity.LogInfo; import com.xxxxx.utils.TimeOperationUtil; import lombok.extern.slf4j.Slf4j; import org.apache.http.HttpHost; import org.elasticsearch.action.bulk.BulkRequest; import org.elasticsearch.action.index.IndexRequest; import org.elasticsearch.client.RequestOptions; import org.elasticsearch.client.RestClient; import org.elasticsearch.client.RestHighLevelClient; import org.elasticsearch.common.xcontent.XContentType; import java.util.List; import java.util.Map; import java.util.Random; @Slf4j public class InitLogData { private static final String ES_INDEX = "index"; public static void main(String[] args) throws Exception { RestHighLevelClient restHighLevelClient = null; try { restHighLevelClient = new RestHighLevelClient( // RestClient.builder(new HttpHost("ip", 9200, "http"))); RestClient.builder(new HttpHost("ip", 9900, "http"))); createInfoBulk(restHighLevelClient); } finally { restHighLevelClient.close(); } } private static void createInfoBulk(RestHighLevelClient restHighLevelClient) throws Exception { BulkRequest bulkRequestBuilder = new BulkRequest(); for (Map.Entry<Integer, List<Integer>> entry : BaseConfig.MY_MAP.entrySet()) { Integer mapKey = entry.getKey(); List<Integer> mapValue = entry.getValue(); for (Integer integer : mapValue) { for (int i = 0; i < 100; i++) { log.info("key===" + mapKey + "===value:" + integer); LogInfo logInfo = new LogInfo(); IndexRequest indexRequest = new IndexRequest(ES_INDEX); logInfo.setSoftwareSign(mapKey); logInfo.setLogType(getRandom(1, 3)); logInfo.setOperationType(getRandom(1, 5)); logInfo.setOperationDesc("ceshi" + i + mapKey + integer); logInfo.setOperationUserId("zhang" + i + mapKey + integer); logInfo.setOperationTime("2020-" + getRandomDate(1, 12) + "-" + getRandomDate(1, 12) + " " + getRandomDate(1, 12) + ":" + getRandomDate(1, 12) + ":" + getRandomDate(1, 12)); logInfo.setLogCont("2020-" + getRandomDate(1, 12) + "-" + getRandomDate(1, 12) + " " + getRandomDate(1, 12) + ":" + getRandomDate(1, 12) + ":" + getRandomDate(1, 12)); logInfo.setRequestIp(getRandom(200, 500) + "ip"); //2、设置采集时间 logInfo.setCollectTime(TimeOperationUtil.getNowTime()); String source = JSON.toJSONString(logInfo); indexRequest.source(source, XContentType.JSON); bulkRequestBuilder.add(indexRequest); } } } restHighLevelClient.bulk(bulkRequestBuilder, RequestOptions.DEFAULT); } public static int getRandom(int min, int max) { Random random = new Random(); return random.nextInt(max) % (max - min + 1) + min; } public static String getRandomDate(int min, int max) { Random random = new Random(); int s = random.nextInt(max) % (max - min + 1) + min; String s1 = String.valueOf(s).length() == 1 ? "0" + s : String.valueOf(s); return s1; } }
searchAfter
postman请求体： { "indexType":"index", "size":20, "from":20, "map":{ "name":11 }, "objects":[1597108029000, 5] } 代码： public Response queryLogInfos(LogQueryDTO logQueryDTO) { log.info("fenye========" + logQueryDTO); Response response = new Response(); try { if (StringUtils.isEmpty(logQueryDTO.getIndexType())) { return response.failure(BaseConfig.PARAM_IS_NUMM); } int from = logQueryDTO.getFrom() <= -1 ? 0 : logQueryDTO.getFrom(); int size = logQueryDTO.getPageSize(); if (size >= 1000) { size = 1000; } else if (logQueryDTO.getPageSize() <= 0) { size = 10; } SearchRequest searchRequest = new SearchRequest(logQueryDTO.getIndexType()); SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder(); BoolQueryBuilder boolQueryBuilder = QueryBuilders.boolQuery(); if (logQueryDTO.getMap() != null && !logQueryDTO.getMap().isEmpty()) { for (Map.Entry<String, Object> entry : logQueryDTO.getMap().entrySet()) { boolQueryBuilder.filter(QueryBuilders.termQuery(entry.getKey(), entry.getValue())); } } searchSourceBuilder.query(boolQueryBuilder); // searchAfter赋值 if (logQueryDTO.getObjects() != null && logQueryDTO.getObjects().length > 0) { searchSourceBuilder.searchAfter(logQueryDTO.getObjects()); from = 0; } searchSourceBuilder.from(from); searchSourceBuilder.size(size); searchSourceBuilder.sort(new FieldSortBuilder("operationTime").order(SortOrder.DESC)); searchSourceBuilder.sort(new FieldSortBuilder("operationType").order(SortOrder.DESC)); searchRequest.source(searchSourceBuilder); searchRequest.indicesOptions(IndicesOptions.lenientExpandOpen()); SearchResponse search = restHighLevelClient.search(searchRequest, RequestOptions.DEFAULT); LogInfosResponse logInfosResponse = new LogInfosResponse(); List<LogInfoPageResponse> listRes = new ArrayList<>(); // 总条数 long value = search.getHits().getTotalHits().value; logInfosResponse.setTotalValue(value); // sort SearchHit[] searchHits = search.getHits().getHits(); for (SearchHit searchHit : searchHits) { LogInfoPageResponse logInfoResponse = JSON.parseObject(searchHit.getSourceAsString(), LogInfoPageResponse.class); logInfoResponse.setId(searchHit.getId()); logInfoResponse.setObjects(searchHit.getSortValues()); listRes.add(logInfoResponse); logInfosResponse.setInfoResponseList(listRes); } response.setData(logInfosResponse); // 封装Map参数返回 log.info("search=====" + search); } catch (Exception e) { e.printStackTrace(); log.error(e.getMessage()); return response.failure(e.getMessage()); } return response; }
游标分页等
package com.example.es7.demo.es7.demo; import lombok.extern.slf4j.Slf4j; import org.apache.http.HttpHost; import org.elasticsearch.action.ActionListener; import org.elasticsearch.action.get.GetRequest; import org.elasticsearch.action.get.GetResponse; import org.elasticsearch.action.index.IndexRequest; import org.elasticsearch.action.index.IndexResponse; import org.elasticsearch.action.search.*; import org.elasticsearch.action.support.IndicesOptions; import org.elasticsearch.action.update.UpdateRequest; import org.elasticsearch.action.update.UpdateResponse; import org.elasticsearch.client.RequestOptions; import org.elasticsearch.client.RestClient; import org.elasticsearch.client.RestHighLevelClient; import org.elasticsearch.common.unit.TimeValue; import org.elasticsearch.index.query.QueryBuilders; import org.elasticsearch.index.query.RangeQueryBuilder; import org.elasticsearch.search.Scroll; import org.elasticsearch.search.SearchHit; import org.elasticsearch.search.builder.SearchSourceBuilder; import org.elasticsearch.search.fetch.subphase.FetchSourceContext; import java.io.IOException; import java.util.Date; import java.util.HashMap; import java.util.Map; import java.util.concurrent.TimeUnit; @Slf4j public class ElasticSearch1 { private static final String ES_INDEX = "gaofm01"; public static void main(String[] args) throws Exception { RestHighLevelClient restHighLevelClient = null; try { restHighLevelClient = new RestHighLevelClient( RestClient.builder(new HttpHost("ip", 9900, "http"))); // createIndex(restHighLevelClient); // getIndex(restHighLevelClient); // checkExists(restHighLevelClient); // updateIndex(restHighLevelClient); // searchAfterSearch(restHighLevelClient); searchScroll(restHighLevelClient); } finally { restHighLevelClient.close(); } } public static void searchScroll(RestHighLevelClient client) throws IOException { Scroll scroll = new Scroll(TimeValue.timeValueMinutes(10L)); SearchRequest searchRequest = new SearchRequest(ES_INDEX); SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder(); RangeQueryBuilder price = QueryBuilders.rangeQuery("operationTime").gte("1600756706000"); searchSourceBuilder.query(price); searchSourceBuilder.size(100); searchRequest.scroll(scroll); searchRequest.source(searchSourceBuilder); SearchResponse search = client.search(searchRequest, RequestOptions.DEFAULT); log.info("search=====" + search); String scrollId = search.getScrollId(); SearchHit[] hits = search.getHits().getHits(); while (hits != null && hits.length > 0) { SearchScrollRequest scrollRequest = new SearchScrollRequest(scrollId); scrollRequest.scroll(scroll); search = client.scroll(scrollRequest, RequestOptions.DEFAULT); scrollId = search.getScrollId(); SearchHit[] hits1 = search.getHits().getHits(); if (hits1 != null && hits1.length > 0) { for (SearchHit documentFields : hits1) { log.info("documentFields====" + documentFields); } } } // clean scroll ClearScrollRequest clearScrollRequest = new ClearScrollRequest(); clearScrollRequest.addScrollId(scrollId); ClearScrollResponse clearScrollResponse = null; try { clearScrollResponse = client.clearScroll(clearScrollRequest, RequestOptions.DEFAULT); } catch (IOException e) { log.error("clear-scroll-error:{}", e); } boolean succeeded = clearScrollResponse.isSucceeded(); System.out.println(succeeded); } public static void searchAfterSearch(RestHighLevelClient client) { try { String startTime = "2020-02-06 07:06:06"; String endTime = "2020-03-06 07:06:06"; int from = 0; int size = 10; SearchRequest searchRequest = new SearchRequest(ES_INDEX); SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder(); searchSourceBuilder.query(QueryBuilders.boolQuery() // 查询条件，按照时间查询错误日志 .must(QueryBuilders.matchQuery("user", "kimchy")) .must(QueryBuilders.rangeQuery("postDate") .timeZone("GMT+8") .gte(startTime) .lte(endTime)) ); //分页 from = from <= -1 ? 0 : from; size = size >= 1000 ? 1000 : size; size = size <= 0 ? 10 : size; searchSourceBuilder.from(from); searchSourceBuilder.size(size); //超时 searchSourceBuilder.timeout(new TimeValue(60, TimeUnit.SECONDS)); //searchSourceBuilder.sort(new FieldSortBuilder("timestamp").order(SortOrder.DESC)); searchRequest.source(searchSourceBuilder); // Setting IndicesOptions controls how unavailable indices are resolved and how wildcard expressions are expanded searchRequest.indicesOptions(IndicesOptions.lenientExpandOpen()); SearchResponse search = client.search(searchRequest, RequestOptions.DEFAULT); log.info("search=====" + search); } catch (Exception e) { } } public static void updateIndex(RestHighLevelClient client) { Map<String, Object> jsonMap = new HashMap<>(); jsonMap.put("message", "update"); UpdateRequest request = new UpdateRequest(ES_INDEX, "1").doc(jsonMap); // request.routing("routing"); //路由值 // request.timeout(TimeValue.timeValueSeconds(1)); //设置超时 // request.timeout("1s"); ////以字符串形式设置超时时间 // request.setRefreshPolicy(WriteRequest.RefreshPolicy.WAIT_UNTIL); //以WriteRequest.RefreshPolicy实例形式设置刷新策略 // request.setRefreshPolicy("wait_for");//以字符串形式刷新策略 // request.retryOnConflict(3); //如果要更新的文档在更新操作的获取和索引阶段之间被另一个操作更改，重试更新操作的次数 // request.fetchSource(true); //启用源检索，默认情况下禁用 // String[] includes = new String[]{"updated", "r*"}; // String[] excludes = Strings.EMPTY_ARRAY; // request.fetchSource( // new FetchSourceContext(true, includes, excludes)); //为特定字段配置源包含 // String[] includes = Strings.EMPTY_ARRAY; // String[] excludes = new String[]{"updated"}; // request.fetchSource( // new FetchSourceContext(true, includes, excludes)); //为特定字段配置源排除 // request.setIfSeqNo(2L); //ifSeqNo // request.setIfPrimaryTerm(1L); //ifPrimaryTerm // request.detectNoop(false); //禁用noop检测 // request.scriptedUpsert(true); //指出无论文档是否存在，脚本都必须运行，即如果文档不存在，脚本负责创建文档。 // request.docAsUpsert(true); //指示如果部分文档尚不存在，则必须将其用作upsert文档。 // request.waitForActiveShards(2); //设置在继续更新操作之前必须活动的碎片副本数量。 // request.waitForActiveShards(ActiveShardCount.ALL); //ActiveShardCount的碎片副本数。可选值：ActiveShardCount.ALL, ActiveShardCount.ONE或者 ActiveShardCount.DEFAULT try { UpdateResponse update = client.update(request, RequestOptions.DEFAULT); log.info("update====" + update.getShardInfo()); log.info("update====" + update.status()); } catch (IOException e) { e.printStackTrace(); } } // 检查文档是否存在 public static void checkExists(RestHighLevelClient client) { GetRequest getRequest = new GetRequest( ES_INDEX, //索引 "2"); //文档id getRequest.fetchSourceContext(new FetchSourceContext(false)); //禁用fetching _source. getRequest.storedFields("_none_"); try { boolean exists = client.exists(getRequest, RequestOptions.DEFAULT); log.info("exists====" + exists); } catch (IOException e) { e.printStackTrace(); log.info("e=========" + e.getMessage()); } } // 创建index public static void createIndex(RestHighLevelClient restHighLevelClient) throws IOException { Map<String, Object> jsonMap = new HashMap<>(); jsonMap.put("user", "kimchy"); jsonMap.put("postDate", new Date()); jsonMap.put("message", "trying out Elasticsearch"); IndexRequest indexRequest = new IndexRequest(ES_INDEX) .id("1").source(jsonMap); //以Map形式提供的文档源，可自动转换为JSON格式 // indexRequest.routing("routing"); //路由值 // indexRequest.timeout(TimeValue.timeValueSeconds(1)); //设置超时 // indexRequest.timeout("1s"); ////以字符串形式设置超时时间 // indexRequest.setRefreshPolicy(WriteRequest.RefreshPolicy.WAIT_UNTIL); //以WriteRequest.RefreshPolicy实例形式设置刷新策略 // indexRequest.setRefreshPolicy("wait_for");//以字符串形式刷新策略 // indexRequest.version(2); //文档版本 // indexRequest.versionType(VersionType.EXTERNAL); //文档类型 // indexRequest.opType(DocWriteRequest.OpType.CREATE); //操作类型 // indexRequest.opType("create"); //操作类型 可选create或update // indexRequest.setPipeline("pipeline"); //索引文档之前要执行的摄取管道的名称 IndexResponse index = restHighLevelClient.index(indexRequest, RequestOptions.DEFAULT); log.info("create===" + index); } // 异步获取信息，客户端不可以close public static void getIndex(RestHighLevelClient restHighLevelClient) { GetRequest getRequest = new GetRequest(ES_INDEX); getRequest.id("1"); ActionListener<GetResponse> listener = new ActionListener<GetResponse>() { @Override public void onResponse(GetResponse getResponse) { String index = getResponse.getIndex(); String id = getResponse.getId(); if (getResponse.isExists()) { long version = getResponse.getVersion(); String sourceAsString = getResponse.getSourceAsString(); //以字符串形式检索文档 Map<String, Object> sourceAsMap = getResponse.getSourceAsMap(); //以Map<String, Object>的形式检索文档 byte[] sourceAsBytes = getResponse.getSourceAsBytes(); //以byte[]形式检索文档 log.info("version===" + version); log.info("sourceAsString===" + sourceAsString); log.info("sourceAsMap===" + sourceAsMap); log.info("sourceAsBytes===" + sourceAsBytes.toString()); } else { } log.info("id===" + id); } @Override public void onFailure(Exception e) { log.info("e===========" + e.getMessage()); } }; restHighLevelClient.getAsync(getRequest, RequestOptions.DEFAULT, listener); } }
参考文档：
https://www.cnblogs.com/westlin/p/10909912.html