最简单的es理解 数据库视角看写 ES 加 java正删改查深度分页
目录
mget可以批量的根据index，type，ids三个字段来获取一批数据
批量bulk操作 已存在，覆盖（如果进行更新操作时，params中的数据需要和ES中的数据元素保持一致，不然就会丢失部分数据）
介绍
用数据库角度看es 更容易理解，这里总结些es组件 及 使用java 代码正删改查深度分野 正删改查
es数据结构
1.索引index=数据库
2.类型type = 表 实体类
3.文档document = 行数据 行id不填自动出
4.字段field = 列 可单独配置索引和属性
5.映射mapping = 定义文档结构 字段数据类型，索引，分析器
6.分片shard = 物理存储单元 ，每个索引多个分片，每个分片分步多个节点上
7.副本replica = 分片备份提高可用性查询性能
8.节点Node = 集群中实例，负责存储和查询
9.集群Cluster=单多节点组成共同处理存查
10.分析器Analyzer=处理文本数据分词过滤标准化可自定义便于有效索引和搜索
java 操作
根据script update
/** * 根据script update * * @param indexName indexName * @param typeName typeName * @param params 需要更新的参数 * @param scriptStr scriptStr */ public Long updateByScript(String indexName, String typeName, String id, Map<String, Object> params, String scriptStr) { Script script = new Script(ScriptType.INLINE, Script.DEFAULT_SCRIPT_LANG, scriptStr, params); UpdateRequestBuilder urb = elasticsearchTemplate.getClient().prepareUpdate(indexName, typeName, id); urb.setScript(script); //设置在插入时是否运行脚本 urb.setScriptedUpsert(true); //必须有这个值，否则会报document missing exception(文档不存在时设置文档的_source) urb.setUpsert(new HashMap()); urb.setRetryOnConflict(RETRY_ON_CONFLICT); try { UpdateResponse resp = urb.execute().actionGet(); return resp.getVersion(); } catch (VersionConflictEngineException e) { // log.error("occurred version conflict engine exception", e); throw e; } catch (Exception e) { log.error("occurred error", e); log.warn("updateByScript has error param:{}", JSON.toJSONString(params)); throw new BusinessException("occurred error when update es"); } } mget可以批量的根据index，type，ids三个字段来获取一批数据
/** * mget可以批量的根据index，type，id三个字段来获取一批数据，它不能用来查询，最少得需要知道index 和 id两个字段的值，才能进行get，这一点与query是不一样的。 * * @param indexName 索引名 * @param typeName 表名 * @param docIds 文档编号集合 * @param targetClass 获取数据封装对象 */ public <T> List<T> multiGetDoc(String indexName, String typeName, List<Object> docIds, Class<T> targetClass) { List<T> result = new ArrayList<>(); MultiGetRequestBuilder prepareMultiGet = elasticsearchTemplate.getClient().prepareMultiGet(); docIds.forEach(docId -> prepareMultiGet.add(indexName, typeName, docId.toString())); MultiGetResponse multiGetItemResponses = prepareMultiGet.get(); for (MultiGetItemResponse itemResponse : multiGetItemResponses) { GetResponse response = itemResponse.getResponse(); if (response.isExists()) { String res = response.getSourceAsString(); if (StringUtils.isEmpty(res)) { continue; } try { result.add(JSON.parseObject(res, targetClass)); } catch (Exception e) { log.error("occurred error", e); throw new BusinessException(String.format("occurred error when parse es result to Object:%s", targetClass.getCanonicalName())); } } } return result; }根据文档编号进行获取
/** * 根据文档编号进行获取 * * @param indexName 索引名 * @param typeName 表名 * @param id 文档编号 * @param targetClass 获取数据封装对象 */ public <T> T getById(String indexName, String typeName, String id, Class<T> targetClass) { GetRequest getRequest = new GetRequest(); getRequest.index(indexName).type(typeName).id(id); try { GetResponse getResponse = elasticsearchTemplate.getClient().get(getRequest).get(); String result = getResponse.getSourceAsString(); if (StringUtils.isEmpty(result)) { return null; } return JSON.parseObject(result, targetClass); } catch (Exception e) { log.error("occurred error", e); throw new BusinessException("occurred error when query from es!"); } }验证ES查询对象
/** * 验证ES查询对象 * 抛出异常则未查到，正常则通过 */ private void validationESIndexAndType(ESQueryObj esQueryObj) { Preconditions.checkArgument(!StringUtils.isEmpty(esQueryObj.getIndexName()), "indexName can't be null"); Preconditions.checkArgument(!StringUtils.isEmpty(esQueryObj.getTypeName()), "typeName can't be null"); }根据条件删除，谨慎使用，整不好就把整个索引清空了
/** * 根据条件删除，谨慎使用，整不好就把整个索引清空了 * @param indexName * @param boolQueryBuilder * @return */ public long deleteDocByQuery(String indexName, BoolQueryBuilder boolQueryBuilder) { DeleteByQueryRequestBuilder builder = DeleteByQueryAction.INSTANCE.newRequestBuilder(elasticsearchTemplate.getClient()).refresh(true).filter(boolQueryBuilder).source(indexName); long deleted = builder.get().getDeleted(); return deleted; }删除文档库
/** * 删除文档库 * * @param indexName 索引名 * @param typeName 表名 * @param id 文档编号 */ public long deleteDoc(String indexName, String typeName, String id) { DeleteRequest deleteRequest = new DeleteRequest(); deleteRequest.id(id); deleteRequest.index(indexName); deleteRequest.type(typeName); deleteRequest.setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE); try { DeleteResponse resp = elasticsearchTemplate.getClient().delete(deleteRequest).get(); return resp.getVersion(); } catch (Exception e) { log.error("occurred error", e); throw new BusinessException("occurred error when delete doc from es!"); } }插入文档(如果重复插入就会导致后面的将前面的数据进行覆盖)
/** * 插入文档(如果重复插入就会导致后面的将前面的数据进行覆盖) * * @param indexName 索引名 * @param typeName 表名 * @param id 文档编号 * @param param 文档内容 * @return String */ public String insert(String indexName, String typeName, String id, Map<String, Object> param) { // 指定索引名称，type名称和documentId(documentId可选，不设置则系统自动生成)创建document IndexRequestBuilder indexRequestBuilder = elasticsearchTemplate.getClient().prepareIndex(indexName, typeName).setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE); IndexResponse response = indexRequestBuilder.setSource(param).setId(id).get(); if (response.getResult().equals(DocWriteResponse.Result.CREATED)) { return response.getId(); } return null; }批量bulk操作 已存在，覆盖（如果进行更新操作时，params中的数据需要和ES中的数据元素保持一致，不然就会丢失部分数据）
/** * 批量bulk操作 已存在，覆盖（如果进行更新操作时，params中的数据需要和ES中的数据元素保持一致，不然就会丢失部分数据） * * @param indexName 索引名 * @param typeName 表名 * @param params 一次操作的数据量最好保持在5-15MB大小间。 * @param primaryKeyName params里面主键key,对应的主键ID不能为空 */ public void bulk(String indexName, String typeName, List<Map<String, Object>> params, String primaryKeyName) { BulkRequestBuilder bulkRequest = elasticsearchTemplate.getClient().prepareBulk(); for (Map<String, Object> param : params) { if (param.get(primaryKeyName) == null) { log.error("occurred error when bulk doc"); throw new BusinessException("occurred error when bulk es doc,the ID is null"); } bulkRequest.setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE); bulkRequest.add(elasticsearchTemplate.getClient().prepareIndex(indexName, typeName) .setOpType(DocWriteRequest.OpType.INDEX) .setId(String.valueOf(param.get(primaryKeyName))) .setSource(param)); } bulkRequest.execute().actionGet(); }更新插入操作 upsert操作
/** * upsert操作 * * @param indexName 索引名 * @param typeName 表名 * @param id 文档编号 */ public long upsertDoc(String indexName, String typeName, String id, Map<String, Object> param) { IndexRequest indexRequest = new IndexRequest(indexName, typeName, id); indexRequest.source(param); UpdateRequest updateRequest = new UpdateRequest(); updateRequest.index(indexName); updateRequest.type(typeName); updateRequest.id(id); updateRequest.doc(param); updateRequest.upsert(indexRequest); updateRequest.retryOnConflict(RETRY_ON_CONFLICT); updateRequest.setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE); try { UpdateResponse resp = elasticsearchTemplate.getClient().update(updateRequest).get(); return resp.getVersion(); } catch (Exception e) { log.error("occurred error when update doc", e); throw new BusinessException("occurred error when upsert es doc"); } }通过主键批量更新
/** * 通过主键批量更新 * * @param indexName 索引名称 * @param typeName 类型名称 * @param idToParams 主键ID及对应需要更新的参数和值 */ public void bulkUpdateByIds(String indexName, String typeName, Map<String, Map<String, Object>> idToParams) { BulkRequest bulkRequest = new BulkRequest(); for (Map.Entry<String, Map<String, Object>> entry : idToParams.entrySet()) { String id = entry.getKey(); Map<String, Object> paramFields = entry.getValue(); UpdateRequest updateRequest = new UpdateRequest(); updateRequest.index(indexName); updateRequest.type(typeName); updateRequest.id(id); updateRequest.doc(paramFields); updateRequest.retryOnConflict(RETRY_ON_CONFLICT); bulkRequest.add(updateRequest); } ActionFuture<BulkResponse> future = elasticsearchTemplate.getClient().bulk(bulkRequest); BulkResponse bulkResponse = future.actionGet(); if (bulkResponse.hasFailures()) { log.error("bulkUpdateByIds fail bulkResponse={}", JSON.toJSONString(bulkResponse)); throw new BusinessException("bulkUpdateByIds error"); } }id更新文档
/** * 更新文档库 * * @param indexName 索引名 * @param typeName 表名 * @param id 文档编号 */ public long updateDoc(String indexName, String typeName, String id, Map<String, Object> param) { UpdateRequest updateRequest = new UpdateRequest(); updateRequest.index(indexName); updateRequest.type(typeName); updateRequest.id(id); updateRequest.doc(param); updateRequest.retryOnConflict(RETRY_ON_CONFLICT); updateRequest.setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE); try { UpdateResponse resp = elasticsearchTemplate.getClient().update(updateRequest).get(); return resp.getVersion(); } catch (Exception e) { log.error("occurred error when update doc", e); throw new BusinessException("occurred error when update doc"); } }scrollSearch 使用游标方式搜索， 适用于10000以上的数据获取，不支持排序
/** * 使用游标方式搜索， 适用于10000以上的数据获取，不支持排序 * * @param esQueryObj 查询条件 * @param scrollPageRequest 分页信息 * @param targetClass 需要封装的对象 * @return ScrollPage */ public <T> ScrollPage<T> scrollSearch(ESQueryObj esQueryObj, ScrollPageRequest scrollPageRequest, Class<T> targetClass) { ScrollPage<T> returnPage = new ScrollPage<>(); returnPage.setSize(scrollPageRequest.getPageSize()); SearchResponse scrollResp; // 没有ScrollId 说明是首次请求 if (StringUtils.isBlank(scrollPageRequest.getScrollId())) { SearchRequestBuilder srb = elasticsearchTemplate.getClient().prepareSearch(esQueryObj.getIndexName()).setTypes(esQueryObj.getTypeName()).setScroll(new TimeValue(scrollPageRequest.getScrollSeconds() * 1000)).setQuery(esQueryObj.getQuery()).setSize(scrollPageRequest.getPageSize()).setFetchSource(esQueryObj.getIncludeFields(), esQueryObj.getExcludeFields()); if (!esQueryObj.getSortField().isEmpty()) { Map<String, SortOrder> sortField = esQueryObj.getSortField(); for (String key : sortField.keySet()) { srb.addSort(key, sortField.get(key)); } } scrollResp = srb.execute().actionGet(); } else { // 使用ScrollId获取剩余的数据 scrollResp = elasticsearchTemplate.getClient().prepareSearchScroll(scrollPageRequest.getScrollId()).setScroll(new TimeValue(scrollPageRequest.getScrollSeconds() * 1000)).execute().actionGet(); } SearchHits searchHits = scrollResp.getHits(); long totalCount = searchHits.getTotalHits(); if (totalCount == 0) { return returnPage; } returnPage.setTotalElements((int) totalCount); for (SearchHit hit : searchHits.getHits()) { returnPage.getContent().add(JSON.parseObject(hit.getSourceAsString(), targetClass)); } returnPage.setScrollId(scrollResp.getScrollId()); return returnPage; } //对应分页实体类 import java.io.Serializable; public class ScrollPageRequest implements Serializable { private int pageSize = 100; private long scrollSeconds = 60L; private String scrollId; public ScrollPageRequest() { } public int getPageSize() { return this.pageSize; } public void setPageSize(int pageSize) { this.pageSize = pageSize; } public long getScrollSeconds() { return this.scrollSeconds; } public void setScrollSeconds(long scrollSeconds) { this.scrollSeconds = scrollSeconds; } public String getScrollId() { return this.scrollId; } public void setScrollId(String scrollId) { this.scrollId = scrollId; } } 查表行数
public Long queryCount(String indexName, String typeName) throws BusinessException, BadArgumentException, SystemException { SearchRequestBuilder esSearch = elasticsearchTemplate.getClient().prepareSearch(indexName); esSearch.setTypes(typeName); SearchResponse response = esSearch.execute().actionGet(); SearchHits searchHits = response.getHits(); return searchHits.getTotalHits(); }条件查表行数
public Long queryCountByFilterQuery(ESQueryObj esQueryObj) throws BusinessException, BadArgumentException, SystemException { validationESIndexAndType(esQueryObj); //创建ES查询Request对象 SearchRequestBuilder esSearch = elasticsearchTemplate.getClient().prepareSearch(esQueryObj.getIndexName()); esSearch.setTypes(esQueryObj.getTypeName()).setSearchType(SearchType.QUERY_THEN_FETCH).setFrom(esQueryObj.getFromIndex()).setSize(esQueryObj.getSize()).setFetchSource(esQueryObj.getIncludeFields(), esQueryObj.getExcludeFields()); //执行查询 SearchResponse response = esSearch.execute().actionGet(); return response.getHits().getTotalHits(); }条件加游标SearchAfter深度分页
/** * 根据过滤条件查询出数据列表.需要传递索引和表名 * * @param esQueryObj ES查询对象 * @param targetClass ES结果需要转换的类型 * @return Page<T> */ public <T> Page<T> queryObjectPageByFilterQuery(ESQueryObj esQueryObj, Class<T> targetClass) throws BusinessException, BadArgumentException, SystemException { validationESIndexAndType(esQueryObj); Page<T> page = new Page<>(); long startCountTime = System.currentTimeMillis(); if (esQueryObj.getFromIndex() + esQueryObj.getSize() > DEFAULT_MAX_SIZE) { throw new BadArgumentException("分页深度不能超过10000"); } //创建ES查询Request对象 SearchRequestBuilder esSearch = elasticsearchTemplate.getClient().prepareSearch(esQueryObj.getIndexName()); esSearch.setTypes(esQueryObj.getTypeName()).setSearchType(SearchType.QUERY_THEN_FETCH).setFrom(esQueryObj.getFromIndex()).setSize(esQueryObj.getSize()).setFetchSource(esQueryObj.getIncludeFields(), esQueryObj.getExcludeFields()); //添加查询条件 if (esQueryObj.getQuery() != null) { esSearch.setQuery(esQueryObj.getQuery()); } //添加多级排序 if (esQueryObj.getSortField() != null) { for (Map.Entry<String, SortOrder> entry : esQueryObj.getSortField().entrySet()) { esSearch.addSort(entry.getKey(), entry.getValue()); } } //深分页 if (esQueryObj.getSearchAfter() != null && esQueryObj.getSearchAfter().length != 0) { esSearch.searchAfter(esQueryObj.getSearchAfter()); } //执行查询 SearchResponse response = esSearch.execute().actionGet(); for (SearchHit hit : response.getHits()) { page.getContent().add(JSON.parseObject(hit.getSourceAsString(), targetClass)); } // Object[] sortValues = response.getHits().getAt(9).getSortValues(); // log.info("EsDbUtils queryObjectPageByFilterQuery search:,esQueryObj:{},time consuming：{}ms.", response.getHits().getTotalHits(), (System.currentTimeMillis() - startCountTime)); page.setTotalElements(response.getHits().getTotalHits()); page.setSize(esQueryObj.getSize()); return page; }**分页返回对象***
import java.io.Serializable; import java.util.ArrayList; import java.util.List; public class Page<T> implements Serializable { private long totalElements = 0L; private int size = 10; private int page = 1; private List<T> content = new ArrayList(); public Page() { } public int getSize() { return this.size; } public void setSize(int size) { this.size = size; } public long getTotalElements() { return this.totalElements; } public void setTotalElements(long totalElements) { this.totalElements = totalElements; } public long getTotalPage() { return this.totalElements / (long)this.size + (long)(this.totalElements % (long)this.size == 0L ? 0 : 1); } public List<T> getContent() { return this.content; } public void setContent(List<T> content) { this.content = content; } public int getPage() { return this.page; } public void setPage(int page) { this.page = page; } } ***ES查询对象***
import lombok.Data; import org.apache.commons.lang3.builder.ToStringBuilder; import org.apache.commons.lang3.builder.ToStringStyle; import org.elasticsearch.common.Strings; import org.elasticsearch.index.query.QueryBuilder; import org.elasticsearch.search.sort.SortOrder; import java.util.Map; /** * @description: ES查询对象 * */ @Data public class ESQueryObj { /** * ES索引名 */ private String indexName; /** * ES TYPE名（表名) */ private String typeName; /** * 查询条件组合，类似于 "boolQuery().must( QueryBuilders.termQuery("opTime", "2016-03-30")).must(QueryBuilders.termQuery("operErpID", "xingkai"))" */ private QueryBuilder query; /** * 排序字段键值对，可以添加多个，类似于("opTime","DESC") */ private Map<String, SortOrder> sortField; /** * 取值的起始位置，默认为0 */ private int fromIndex; /** * 返回数据数量，默认为100，最大不能超过100000 */ private int size = 100; /** * 需要返回的字段 */ private String[] includeFields; /** * 不需要返回的字段 */ private String[] excludeFields = Strings.EMPTY_ARRAY; //https://blog.ZEEKLOG.net/yiyiholic/article/details/81661919 /** * 通过提供一个live cursor来规避消耗存储和时间的性能问题,通过上一页的结果帮助检索下一页,用第一次检索的最后一个值作为下一个检索search_after的参数,当使用search_after参数时，from的值必须被设为0或者-1 * 使用时注意必须按照某一字段进行排序 */ private Object[] searchAfter; @Override public String toString() { return ToStringBuilder.reflectionToString(this, ToStringStyle.JSON_STYLE); } } 完整代码
es 相关代码
ESDbUtils es工具类
package com.esm.common; import com.alibaba.fastjson.JSON; import com.google.common.base.Preconditions; import com.commons.data.page.Page; import com.commons.data.page.ScrollPage; import com.commons.data.page.ScrollPageRequest; import com.commons.exception.BadArgumentException; import com.commons.exception.BusinessException; import com.commons.exception.SystemException; import lombok.extern.slf4j.Slf4j; import org.apache.commons.lang3.StringUtils; import org.elasticsearch.action.ActionFuture; import org.elasticsearch.action.DocWriteRequest; import org.elasticsearch.action.DocWriteResponse; import org.elasticsearch.action.bulk.BulkRequest; import org.elasticsearch.action.bulk.BulkRequestBuilder; import org.elasticsearch.action.bulk.BulkResponse; import org.elasticsearch.action.delete.DeleteRequest; import org.elasticsearch.action.delete.DeleteResponse; import org.elasticsearch.action.get.*; import org.elasticsearch.action.index.IndexRequest; import org.elasticsearch.action.index.IndexRequestBuilder; import org.elasticsearch.action.index.IndexResponse; import org.elasticsearch.action.search.SearchRequestBuilder; import org.elasticsearch.action.search.SearchResponse; import org.elasticsearch.action.search.SearchType; import org.elasticsearch.action.support.WriteRequest; import org.elasticsearch.action.update.UpdateRequest; import org.elasticsearch.action.update.UpdateRequestBuilder; import org.elasticsearch.action.update.UpdateResponse; import org.elasticsearch.common.unit.TimeValue; import org.elasticsearch.index.engine.VersionConflictEngineException; import org.elasticsearch.index.query.BoolQueryBuilder; import org.elasticsearch.index.reindex.DeleteByQueryAction; import org.elasticsearch.index.reindex.DeleteByQueryRequestBuilder; import org.elasticsearch.script.Script; import org.elasticsearch.script.ScriptType; import org.elasticsearch.search.SearchHit; import org.elasticsearch.search.SearchHits; import org.elasticsearch.search.sort.SortOrder; import org.springframework.stereotype.Component; import javax.annotation.Resource; import java.text.MessageFormat; import java.util.ArrayList; import java.util.HashMap; import java.util.List; import java.util.Map; /** * ES基础操作工具类 * */ @Slf4j @Component("eSDbUtils") public class ESDbUtils { /** * 获取数据的最大条数 */ private static final int DEFAULT_MAX_SIZE = 10000; /** * 并发更新存在版本冲突是重试次数 */ private static final int RETRY_ON_CONFLICT = 3; /** * ES 客户端工厂类 */ @Resource private ElasticsearchTemplate elasticsearchTemplate; /** * 根据过滤条件查询出数据列表.需要传递索引和表名 * * @param esQueryObj ES查询对象 * @param targetClass ES结果需要转换的类型 * @return Page<T> */ public <T> Page<T> queryObjectPageByFilterQuery(ESQueryObj esQueryObj, Class<T> targetClass) throws BusinessException, BadArgumentException, SystemException { validationESIndexAndType(esQueryObj); Page<T> page = new Page<>(); long startCountTime = System.currentTimeMillis(); if (esQueryObj.getFromIndex() + esQueryObj.getSize() > DEFAULT_MAX_SIZE) { throw new BadArgumentException("分页深度不能超过10000"); } //创建ES查询Request对象 SearchRequestBuilder esSearch = elasticsearchTemplate.getClient().prepareSearch(esQueryObj.getIndexName()); esSearch.setTypes(esQueryObj.getTypeName()).setSearchType(SearchType.QUERY_THEN_FETCH).setFrom(esQueryObj.getFromIndex()).setSize(esQueryObj.getSize()).setFetchSource(esQueryObj.getIncludeFields(), esQueryObj.getExcludeFields()); //添加查询条件 if (esQueryObj.getQuery() != null) { esSearch.setQuery(esQueryObj.getQuery()); } //添加多级排序 if (esQueryObj.getSortField() != null) { for (Map.Entry<String, SortOrder> entry : esQueryObj.getSortField().entrySet()) { esSearch.addSort(entry.getKey(), entry.getValue()); } } //深分页 if (esQueryObj.getSearchAfter() != null && esQueryObj.getSearchAfter().length != 0) { esSearch.searchAfter(esQueryObj.getSearchAfter()); } //执行查询 SearchResponse response = esSearch.execute().actionGet(); for (SearchHit hit : response.getHits()) { page.getContent().add(JSON.parseObject(hit.getSourceAsString(), targetClass)); } // Object[] sortValues = response.getHits().getAt(9).getSortValues(); // log.info("EsDbUtils queryObjectPageByFilterQuery search:,esQueryObj:{},time consuming：{}ms.", response.getHits().getTotalHits(), (System.currentTimeMillis() - startCountTime)); page.setTotalElements(response.getHits().getTotalHits()); page.setSize(esQueryObj.getSize()); return page; } public Long queryCountByFilterQuery(ESQueryObj esQueryObj) throws BusinessException, BadArgumentException, SystemException { validationESIndexAndType(esQueryObj); //创建ES查询Request对象 SearchRequestBuilder esSearch = elasticsearchTemplate.getClient().prepareSearch(esQueryObj.getIndexName()); esSearch.setTypes(esQueryObj.getTypeName()).setSearchType(SearchType.QUERY_THEN_FETCH).setFrom(esQueryObj.getFromIndex()).setSize(esQueryObj.getSize()).setFetchSource(esQueryObj.getIncludeFields(), esQueryObj.getExcludeFields()); //执行查询 SearchResponse response = esSearch.execute().actionGet(); return response.getHits().getTotalHits(); } public Long queryCount(String indexName, String typeName) throws BusinessException, BadArgumentException, SystemException { SearchRequestBuilder esSearch = elasticsearchTemplate.getClient().prepareSearch(indexName); esSearch.setTypes(typeName); SearchResponse response = esSearch.execute().actionGet(); SearchHits searchHits = response.getHits(); return searchHits.getTotalHits(); } /** * 使用游标方式搜索， 适用于10000以上的数据获取，不支持排序 * * @param esQueryObj 查询条件 * @param scrollPageRequest 分页信息 * @param targetClass 需要封装的对象 * @return ScrollPage */ public <T> ScrollPage<T> scrollSearch(ESQueryObj esQueryObj, ScrollPageRequest scrollPageRequest, Class<T> targetClass) { ScrollPage<T> returnPage = new ScrollPage<>(); returnPage.setSize(scrollPageRequest.getPageSize()); SearchResponse scrollResp; // 没有ScrollId 说明是首次请求 if (StringUtils.isBlank(scrollPageRequest.getScrollId())) { SearchRequestBuilder srb = elasticsearchTemplate.getClient().prepareSearch(esQueryObj.getIndexName()).setTypes(esQueryObj.getTypeName()).setScroll(new TimeValue(scrollPageRequest.getScrollSeconds() * 1000)).setQuery(esQueryObj.getQuery()).setSize(scrollPageRequest.getPageSize()).setFetchSource(esQueryObj.getIncludeFields(), esQueryObj.getExcludeFields()); if (!esQueryObj.getSortField().isEmpty()) { Map<String, SortOrder> sortField = esQueryObj.getSortField(); for (String key : sortField.keySet()) { srb.addSort(key, sortField.get(key)); } } scrollResp = srb.execute().actionGet(); } else { // 使用ScrollId获取剩余的数据 scrollResp = elasticsearchTemplate.getClient().prepareSearchScroll(scrollPageRequest.getScrollId()).setScroll(new TimeValue(scrollPageRequest.getScrollSeconds() * 1000)).execute().actionGet(); } SearchHits searchHits = scrollResp.getHits(); long totalCount = searchHits.getTotalHits(); if (totalCount == 0) { return returnPage; } returnPage.setTotalElements((int) totalCount); for (SearchHit hit : searchHits.getHits()) { returnPage.getContent().add(JSON.parseObject(hit.getSourceAsString(), targetClass)); } returnPage.setScrollId(scrollResp.getScrollId()); return returnPage; } /** * 更新文档库 * * @param indexName 索引名 * @param typeName 表名 * @param id 文档编号 */ public long updateDoc(String indexName, String typeName, String id, Map<String, Object> param) { UpdateRequest updateRequest = new UpdateRequest(); updateRequest.index(indexName); updateRequest.type(typeName); updateRequest.id(id); updateRequest.doc(param); updateRequest.retryOnConflict(RETRY_ON_CONFLICT); updateRequest.setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE); try { UpdateResponse resp = elasticsearchTemplate.getClient().update(updateRequest).get(); return resp.getVersion(); } catch (Exception e) { log.error("occurred error when update doc", e); throw new BusinessException("occurred error when update doc"); } } /** * 通过主键批量更新 * * @param indexName 索引名称 * @param typeName 类型名称 * @param idToParams 主键ID及对应需要更新的参数和值 */ public void bulkUpdateByIds(String indexName, String typeName, Map<String, Map<String, Object>> idToParams) { BulkRequest bulkRequest = new BulkRequest(); for (Map.Entry<String, Map<String, Object>> entry : idToParams.entrySet()) { String id = entry.getKey(); Map<String, Object> paramFields = entry.getValue(); UpdateRequest updateRequest = new UpdateRequest(); updateRequest.index(indexName); updateRequest.type(typeName); updateRequest.id(id); updateRequest.doc(paramFields); updateRequest.retryOnConflict(RETRY_ON_CONFLICT); bulkRequest.add(updateRequest); } ActionFuture<BulkResponse> future = elasticsearchTemplate.getClient().bulk(bulkRequest); BulkResponse bulkResponse = future.actionGet(); if (bulkResponse.hasFailures()) { log.error("bulkUpdateByIds fail bulkResponse={}", JSON.toJSONString(bulkResponse)); throw new BusinessException("bulkUpdateByIds error"); } } /** * upsert操作 * * @param indexName 索引名 * @param typeName 表名 * @param id 文档编号 */ public long upsertDoc(String indexName, String typeName, String id, Map<String, Object> param) { IndexRequest indexRequest = new IndexRequest(indexName, typeName, id); indexRequest.source(param); UpdateRequest updateRequest = new UpdateRequest(); updateRequest.index(indexName); updateRequest.type(typeName); updateRequest.id(id); updateRequest.doc(param); updateRequest.upsert(indexRequest); updateRequest.retryOnConflict(RETRY_ON_CONFLICT); updateRequest.setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE); try { UpdateResponse resp = elasticsearchTemplate.getClient().update(updateRequest).get(); return resp.getVersion(); } catch (Exception e) { log.error("occurred error when update doc", e); throw new BusinessException("occurred error when upsert es doc"); } } /** * 批量bulk操作 已存在，覆盖（如果进行更新操作时，params中的数据需要和ES中的数据元素保持一致，不然就会丢失部分数据） * * @param indexName 索引名 * @param typeName 表名 * @param params 一次操作的数据量最好保持在5-15MB大小间。 * @param primaryKeyName params里面主键key,对应的主键ID不能为空 */ public void bulk(String indexName, String typeName, List<Map<String, Object>> params, String primaryKeyName) { BulkRequestBuilder bulkRequest = elasticsearchTemplate.getClient().prepareBulk(); for (Map<String, Object> param : params) { if (param.get(primaryKeyName) == null) { log.error("occurred error when bulk doc"); throw new BusinessException("occurred error when bulk es doc,the ID is null"); } bulkRequest.setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE); bulkRequest.add(elasticsearchTemplate.getClient().prepareIndex(indexName, typeName) .setOpType(DocWriteRequest.OpType.INDEX) .setId(String.valueOf(param.get(primaryKeyName))) .setSource(param)); } bulkRequest.execute().actionGet(); } /** * 插入文档(如果重复插入就会导致后面的将前面的数据进行覆盖) * * @param indexName 索引名 * @param typeName 表名 * @param id 文档编号 * @param param 文档内容 * @return String */ public String insert(String indexName, String typeName, String id, Map<String, Object> param) { // 指定索引名称，type名称和documentId(documentId可选，不设置则系统自动生成)创建document IndexRequestBuilder indexRequestBuilder = elasticsearchTemplate.getClient().prepareIndex(indexName, typeName).setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE); IndexResponse response = indexRequestBuilder.setSource(param).setId(id).get(); if (response.getResult().equals(DocWriteResponse.Result.CREATED)) { return response.getId(); } return null; } /** * 删除文档库 * * @param indexName 索引名 * @param typeName 表名 * @param id 文档编号 */ public long deleteDoc(String indexName, String typeName, String id) { DeleteRequest deleteRequest = new DeleteRequest(); deleteRequest.id(id); deleteRequest.index(indexName); deleteRequest.type(typeName); deleteRequest.setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE); try { DeleteResponse resp = elasticsearchTemplate.getClient().delete(deleteRequest).get(); return resp.getVersion(); } catch (Exception e) { log.error("occurred error", e); throw new BusinessException("occurred error when delete doc from es!"); } } /** * 根据条件删除，谨慎使用，整不好就把整个索引清空了 * @param indexName * @param boolQueryBuilder * @return */ public long deleteDocByQuery(String indexName, BoolQueryBuilder boolQueryBuilder) { DeleteByQueryRequestBuilder builder = DeleteByQueryAction.INSTANCE.newRequestBuilder(elasticsearchTemplate.getClient()).refresh(true).filter(boolQueryBuilder).source(indexName); long deleted = builder.get().getDeleted(); return deleted; } /** * 验证ES查询对象 */ private void validationESIndexAndType(ESQueryObj esQueryObj) { Preconditions.checkArgument(!StringUtils.isEmpty(esQueryObj.getIndexName()), "indexName can't be null"); Preconditions.checkArgument(!StringUtils.isEmpty(esQueryObj.getTypeName()), "typeName can't be null"); } /** * 根据文档编号进行获取 * * @param indexName 索引名 * @param typeName 表名 * @param id 文档编号 * @param targetClass 获取数据封装对象 */ public <T> T getById(String indexName, String typeName, String id, Class<T> targetClass) { GetRequest getRequest = new GetRequest(); getRequest.index(indexName).type(typeName).id(id); try { GetResponse getResponse = elasticsearchTemplate.getClient().get(getRequest).get(); String result = getResponse.getSourceAsString(); if (StringUtils.isEmpty(result)) { return null; } return JSON.parseObject(result, targetClass); } catch (Exception e) { log.error("occurred error", e); throw new BusinessException("occurred error when query from es!"); } } /** * mget可以批量的根据index，type，id三个字段来获取一批数据，它不能用来查询，最少得需要知道index 和 id两个字段的值，才能进行get，这一点与query是不一样的。 * * @param indexName 索引名 * @param typeName 表名 * @param docIds 文档编号集合 * @param targetClass 获取数据封装对象 */ public <T> List<T> multiGetDoc(String indexName, String typeName, List<Object> docIds, Class<T> targetClass) { List<T> result = new ArrayList<>(); MultiGetRequestBuilder prepareMultiGet = elasticsearchTemplate.getClient().prepareMultiGet(); docIds.forEach(docId -> prepareMultiGet.add(indexName, typeName, docId.toString())); MultiGetResponse multiGetItemResponses = prepareMultiGet.get(); for (MultiGetItemResponse itemResponse : multiGetItemResponses) { GetResponse response = itemResponse.getResponse(); if (response.isExists()) { String res = response.getSourceAsString(); if (StringUtils.isEmpty(res)) { continue; } try { result.add(JSON.parseObject(res, targetClass)); } catch (Exception e) { log.error("occurred error", e); throw new BusinessException(String.format("occurred error when parse es result to Object:%s", targetClass.getCanonicalName())); } } } return result; } /** * 根据script update * * @param indexName indexName * @param typeName typeName * @param params 需要更新的参数 * @param scriptStr scriptStr */ public Long updateByScript(String indexName, String typeName, String id, Map<String, Object> params, String scriptStr) { Script script = new Script(ScriptType.INLINE, Script.DEFAULT_SCRIPT_LANG, scriptStr, params); UpdateRequestBuilder urb = elasticsearchTemplate.getClient().prepareUpdate(indexName, typeName, id); urb.setScript(script); //设置在插入时是否运行脚本 urb.setScriptedUpsert(true); //必须有这个值，否则会报document missing exception(文档不存在时设置文档的_source) urb.setUpsert(new HashMap()); urb.setRetryOnConflict(RETRY_ON_CONFLICT); try { UpdateResponse resp = urb.execute().actionGet(); return resp.getVersion(); } catch (VersionConflictEngineException e) { // log.error("occurred version conflict engine exception", e); throw e; } catch (Exception e) { log.error("occurred error", e); log.wran("updateByScript has error param:{}", JSON.toJSONString(params))); throw new BusinessException("occurred error when update es"); } } } ElasticsearchTemplate es链接
package com.esm.common; import com.commons.exception.SystemException; import org.apache.commons.lang3.StringUtils; import org.elasticsearch.client.Client; import org.elasticsearch.common.settings.Settings; import org.elasticsearch.common.transport.TransportAddress; import org.elasticsearch.transport.client.PreBuiltTransportClient; import org.springframework.beans.factory.annotation.Value; import org.springframework.stereotype.Service; import javax.annotation.PostConstruct; import javax.annotation.PreDestroy; import java.net.InetAddress; import java.nio.ByteBuffer; import java.nio.CharBuffer; import java.nio.charset.StandardCharsets; import java.util.Arrays; import java.util.Base64; /** * ES 客户端 */ @Service public class ElasticsearchTemplate { @Value("${es.clusterNodes}") private String clusterNodes; @Value("${es.clusterName}") private String clusterName; //是否扫描集群 @Value("${es.clientTransportSniff}") private Boolean clientTransportSniff; @Value("${es.userName}") private String securityUser; @Value("${es.password}") private String securityPassword; private Client client; //禁用ES设置netty进程数的配置 static { System.setProperty("es.set.netty.runtime.available.processors", "false"); } @PostConstruct public void init() { Settings settings = Settings.builder() .put("cluster.name", clusterName) .put("client.transport.sniff", clientTransportSniff) .put("request.headers.Authorization", basicAuthHeaderValue(securityUser, securityPassword)) //ElasticSearch5分钟内执行脚本编译超过75个，编译太多而拒绝编译。编译是非常耗时的，这是ES的自我保护功能。 //.put("script.max_compilations_rate", "100/1m") .put("script.cache.max_size", 200) .build(); if (StringUtils.isBlank(clusterNodes)) { throw new SystemException("clusterNodes is empty."); } //创建集群client并添加集群节点地址 PreBuiltTransportClient transportClient = new PreBuiltTransportClient(settings); try { String[] nodes = clusterNodes.split(","); for (String node : nodes) { String[] host = node.split(":"); transportClient.addTransportAddress(new TransportAddress(InetAddress.getByName(host[0]), Integer.parseInt(host[1]))); } } catch (Exception e) { throw new SystemException("ElasticsearchTemplate init error", e); } client = transportClient; } /** * 基础的base64生成 * * @param username 用户名 * @param passwd 密码 * @return String */ private static String basicAuthHeaderValue(String username, String passwd) { CharBuffer chars = CharBuffer.allocate(username.length() + passwd.length() + 1); byte[] charBytes = null; try { chars.put(username).put(':').put(passwd.toCharArray()); charBytes = toUtf8Bytes(chars.array()); String basicToken = Base64.getEncoder().encodeToString(charBytes); return "Basic " + basicToken; } finally { Arrays.fill(chars.array(), (char) 0); if (charBytes != null) { Arrays.fill(charBytes, (byte) 0); } } } /** * 编码 */ private static byte[] toUtf8Bytes(char[] chars) { CharBuffer charBuffer = CharBuffer.wrap(chars); ByteBuffer byteBuffer = StandardCharsets.UTF_8.encode(charBuffer); byte[] bytes = Arrays.copyOfRange(byteBuffer.array(), byteBuffer.position(), byteBuffer.limit()); Arrays.fill(byteBuffer.array(), (byte) 0); return bytes; } /** * 服务销毁前关闭ES服务 */ @PreDestroy public void destory() { client.close(); } public Client getClient() { return client; } public void setClient(Client client) { this.client = client; } } ESQueryObj 查询对象
package com.esm.common; import lombok.Data; import org.apache.commons.lang3.builder.ToStringBuilder; import org.apache.commons.lang3.builder.ToStringStyle; import org.elasticsearch.common.Strings; import org.elasticsearch.index.query.QueryBuilder; import org.elasticsearch.search.sort.SortOrder; import java.util.Map; /** * @description: ES查询对象 */ @Data public class ESQueryObj { /** * ES索引名 */ private String indexName; /** * ES TYPE名（表名) */ private String typeName; /** * 查询条件组合，类似于 "boolQuery().must( QueryBuilders.termQuery("opTime", "2016-03-30")).must(QueryBuilders.termQuery("operErpID", "xingkai"))" */ private QueryBuilder query; /** * 排序字段键值对，可以添加多个，类似于("opTime","DESC") */ private Map<String, SortOrder> sortField; /** * 取值的起始位置，默认为0 */ private int fromIndex; /** * 返回数据数量，默认为100，最大不能超过100000 */ private int size = 100; /** * 需要返回的字段 */ private String[] includeFields; /** * 不需要返回的字段 */ private String[] excludeFields = Strings.EMPTY_ARRAY; //https://blog.ZEEKLOG.net/yiyiholic/article/details/81661919 /** * 通过提供一个live cursor来规避消耗存储和时间的性能问题,通过上一页的结果帮助检索下一页,用第一次检索的最后一个值作为下一个检索search_after的参数,当使用search_after参数时，from的值必须被设为0或者-1 * 使用时注意必须按照某一字段进行排序 */ private Object[] searchAfter; @Override public String toString() { return ToStringBuilder.reflectionToString(this, ToStringStyle.JSON_STYLE); } } EsScriptUtil 脚本工具
package com.esm.common; import org.apache.commons.collections.CollectionUtils; import java.text.MessageFormat; import java.util.List; /** * 工具es脚本 */ public class EsScriptUtil { /** * 判断ES上下文中是否包含某个属性,如果不包含,则进行添加(前提是事先定义的mapping中有该字段) * * @param field 属性名称 * @return script */ public static String appendFieldToCtx(String field, String key) { return MessageFormat.format("if(!ctx._source.containsKey(''{0}''))'{'ctx._source.{0}=null}", field, key); } public static String appendArrFieldToCtx(String field, String key) { return MessageFormat.format("if(!ctx._source.containsKey(''{0}''))'{'ctx._source.{0}=[]}", field, key); } /** * 判断ES数组字段中是否包含某个值,如果不包含则进行追加 * * @param field 属性名称 * @return script */ public static String appendValueToArr(String field, String key) { return MessageFormat.format("if(ctx._source.{0}==='''' || ctx._source.{0}=== null || ctx._source.{0}===''null'')'{'ctx._source.{0}=[{1}.{0}];}" + "else if(ctx._source.{0}.indexOf({1}.{0}) === -1)'{'ctx._source.{0}.add({1}.{0});}", field, key); } /** * 判断ES数组字段中是否包含某个值,如果包含则进行删除 * * @param field 属性名称 * @return script */ public static String removeValueFromArr(String field, String key) { return MessageFormat.format("if(ctx._source.{0} !=='''' && ctx._source.{0}!== null && ctx._source.{0} !==''null''&& ctx._source.{0}.indexOf({1}.{0}) != -1)'{'ctx._source.{0}.remove(ctx._source.{0}.indexOf({1}.{0}));}", field, key); } /** * 更新某个字段 * * @param field 属性名称 * @return script */ public static String updateField(String field, String key) { return MessageFormat.format("ctx._source.{0} = {1}.{0};", field, key); } /** * 判断ES对象数组字段中是否包含某个值,如果不包含则进行追加 * * @param field 对象名称 * @param list 对象包含的所有属性名称 * @param key 更新map对象的名称 * @return script */ public static String appendValueToObjectArr(String field, List<String> list, String key) { StringBuilder sb = new StringBuilder(); if (CollectionUtils.isNotEmpty(list)) { sb.append(MessageFormat.format("for(int i=0; i<ctx._source.{0}.size(); i++)", field)).append("{"); for (int i = 0; i < list.size(); i++) { if (i == 0) { sb.append(MessageFormat.format("if (ctx._source.{0}[i][''{1}''] == {2}.{0}.{1}", field, list.get(i), key)); } else { sb.append(MessageFormat.format("&& ctx._source.{0}[i][''{1}'']== {2}.{0}.{1}", field, list.get(i), key)); } } sb.append(MessageFormat.format(")'{'ctx._source.{0}.remove(i);break;}}ctx._source.{0}.add({1}.{0});", field, key)); } else { //无需校验字段是否相等,直接进行添加 sb.append(MessageFormat.format("ctx._source.{0}.add({1}.{0});", field, key)); } return sb.toString(); } /** * 清空数组 */ public static String clearArr(String field) { return MessageFormat.format("ctx._source.{0}.clear();", field); } } 异常
SystemException
package com.commons.exception; public class SystemException extends BaseException { public SystemException(String message) { super(message); } public SystemException(String message, Throwable cause) { super(message, cause); } public SystemException(Throwable cause) { super(cause); } public SystemException() { } public SystemException(int code) { super(code); } public SystemException(int code, String message) { super(code, message); } } BusinessException
package com.commons.exception; public class BusinessException extends BaseException { public BusinessException() { } public BusinessException(String message) { super(message); } public BusinessException(String message, Throwable cause) { super(message, cause); } public BusinessException(Throwable cause) { super(cause); } public BusinessException(int code) { super(code); } public BusinessException(int code, String message) { super(code, message); } } BaseException
package com.commons.exception; public class BaseException extends RuntimeException { private int code; public BaseException() { } public BaseException(String message) { super(message); } public BaseException(String message, Throwable cause) { super(message, cause); } public BaseException(Throwable cause) { super(cause); } public BaseException(int code) { this.code = code; } public BaseException(int code, String message) { super(message); this.code = code; } public int getCode() { return this.code; } } BadArgumentException
package com.commons.exception; public class BadArgumentException extends BaseException { public BadArgumentException() { } public BadArgumentException(String message) { super(message); } public BadArgumentException(String message, Throwable cause) { super(message, cause); } public BadArgumentException(Throwable cause) { super(cause); } public BadArgumentException(int code) { super(code); } public BadArgumentException(int code, String message) { super(code, message); } } 分页代码
Page
package com.commons.data.page; import java.io.Serializable; import java.util.ArrayList; import java.util.List; public class Page<T> implements Serializable { private long totalElements = 0L; private int size = 10; private int page = 1; private List<T> content = new ArrayList(); public Page() { } public int getSize() { return this.size; } public void setSize(int size) { this.size = size; } public long getTotalElements() { return this.totalElements; } public void setTotalElements(long totalElements) { this.totalElements = totalElements; } public long getTotalPage() { return this.totalElements / (long)this.size + (long)(this.totalElements % (long)this.size == 0L ? 0 : 1); } public List<T> getContent() { return this.content; } public void setContent(List<T> content) { this.content = content; } public int getPage() { return this.page; } public void setPage(int page) { this.page = page; } } PageRequest
package com.commons.data.page; import java.io.Serializable; public class PageRequest implements Serializable { private int page = 1; private int pageSize = 10; public PageRequest() { } public int getPage() { return this.page; } public void setPage(int page) { this.page = page; } public int getPageSize() { return this.pageSize; } public void setPageSize(int pageSize) { this.pageSize = pageSize; } public long getOffset() { return (long)(this.page - 1) * (long)this.pageSize; } } ScrollPage
package com.commons.data.page; import java.io.Serializable; import java.util.ArrayList; import java.util.List; public class ScrollPage<T> implements Serializable { private long totalElements = 0L; private int size = 100; private String scrollId; private List<T> content = new ArrayList(); public ScrollPage() { } public int getSize() { return this.size; } public void setSize(int size) { this.size = size; } public long getTotalElements() { return this.totalElements; } public void setTotalElements(long totalElements) { this.totalElements = totalElements; } public long getTotalPage() { return this.totalElements / (long)this.size + (long)(this.totalElements % (long)this.size == 0L ? 0 : 1); } public List<T> getContent() { return this.content; } public void setContent(List<T> content) { this.content = content; } public String getScrollId() { return this.scrollId; } public void setScrollId(String scrollId) { this.scrollId = scrollId; } } ScrollPageRequest
package com.commons.data.page; import java.io.Serializable; public class ScrollPageRequest implements Serializable { private int pageSize = 100; private long scrollSeconds = 60L; private String scrollId; public ScrollPageRequest() { } public int getPageSize() { return this.pageSize; } public void setPageSize(int pageSize) { this.pageSize = pageSize; } public long getScrollSeconds() { return this.scrollSeconds; } public void setScrollSeconds(long scrollSeconds) { this.scrollSeconds = scrollSeconds; } public String getScrollId() { return this.scrollId; } public void setScrollId(String scrollId) { this.scrollId = scrollId; } } 测试
查询
public List<TestEntity> queryNotExpireDiscountTestFormES( Long id) { List<TestEntities> testEntities = new ArrayList<>(); ESQueryObj esQueryObj = new ESQueryObj(); BoolQueryBuilder boolQuery = QueryBuilders.boolQuery(); esQueryObj.setIndexName(ESConstans.INDEX_NAME); esQueryObj.setTypeName(ESConstans.TYPE_NAME); esQueryObj.setFromIndex(0); esQueryObj.setSize(50); if (id != null) { boolQuery.must(QueryBuilders.termQuery("id", id)); } esQueryObj.setQuery(boolQuery); Page<TestIndexBean> tEntityPage = esDbUtils.queryObjectPageByFilterQuery(esQueryObj, TestIndexBean.class); if (CollectionUtils.isNotEmpty(tEntityPage.getContent())) { testEntities = tEntityPage.getContent().stream().map(this::convertEsDataToTestEntity).collect(Collectors.toList()); } logger.info("查询出符合的实体testEntities:{}", JSON.toJSONString(testEntities)); return tEntityPage; } private TestEntity convertEsDataToTestEntity(TestIndexBean testIndexBean) { PromotionEntity promotionEntity = new PromotionEntity(); promotionEntity.setName(testIndexBean.getName()); return promotionEntity; }持续更新
持续更新
持续更新
持续更新
