tianzeOrg
/
TextBigData
8
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

修改ID类型

Browse Source
master
xiaoCJ 9 months ago
parent 9ca1b34729
commit 3d634f432d
4 changed files with 171 additions and 180 deletions
Show Stats Download Patch File Download Diff File

25
ruoyi-admin/src/main/resources/application-druid.yml
Unescape Escape View File

@ -3,19 +3,10 @@
# data: # data:
# mongodb: # mongodb:
# database: makesoft # database: makesoft
## host: 39.108.144.227
## port: 1471
# host: 120.78.220.29 # host: 120.78.220.29
# port: 27017 # port: 27017
# username: makesoft # username: makesoft
# password: makesoft # password: makesoft
## host: 118.31.7.2
## port: 27017
## username: root
## password: root
# #elasticsearch:
# #rest:
# #uris: http://39.108.144.227:9200
# #
# datasource: # datasource:
# type: com.alibaba.druid.pool.DruidDataSource # type: com.alibaba.druid.pool.DruidDataSource
@ -26,12 +17,6 @@
# url: jdbc:mysql://120.78.220.29:3306/makesoft?useUnicode=true&characterEncoding=utf8&zeroDateTimeBehavior=convertToNull&useSSL=true&serverTimezone=GMT%2B8 # url: jdbc:mysql://120.78.220.29:3306/makesoft?useUnicode=true&characterEncoding=utf8&zeroDateTimeBehavior=convertToNull&useSSL=true&serverTimezone=GMT%2B8
# username: root # username: root
# password: Biemo123 # password: Biemo123
## url: jdbc:mysql://39.108.144.227:1637/makesoft?useUnicode=true&characterEncoding=utf8&zeroDateTimeBehavior=convertToNull&useSSL=true&serverTimezone=GMT%2B8
## username: root
## password: Biemo123
## url: jdbc:mysql://118.31.7.2:3306/makesoft?useUnicode=true&characterEncoding=utf8&zeroDateTimeBehavior=convertToNull&useSSL=true&serverTimezone=GMT%2B8
## username: root
## password: sztzjy2017
# # 从库数据源 # # 从库数据源
# slave: # slave:
# # 从数据源开关/默认关闭 # # 从数据源开关/默认关闭
@ -82,15 +67,11 @@
#elasticsearch: #elasticsearch:
# schema: http # schema: http
# address: 120.78.220.29:9200 # address: 120.78.220.29:9200
## address: 39.108.144.227:3135
## address: 118.31.7.2:9200
# connectTimeout: 10000 # connectTimeout: 10000
# socketTimeout: 10000 # socketTimeout: 10000
# connectionRequestTimeout: 10000 # connectionRequestTimeout: 10000
# maxConnectNum: 100 # maxConnectNum: 100
# maxConnectPerRoute: 100 # maxConnectPerRoute: 100
## username: elastic
## password: es@test001
# #
#sso: #sso:
# apiUrl: http://test.vsp.cqooc.com/open/api/v2/oauth/token?appId={APPID}&signature={SIGNATURE}&ticket={TICKET} # apiUrl: http://test.vsp.cqooc.com/open/api/v2/oauth/token?appId={APPID}&signature={SIGNATURE}&ticket={TICKET}
@ -105,9 +86,9 @@
# getReportListUrl: http://test.vsp.cqooc.com/open/api/v2/query/data?appId=APPID&accessToken=ACCESS_TOKEN&signature=SIGNATURE # getReportListUrl: http://test.vsp.cqooc.com/open/api/v2/query/data?appId=APPID&accessToken=ACCESS_TOKEN&signature=SIGNATURE
##################重庆二师配置 #################重庆二师配置
# 项目相关配置 项目相关配置
# 数据源配置 数据源配置
spring: spring:
data: data:
mongodb: mongodb:

2
ruoyi-biemo/src/main/java/com/ruoyi/biemo/business/domain/DocInfo.java
Unescape Escape View File

@ -15,7 +15,7 @@ public class DocInfo extends BiemoEntity {
private static final long serialVersionUID = 1L; private static final long serialVersionUID = 1L;
@EsId @EsId
private String id; private Object id;
@FieldInfo(type = "string",participle = 3) @FieldInfo(type = "string",participle = 3)
private String title; private String title;
//文档类别 1:新闻、2:公众号 //文档类别 1:新闻、2:公众号

5
ruoyi-biemo/src/main/java/com/ruoyi/biemo/business/service/DocInfoService.java
Unescape Escape View File

@ -58,11 +58,10 @@ public class DocInfoService extends EsService<DocInfo> {
@Override @Override
protected DocInfo loadData(SearchSourceBuilder context, SearchHit hit) { protected DocInfo loadData(SearchSourceBuilder context, SearchHit hit) {
String id = hit.getId(); String id = hit.getId();
System.out.println("id:"+id);
System.out.println("TTTT:"+id+"TTTT");
if (Objects.isNull(id) || StringUtil.isEmpty(id)) return null; if (Objects.isNull(id) || StringUtil.isEmpty(id)) return null;
System.out.println("hit.getId()===========>:"+id);
DocInfo byId = mongoHelper.findById(id, DocInfo.class); DocInfo byId = mongoHelper.findById(id, DocInfo.class);
System.out.println("mongo:"+byId.getId()); System.out.println("mongo.FindById结果======》:"+byId.getId());
return mongoHelper.findById(id, DocInfo.class); return mongoHelper.findById(id, DocInfo.class);
} }

319
ruoyi-biemo/src/main/java/com/ruoyi/biemo/elasticsearch/util/EsService.java
Unescape Escape View File

@ -10,14 +10,9 @@ import com.ruoyi.biemo.elasticsearch.entity.FieldMapping;
import com.ruoyi.biemo.elasticsearch.function.GFunction; import com.ruoyi.biemo.elasticsearch.function.GFunction;
import com.ruoyi.biemo.mongodb.config.Constant; import com.ruoyi.biemo.mongodb.config.Constant;
import com.ruoyi.common.exception.CustomException; import com.ruoyi.common.exception.CustomException;
import com.ruoyi.common.exception.ServiceException;
import lombok.Data; import lombok.Data;
import org.apache.commons.lang3.StringUtils; import org.apache.commons.lang3.StringUtils;
import org.apache.http.util.EntityUtils; import org.apache.http.util.EntityUtils;
import org.apache.poi.ss.formula.functions.T;
import org.elasticsearch.action.admin.indices.analyze.AnalyzeAction;
import org.elasticsearch.action.admin.indices.analyze.AnalyzeRequestBuilder;
import org.elasticsearch.action.admin.indices.create.CreateIndexRequestBuilder;
import org.elasticsearch.action.admin.indices.delete.DeleteIndexRequest; import org.elasticsearch.action.admin.indices.delete.DeleteIndexRequest;
import org.elasticsearch.action.bulk.BulkRequest; import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.delete.DeleteRequest; import org.elasticsearch.action.delete.DeleteRequest;
@ -26,11 +21,13 @@ import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse; import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.search.SearchScrollRequest; import org.elasticsearch.action.search.SearchScrollRequest;
import org.elasticsearch.action.support.master.AcknowledgedResponse; import org.elasticsearch.action.support.master.AcknowledgedResponse;
import org.elasticsearch.client.*; import org.elasticsearch.client.Request;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.Response;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.client.indices.CreateIndexRequest; import org.elasticsearch.client.indices.CreateIndexRequest;
import org.elasticsearch.client.indices.CreateIndexResponse; import org.elasticsearch.client.indices.CreateIndexResponse;
import org.elasticsearch.client.indices.GetIndexRequest; import org.elasticsearch.client.indices.GetIndexRequest;
import org.elasticsearch.client.transport.TransportClient;
import org.elasticsearch.common.lucene.search.function.CombineFunction; import org.elasticsearch.common.lucene.search.function.CombineFunction;
import org.elasticsearch.common.lucene.search.function.FunctionScoreQuery; import org.elasticsearch.common.lucene.search.function.FunctionScoreQuery;
import org.elasticsearch.common.settings.Settings; import org.elasticsearch.common.settings.Settings;
@ -50,14 +47,14 @@ import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.elasticsearch.search.sort.SortOrder; import org.elasticsearch.search.sort.SortOrder;
import org.slf4j.Logger; import org.slf4j.Logger;
import org.slf4j.LoggerFactory; import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.data.elasticsearch.annotations.Document; import org.springframework.data.elasticsearch.annotations.Document;
import javax.print.Doc;
import java.io.IOException; import java.io.IOException;
import java.lang.reflect.*; import java.lang.reflect.*;
import java.util.*; import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Objects;
import java.util.stream.Collectors; import java.util.stream.Collectors;
/** /**
@ -67,7 +64,7 @@ import java.util.stream.Collectors;
*/ */
public abstract class EsService<T> { public abstract class EsService<T> {
protected Logger logger = LoggerFactory.getLogger(getClass()); protected Logger logger = LoggerFactory.getLogger(getClass());
// ES 默认最大返回 10000 与 scroll查询有关 // ES 默认最大返回 10000 与 scroll查询有关
int MAX_RESULT_WINDOW = 10000; int MAX_RESULT_WINDOW = 10000;
@ -76,7 +73,7 @@ public abstract class EsService<T> {
// 加载数据超过200 时用批量加载 减少数据库连接 // 加载数据超过200 时用批量加载 减少数据库连接
int BATCH_LOAD_SIZE = 200; int BATCH_LOAD_SIZE = 200;
// ES 滚动查询 有效时间 与 scroll查询有关 // ES 滚动查询 有效时间 与 scroll查询有关
TimeValue SCROLL_TIME = TimeValue.timeValueMinutes(1); TimeValue SCROLL_TIME = TimeValue.timeValueMinutes(1);
private String index; private String index;
@ -85,21 +82,22 @@ public abstract class EsService<T> {
private Method getId; private Method getId;
protected abstract T loadData(SearchSourceBuilder context,SearchHit hit); protected abstract T loadData(SearchSourceBuilder context, SearchHit hit);
/** /**
* 批量加载数据 * 批量加载数据
*
* @param context * @param context
* @param hitArr * @param hitArr
* @return * @return
*/ */
public abstract List<T> batchLoadData(SearchSourceBuilder context,SearchHit[] hitArr); public abstract List<T> batchLoadData(SearchSourceBuilder context, SearchHit[] hitArr);
/** /**
* 索引设置 * 索引设置
*/ */
protected Settings.Builder createSettings(int shards, int replicas){ protected Settings.Builder createSettings(int shards, int replicas) {
// 设置分片数和副本数 // 设置分片数和副本数
return Settings.builder().put("index.number_of_shards", shards).put("index.number_of_replicas", replicas); return Settings.builder().put("index.number_of_shards", shards).put("index.number_of_replicas", replicas);
} }
@ -128,35 +126,37 @@ public abstract class EsService<T> {
} }
} }
} }
/** /**
* 创建mapping * 创建mapping
*
* @param * @param
* @param * @param
* @param clazz 索引类型 * @param clazz 索引类型
*/ */
public boolean createIndexAndCreateMapping( Class clazz, boolean rebuild, int number_of_shards, int number_of_replicas) { public boolean createIndexAndCreateMapping(Class clazz, boolean rebuild, int number_of_shards, int number_of_replicas) {
RestHighLevelClient client = null; RestHighLevelClient client = null;
try{ try {
client = ElasticSearchPoolUtil.getClient(); client = ElasticSearchPoolUtil.getClient();
if (indexExist()) { if (indexExist()) {
if(rebuild){ if (rebuild) {
AcknowledgedResponse deleteIndexResponse = client.indices().delete(new DeleteIndexRequest(index), RequestOptions.DEFAULT); AcknowledgedResponse deleteIndexResponse = client.indices().delete(new DeleteIndexRequest(index), RequestOptions.DEFAULT);
if(deleteIndexResponse.isAcknowledged()){ if (deleteIndexResponse.isAcknowledged()) {
return createIndexAndCreateMapping(index,ElasticSearchUtils.getFieldInfo(clazz), number_of_shards, number_of_replicas); return createIndexAndCreateMapping(index, ElasticSearchUtils.getFieldInfo(clazz), number_of_shards, number_of_replicas);
}else{ } else {
logger.error("删除旧索引数据失败，创建索引mapping失败"); logger.error("删除旧索引数据失败，创建索引mapping失败");
return false; return false;
} }
}else{ } else {
logger.info("不需要删除旧的索引，没有进一步创建索引结构哦~"); logger.info("不需要删除旧的索引，没有进一步创建索引结构哦~");
return true; return true;
} }
}else{ } else {
return createIndexAndCreateMapping(index,ElasticSearchUtils.getFieldInfo(clazz), number_of_shards, number_of_replicas); return createIndexAndCreateMapping(index, ElasticSearchUtils.getFieldInfo(clazz), number_of_shards, number_of_replicas);
} }
}catch (Exception e){ } catch (Exception e) {
throw new RuntimeException(); throw new RuntimeException();
}finally { } finally {
ElasticSearchPoolUtil.returnClient(client); ElasticSearchPoolUtil.returnClient(client);
} }
} }
@ -164,11 +164,12 @@ public abstract class EsService<T> {
/** /**
* 根据信息自动创建索引与mapping * 根据信息自动创建索引与mapping
* 构建mapping描述 * 构建mapping描述
* @param index 索引名称 *
* @param index 索引名称
* @param * @param
* @param fieldMappingList 字段信息 * @param fieldMappingList 字段信息
* @param * @param
* @param number_of_shards 分片数 * @param number_of_shards 分片数
* @param number_of_replicas 副本数 * @param number_of_replicas 副本数
* @return * @return
*/ */
@ -178,78 +179,78 @@ public abstract class EsService<T> {
try { try {
mapping = XContentFactory.jsonBuilder() mapping = XContentFactory.jsonBuilder()
.startObject() .startObject()
.startObject("_source").field("enabled","false").endObject() .startObject("_source").field("enabled", "false").endObject()
.startObject("properties"); //设置自定义字段 .startObject("properties"); //设置自定义字段
for(FieldMapping info : fieldMappingList){ for (FieldMapping info : fieldMappingList) {
String field = info.getField(); String field = info.getField();
String dateType = info.getType(); String dateType = info.getType();
if(dateType == null || "".equals(dateType.trim())){ if (dateType == null || "".equals(dateType.trim())) {
dateType = "string"; dateType = "string";
} }
dateType = dateType.toLowerCase(); dateType = dateType.toLowerCase();
int participle = info.getParticiple(); int participle = info.getParticiple();
if("string".equals(dateType)){ if ("string".equals(dateType)) {
if(participle == 0){ if (participle == 0) {
mapping.startObject(field) mapping.startObject(field)
.field("type","keyword") .field("type", "keyword")
.field("index", false) .field("index", false)
.field("ignore_above", info.getIgnoreAbove()) .field("ignore_above", info.getIgnoreAbove())
.endObject(); .endObject();
} else if(participle == 1) { } else if (participle == 1) {
mapping.startObject(field) mapping.startObject(field)
.field("type","text") .field("type", "text")
.field("analyzer","hanlp") .field("analyzer", "hanlp")
.endObject(); .endObject();
}else if(participle == 2){ } else if (participle == 2) {
mapping.startObject(field) mapping.startObject(field)
.field("type","text") .field("type", "text")
.field("analyzer","hanlp_standard") .field("analyzer", "hanlp_standard")
.endObject(); .endObject();
}else if(participle == 3){ } else if (participle == 3) {
mapping.startObject(field) mapping.startObject(field)
.field("type","text") .field("type", "text")
.field("analyzer","hanlp_index") .field("analyzer", "hanlp_index")
.endObject(); .endObject();
}else if(participle == 4){ } else if (participle == 4) {
mapping.startObject(field) mapping.startObject(field)
.field("type","text") .field("type", "text")
.field("analyzer","hanlp_nlp") .field("analyzer", "hanlp_nlp")
.endObject(); .endObject();
}else if(participle == 5){ } else if (participle == 5) {
mapping.startObject(field) mapping.startObject(field)
.field("type","text") .field("type", "text")
.field("analyzer","hanlp_n_short") .field("analyzer", "hanlp_n_short")
.endObject(); .endObject();
}else if(participle == 6){ } else if (participle == 6) {
mapping.startObject(field) mapping.startObject(field)
.field("type","text") .field("type", "text")
.field("analyzer","hanlp_crf") .field("analyzer", "hanlp_crf")
.endObject(); .endObject();
}else if(participle == 7){ } else if (participle == 7) {
mapping.startObject(field) mapping.startObject(field)
.field("type","text") .field("type", "text")
.field("analyzer","hanlp_speed") .field("analyzer", "hanlp_speed")
.endObject(); .endObject();
} }
}else if("datetime".equals(dateType)){ } else if ("datetime".equals(dateType)) {
mapping.startObject(field) mapping.startObject(field)
.field("type","date") .field("type", "date")
.field("format","yyyy-MM-dd HH:mm:ss||yyyy-MM-dd||epoch_millis") .field("format", "yyyy-MM-dd HH:mm:ss||yyyy-MM-dd||epoch_millis")
.endObject(); .endObject();
}else if ("timestamp".equals(dateType)){ } else if ("timestamp".equals(dateType)) {
mapping.startObject(field) mapping.startObject(field)
.field("type","date") .field("type", "date")
.field("format","strict_date_optional_time||epoch_millis") .field("format", "strict_date_optional_time||epoch_millis")
.endObject(); .endObject();
}else if("float".equals(dateType)||"double".equals(dateType)){ } else if ("float".equals(dateType) || "double".equals(dateType)) {
mapping.startObject(field) mapping.startObject(field)
.field("type","scaled_float") .field("type", "scaled_float")
.field("scaling_factor",100) .field("scaling_factor", 100)
.endObject(); .endObject();
}else { } else {
mapping.startObject(field) mapping.startObject(field)
.field("type",dateType) .field("type", dateType)
.field("index",true) .field("index", true)
.endObject(); .endObject();
} }
} }
@ -257,24 +258,23 @@ public abstract class EsService<T> {
.endObject(); .endObject();
CreateIndexRequest createIndexRequest = new CreateIndexRequest(index); CreateIndexRequest createIndexRequest = new CreateIndexRequest(index);
// 设置分片数和副本数 // 设置分片数和副本数
createIndexRequest.settings(createSettings(number_of_shards,number_of_replicas)); createIndexRequest.settings(createSettings(number_of_shards, number_of_replicas));
// 构建对应index的mapping // 构建对应index的mapping
createIndexRequest.mapping(mapping); createIndexRequest.mapping(mapping);
client = ElasticSearchPoolUtil.getClient(); client = ElasticSearchPoolUtil.getClient();
// 发送创建index的请求 // 发送创建index的请求
CreateIndexResponse createIndexResponse = client.indices().create(createIndexRequest,RequestOptions.DEFAULT); CreateIndexResponse createIndexResponse = client.indices().create(createIndexRequest, RequestOptions.DEFAULT);
return createIndexResponse.isAcknowledged(); return createIndexResponse.isAcknowledged();
} catch (Exception e) { } catch (Exception e) {
logger.error("根据信息自动创建索引与mapping创建失败，失败信息为:"+ e.getMessage()); logger.error("根据信息自动创建索引与mapping创建失败，失败信息为:" + e.getMessage());
throw new RuntimeException(); throw new RuntimeException();
}finally { } finally {
ElasticSearchPoolUtil.returnClient(client); ElasticSearchPoolUtil.returnClient(client);
} }
} }
public boolean indexExist() throws Exception {
public boolean indexExist() throws Exception{
RestHighLevelClient client = null; RestHighLevelClient client = null;
GetIndexRequest request = new GetIndexRequest(index); GetIndexRequest request = new GetIndexRequest(index);
request.local(false); request.local(false);
@ -302,20 +302,20 @@ public abstract class EsService<T> {
client.index(request, RequestOptions.DEFAULT); client.index(request, RequestOptions.DEFAULT);
} catch (Exception e) { } catch (Exception e) {
throw new RuntimeException(e); throw new RuntimeException(e);
}finally { } finally {
ElasticSearchPoolUtil.returnClient(client); ElasticSearchPoolUtil.returnClient(client);
} }
} }
public void insertBatch(List<T> list) { public void insertBatch(List<T> list) {
if(list==null||list.size()==0){ if (list == null || list.size() == 0) {
return; return;
} }
RestHighLevelClient client = null; RestHighLevelClient client = null;
try { try {
client = ElasticSearchPoolUtil.getClient(); client = ElasticSearchPoolUtil.getClient();
if(!indexExist()){ if (!indexExist()) {
createIndexAndCreateMapping(list.get(0).getClass(),true,1,0); createIndexAndCreateMapping(list.get(0).getClass(), true, 1, 0);
} }
BulkRequest request = new BulkRequest(); BulkRequest request = new BulkRequest();
list.forEach(item -> { list.forEach(item -> {
@ -331,7 +331,7 @@ public abstract class EsService<T> {
client.bulk(request, RequestOptions.DEFAULT); client.bulk(request, RequestOptions.DEFAULT);
} catch (Exception e) { } catch (Exception e) {
throw new RuntimeException(e); throw new RuntimeException(e);
}finally { } finally {
ElasticSearchPoolUtil.returnClient(client); ElasticSearchPoolUtil.returnClient(client);
} }
} }
@ -345,10 +345,11 @@ public abstract class EsService<T> {
client.bulk(request, RequestOptions.DEFAULT); client.bulk(request, RequestOptions.DEFAULT);
} catch (Exception e) { } catch (Exception e) {
throw new RuntimeException(e); throw new RuntimeException(e);
}finally { } finally {
ElasticSearchPoolUtil.returnClient(client); ElasticSearchPoolUtil.returnClient(client);
} }
} }
public List<T> search(SearchSourceBuilder builder) { public List<T> search(SearchSourceBuilder builder) {
SearchRequest request = new SearchRequest(index); SearchRequest request = new SearchRequest(index);
request.source(builder); request.source(builder);
@ -360,30 +361,30 @@ public abstract class EsService<T> {
List<T> res = new ArrayList<>(hits.length); List<T> res = new ArrayList<>(hits.length);
for (SearchHit hit : hits) { for (SearchHit hit : hits) {
//res.add(JSON.parseObject(hit.getSourceAsString(), entity)); //res.add(JSON.parseObject(hit.getSourceAsString(), entity));
res.add(loadData(builder,hit)); res.add(loadData(builder, hit));
} }
return res; return res;
} catch (Exception e) { } catch (Exception e) {
throw new RuntimeException(e); throw new RuntimeException(e);
}finally { } finally {
ElasticSearchPoolUtil.returnClient(client); ElasticSearchPoolUtil.returnClient(client);
} }
} }
//分页查询 根据需要返回自定义加载数据 比如从mongodb mysql 等 //分页查询 根据需要返回自定义加载数据 比如从mongodb mysql 等
public Page<T> searchPage(SearchSourceBuilder builder,boolean searchFrom) { public Page<T> searchPage(SearchSourceBuilder builder, boolean searchFrom) {
try { try {
List<SearchHit> content = new ArrayList<>(); List<SearchHit> content = new ArrayList<>();
Long total = search(content,builder); Long total = search(content, builder);
System.out.println("*****:"+content.get(0).getId()); System.out.println("*****:" + content.get(0).getId());
List<T> res = new ArrayList<>(); List<T> res = new ArrayList<>();
if(searchFrom){ if (searchFrom) {
res = sequenceLoadData(content); res = sequenceLoadData(content);
}else{ } else {
res = esLoadData(content); res = esLoadData(content);
} }
Integer pageNo = builder.from(); Integer pageNo = builder.from();
Page<T> page=new Page<>(pageNo<=0?1:pageNo/builder.size()+1,builder.size()); Page<T> page = new Page<>(pageNo <= 0 ? 1 : pageNo / builder.size() + 1, builder.size());
page.setRows(res); page.setRows(res);
page.setTotal(total); page.setTotal(total);
return page; return page;
@ -392,13 +393,15 @@ public abstract class EsService<T> {
} }
} }
protected List<T> sequenceLoadData(List<SearchHit> content){ protected List<T> sequenceLoadData(List<SearchHit> content) {
return content.stream().map(x -> loadData(null, x)).filter(x -> x != null).collect(Collectors.toList()); return content.stream().map(x -> loadData(null, x)).filter(x -> x != null).collect(Collectors.toList());
} }
protected List<T> esLoadData(List<SearchHit> content){
protected List<T> esLoadData(List<SearchHit> content) {
return content.stream().map(x -> JSON.parseObject(x.getSourceAsString(), entity)).filter(x -> x != null).collect(Collectors.toList()); return content.stream().map(x -> JSON.parseObject(x.getSourceAsString(), entity)).filter(x -> x != null).collect(Collectors.toList());
} }
protected long search(List<SearchHit> content, SearchSourceBuilder searchSourceBuilder){
protected long search(List<SearchHit> content, SearchSourceBuilder searchSourceBuilder) {
// 用于判断 是否 返回所有数据 // 用于判断 是否 返回所有数据
boolean flag = true; boolean flag = true;
// 总数量 // 总数量
@ -410,7 +413,7 @@ public abstract class EsService<T> {
// 已经处理过了 // 已经处理过了
int from = pageNo; int from = pageNo;
// 待查询页面内第一条记录的下标 // 待查询页面内第一条记录的下标
int firstRowNum = from + 1 ; int firstRowNum = from + 1;
// 最后一行 // 最后一行
int lastRowNum = from + pageSize; int lastRowNum = from + pageSize;
// 创建查询请求 // 创建查询请求
@ -418,12 +421,12 @@ public abstract class EsService<T> {
// 构建SearchSourceBuilder // 构建SearchSourceBuilder
searchRequest.source(searchSourceBuilder); searchRequest.source(searchSourceBuilder);
// 不分页 最多返回10000条 // 不分页 最多返回10000条
if(pageNo==0 && pageSize==0){ if (pageNo == 0 && pageSize == 0) {
searchSourceBuilder.from(0).size(MAX_RESULT_WINDOW); searchSourceBuilder.from(0).size(MAX_RESULT_WINDOW);
}else{ } else {
// 分页 // 分页
// 小于 MAX_RESULT_WINDOW 10000 // 小于 MAX_RESULT_WINDOW 10000
if(lastRowNum<=MAX_RESULT_WINDOW){ if (lastRowNum <= MAX_RESULT_WINDOW) {
searchSourceBuilder.from(from).size(pageSize); searchSourceBuilder.from(from).size(pageSize);
flag = false; flag = false;
} }
@ -435,25 +438,27 @@ public abstract class EsService<T> {
// 查询数据超过第10000 // 查询数据超过第10000
if (lastRowNum > MAX_RESULT_WINDOW) { if (lastRowNum > MAX_RESULT_WINDOW) {
// 滚动搜索 // 滚动搜索
total = searchScroll(content,firstRowNum,lastRowNum,searchSourceBuilder,searchRequest); total = searchScroll(content, firstRowNum, lastRowNum, searchSourceBuilder, searchRequest);
}else{ } else {
searchResponse = client.search(searchRequest, RequestOptions.DEFAULT); searchResponse = client.search(searchRequest, RequestOptions.DEFAULT);
SearchHits hits = searchResponse.getHits(); SearchHits hits = searchResponse.getHits();
System.out.println(index); System.out.println(index);
SearchHit[] hits1 = hits.getHits(); SearchHit[] hits1 = hits.getHits();
System.out.println(hits1[0].getId()); if (hits1[0] != null) {
total = hits.getTotalHits().value; System.out.println(hits1[0].getId());
}
total = hits.getTotalHits().value;
// 不分页情况下 总条数超过10000 // 不分页情况下 总条数超过10000
if(total>MAX_RESULT_WINDOW&&flag){ if (total > MAX_RESULT_WINDOW && flag) {
searchScroll(content,firstRowNum,(int)total,searchSourceBuilder,searchRequest); searchScroll(content, firstRowNum, (int) total, searchSourceBuilder, searchRequest);
}else{ } else {
content = loadDataByResponse(content, searchResponse); content = loadDataByResponse(content, searchResponse);
} }
} }
} catch (Exception e) { } catch (Exception e) {
logger.error("查询索引错误，错误如下：{}",e); logger.error("查询索引错误，错误如下：{}", e);
throw new CustomException("查询索引出错",e); throw new CustomException("查询索引出错", e);
}finally { } finally {
ElasticSearchPoolUtil.returnClient(client); ElasticSearchPoolUtil.returnClient(client);
} }
return total; return total;
@ -462,6 +467,7 @@ public abstract class EsService<T> {
/** /**
* 通过响应加载hit数据 * 通过响应加载hit数据
*
* @param searchHitList * @param searchHitList
* @param searchResponse * @param searchResponse
* @return * @return
@ -470,14 +476,14 @@ public abstract class EsService<T> {
SearchHits hits = searchResponse.getHits(); SearchHits hits = searchResponse.getHits();
// 条数过多 批量加载 减少数据库连接 // 条数过多 批量加载 减少数据库连接
SearchHit[] hitArr = hits.getHits(); SearchHit[] hitArr = hits.getHits();
if(Objects.nonNull(hitArr)&&hitArr.length>=BATCH_LOAD_SIZE){ if (Objects.nonNull(hitArr) && hitArr.length >= BATCH_LOAD_SIZE) {
searchHitList.addAll(Arrays.asList(hitArr)); searchHitList.addAll(Arrays.asList(hitArr));
}else{ } else {
//逐条加载 //逐条加载
for (SearchHit hit : hitArr) { for (SearchHit hit : hitArr) {
String id=hit.getId(); String id = hit.getId();
if (StringUtils.isBlank(id)){ if (StringUtils.isBlank(id)) {
continue; continue;
} }
searchHitList.add(hit); searchHitList.add(hit);
@ -489,6 +495,7 @@ public abstract class EsService<T> {
/** /**
* scroll 查询 返回总条数 * scroll 查询 返回总条数
*
* @param searchHitList * @param searchHitList
* @param firstRowNum * @param firstRowNum
* @param lastRowNum * @param lastRowNum
@ -496,7 +503,7 @@ public abstract class EsService<T> {
* @param searchRequest * @param searchRequest
* @return * @return
*/ */
private long searchScroll(List<SearchHit> searchHitList,int firstRowNum,int lastRowNum,SearchSourceBuilder searchSourceBuilder,SearchRequest searchRequest){ private long searchScroll(List<SearchHit> searchHitList, int firstRowNum, int lastRowNum, SearchSourceBuilder searchSourceBuilder, SearchRequest searchRequest) {
long total = 0L; long total = 0L;
SearchResponse searchResponse; SearchResponse searchResponse;
// 初始位置 // 初始位置
@ -504,11 +511,11 @@ public abstract class EsService<T> {
// 结束位置 // 结束位置
int endPosition; int endPosition;
// 页面大小 // 页面大小
int pageSize = lastRowNum-firstRowNum+1; int pageSize = lastRowNum - firstRowNum + 1;
// 第一条数据所在滚动指针 // 第一条数据所在滚动指针
int firstScrollCursor = firstRowNum/MAX_RESULT_WINDOW; int firstScrollCursor = firstRowNum / MAX_RESULT_WINDOW;
// 最后一条数据所在指针 // 最后一条数据所在指针
int lastScrollCursor = lastRowNum/MAX_RESULT_WINDOW; int lastScrollCursor = lastRowNum / MAX_RESULT_WINDOW;
// 滚动次数 // 滚动次数
int scrollCount = 0; int scrollCount = 0;
// 滚动id // 滚动id
@ -521,37 +528,37 @@ public abstract class EsService<T> {
// scroll 查询 获取前10000 条数据和scrollId // scroll 查询 获取前10000 条数据和scrollId
searchResponse = client.search(searchRequest, RequestOptions.DEFAULT); searchResponse = client.search(searchRequest, RequestOptions.DEFAULT);
SearchHits hits = searchResponse.getHits(); SearchHits hits = searchResponse.getHits();
total = hits.getTotalHits().value; total = hits.getTotalHits().value;
// 1分钟 内的scrollId 第一次 // 1分钟 内的scrollId 第一次
scrollId = searchResponse.getScrollId(); scrollId = searchResponse.getScrollId();
// 部分数据在10000 以内 先获取一部分 // 部分数据在10000 以内 先获取一部分
if(firstRowNum<MAX_RESULT_WINDOW){ if (firstRowNum < MAX_RESULT_WINDOW) {
startPosition = firstRowNum - scrollCount*MAX_RESULT_WINDOW>0?firstRowNum - scrollCount*MAX_RESULT_WINDOW:0; startPosition = firstRowNum - scrollCount * MAX_RESULT_WINDOW > 0 ? firstRowNum - scrollCount * MAX_RESULT_WINDOW : 0;
endPosition = lastRowNum-scrollCount*MAX_RESULT_WINDOW>MAX_RESULT_WINDOW?MAX_RESULT_WINDOW:lastRowNum-scrollCount*MAX_RESULT_WINDOW; endPosition = lastRowNum - scrollCount * MAX_RESULT_WINDOW > MAX_RESULT_WINDOW ? MAX_RESULT_WINDOW : lastRowNum - scrollCount * MAX_RESULT_WINDOW;
loopLoadDataByPosition(searchHitList, startPosition, endPosition, hits); loopLoadDataByPosition(searchHitList, startPosition, endPosition, hits);
} }
if(StringUtils.isNotEmpty(scrollId)){ if (StringUtils.isNotEmpty(scrollId)) {
// 指针还没有到最后的指针 // 指针还没有到最后的指针
while (scrollCount<lastScrollCursor){ while (scrollCount < lastScrollCursor) {
searchResponse = searchScroll(scrollId); searchResponse = searchScroll(scrollId);
scrollId = searchResponse.getScrollId(); scrollId = searchResponse.getScrollId();
scrollCount++; scrollCount++;
// //
if(scrollCount>=firstScrollCursor){ if (scrollCount >= firstScrollCursor) {
// 从hits取对应数据 剩余容量 // 从hits取对应数据 剩余容量
int capacity = pageSize-searchHitList.size(); int capacity = pageSize - searchHitList.size();
if(capacity>0){ if (capacity > 0) {
hits = searchResponse.getHits(); hits = searchResponse.getHits();
// 命中数量 // 命中数量
int hitCount = hits.getHits().length; int hitCount = hits.getHits().length;
startPosition = firstRowNum - scrollCount*MAX_RESULT_WINDOW>0?firstRowNum - scrollCount*MAX_RESULT_WINDOW:0; startPosition = firstRowNum - scrollCount * MAX_RESULT_WINDOW > 0 ? firstRowNum - scrollCount * MAX_RESULT_WINDOW : 0;
endPosition = lastRowNum-scrollCount*MAX_RESULT_WINDOW>MAX_RESULT_WINDOW?MAX_RESULT_WINDOW:lastRowNum-scrollCount*MAX_RESULT_WINDOW; endPosition = lastRowNum - scrollCount * MAX_RESULT_WINDOW > MAX_RESULT_WINDOW ? MAX_RESULT_WINDOW : lastRowNum - scrollCount * MAX_RESULT_WINDOW;
// 超过命中次数 取命中次数 // 超过命中次数 取命中次数
endPosition = endPosition>hitCount?hitCount:endPosition; endPosition = endPosition > hitCount ? hitCount : endPosition;
if(scrollCount==firstScrollCursor) if (scrollCount == firstScrollCursor)
startPosition--; startPosition--;
if(firstScrollCursor!=lastScrollCursor&&scrollCount==lastScrollCursor) if (firstScrollCursor != lastScrollCursor && scrollCount == lastScrollCursor)
endPosition=capacity>MAX_RESULT_WINDOW?MAX_RESULT_WINDOW:capacity-startPosition; endPosition = capacity > MAX_RESULT_WINDOW ? MAX_RESULT_WINDOW : capacity - startPosition;
loopLoadDataByPosition(searchHitList, startPosition, endPosition, hits); loopLoadDataByPosition(searchHitList, startPosition, endPosition, hits);
} }
} }
@ -560,7 +567,7 @@ public abstract class EsService<T> {
} }
} catch (Exception e) { } catch (Exception e) {
e.printStackTrace(); e.printStackTrace();
}finally { } finally {
ElasticSearchPoolUtil.returnClient(client); ElasticSearchPoolUtil.returnClient(client);
} }
return total; return total;
@ -569,43 +576,46 @@ public abstract class EsService<T> {
/** /**
* 滚动搜索 * 滚动搜索
*
* @param scrollId 滚动Id * @param scrollId 滚动Id
* @return * @return
* @throws IOException * @throws IOException
*/ */
protected SearchResponse searchScroll(String scrollId) throws Exception{ protected SearchResponse searchScroll(String scrollId) throws Exception {
RestHighLevelClient client = null; RestHighLevelClient client = null;
client = ElasticSearchPoolUtil.getClient(); client = ElasticSearchPoolUtil.getClient();
SearchScrollRequest searchScrollRequest = new SearchScrollRequest(scrollId); SearchScrollRequest searchScrollRequest = new SearchScrollRequest(scrollId);
searchScrollRequest.scroll(SCROLL_TIME); searchScrollRequest.scroll(SCROLL_TIME);
SearchResponse response = client.scroll(searchScrollRequest, RequestOptions.DEFAULT); SearchResponse response = client.scroll(searchScrollRequest, RequestOptions.DEFAULT);
ElasticSearchPoolUtil.returnClient(client); ElasticSearchPoolUtil.returnClient(client);
return response; return response;
} }
/** /**
* 循环批量加载数据 * 循环批量加载数据
*
* @param context * @param context
* @param startPosition * @param startPosition
* @param endPosition * @param endPosition
* @param hits * @param hits
*/ */
private List<T> loopLoadDataByPosition( SearchSourceBuilder context, int startPosition, int endPosition, SearchHits hits) { private List<T> loopLoadDataByPosition(SearchSourceBuilder context, int startPosition, int endPosition, SearchHits hits) {
int length = hits.getHits().length; int length = hits.getHits().length;
int destLength = endPosition-startPosition; int destLength = endPosition - startPosition;
destLength = destLength>length?length:destLength; destLength = destLength > length ? length : destLength;
SearchHit[] hitArr = new SearchHit[destLength]; SearchHit[] hitArr = new SearchHit[destLength];
System.arraycopy(hits.getHits(),startPosition,hitArr,0,hitArr.length); System.arraycopy(hits.getHits(), startPosition, hitArr, 0, hitArr.length);
return batchLoadData(context,hitArr); return batchLoadData(context, hitArr);
} }
private List<SearchHit> loopLoadDataByPosition(List<SearchHit> searchHitList, int startPosition, int endPosition, SearchHits hits) { private List<SearchHit> loopLoadDataByPosition(List<SearchHit> searchHitList, int startPosition, int endPosition, SearchHits hits) {
int length = hits.getHits().length; int length = hits.getHits().length;
int destLength = endPosition-startPosition; int destLength = endPosition - startPosition;
destLength = destLength>length?length:destLength; destLength = destLength > length ? length : destLength;
SearchHit[] hitArr = new SearchHit[destLength]; SearchHit[] hitArr = new SearchHit[destLength];
System.arraycopy(hits.getHits(),startPosition,hitArr,0,hitArr.length); System.arraycopy(hits.getHits(), startPosition, hitArr, 0, hitArr.length);
searchHitList.addAll(Arrays.asList(hitArr)); searchHitList.addAll(Arrays.asList(hitArr));
return searchHitList; return searchHitList;
} }
@ -617,7 +627,7 @@ public abstract class EsService<T> {
client.indices().delete(new DeleteIndexRequest(index), RequestOptions.DEFAULT); client.indices().delete(new DeleteIndexRequest(index), RequestOptions.DEFAULT);
} catch (Exception e) { } catch (Exception e) {
throw new RuntimeException(e); throw new RuntimeException(e);
}finally { } finally {
ElasticSearchPoolUtil.returnClient(client); ElasticSearchPoolUtil.returnClient(client);
} }
} }
@ -633,7 +643,7 @@ public abstract class EsService<T> {
client.deleteByQuery(request, RequestOptions.DEFAULT); client.deleteByQuery(request, RequestOptions.DEFAULT);
} catch (Exception e) { } catch (Exception e) {
throw new RuntimeException(e); throw new RuntimeException(e);
}finally { } finally {
ElasticSearchPoolUtil.returnClient(client); ElasticSearchPoolUtil.returnClient(client);
} }
} }
@ -668,7 +678,7 @@ public abstract class EsService<T> {
public EsQueryBuilder getDoc_InfoBy_Id(GFunction<? extends T, Object> gFunction, Object value) { public EsQueryBuilder getDoc_InfoBy_Id(GFunction<? extends T, Object> gFunction, Object value) {
String field = gFunction.field(); String field = gFunction.field();
field="_"+field; field = "_" + field;
this.searchSourceBuilder.postFilter(QueryBuilders.termQuery(field, value)); this.searchSourceBuilder.postFilter(QueryBuilders.termQuery(field, value));
this.boolQueryBuilder.filter(QueryBuilders.termQuery(field, value)); this.boolQueryBuilder.filter(QueryBuilders.termQuery(field, value));
return this; return this;
@ -677,17 +687,17 @@ public abstract class EsService<T> {
public EsQueryBuilder eqAll(T t) { public EsQueryBuilder eqAll(T t) {
Field[] fields = t.getClass().getDeclaredFields(); Field[] fields = t.getClass().getDeclaredFields();
if(fields!=null&&fields.length>0){ if (fields != null && fields.length > 0) {
BoolQueryBuilder funScoreBuilders = QueryBuilders.boolQuery(); BoolQueryBuilder funScoreBuilders = QueryBuilders.boolQuery();
for(Field field : fields){ for (Field field : fields) {
field.setAccessible(true); field.setAccessible(true);
String name = field.getName(); String name = field.getName();
try { try {
Object value = field.get(t); Object value = field.get(t);
if(value==null||name==null||Constant.SERIALVERSIONUID.equals(name)){ if (value == null || name == null || Constant.SERIALVERSIONUID.equals(name)) {
continue; continue;
} }
funScoreBuilders = boolTermQueryBuild( funScoreBuilders,name,value); funScoreBuilders = boolTermQueryBuild(funScoreBuilders, name, value);
} catch (IllegalAccessException e) { } catch (IllegalAccessException e) {
e.printStackTrace(); e.printStackTrace();
} }
@ -701,14 +711,15 @@ public abstract class EsService<T> {
private BoolQueryBuilder boolTermQueryBuild(BoolQueryBuilder boolQueryBuilder, String key, Object value) { private BoolQueryBuilder boolTermQueryBuild(BoolQueryBuilder boolQueryBuilder, String key, Object value) {
return boolQueryBuilder.must( return boolQueryBuilder.must(
QueryBuilders.functionScoreQuery( QueryBuilders.functionScoreQuery(
QueryBuilders.termQuery(key, value), ScoreFunctionBuilders.weightFactorFunction(1) QueryBuilders.termQuery(key, value), ScoreFunctionBuilders.weightFactorFunction(1)
) )
.setMinScore(0f) .setMinScore(0f)
.maxBoost(50f) .maxBoost(50f)
.scoreMode(FunctionScoreQuery.ScoreMode.MAX) .scoreMode(FunctionScoreQuery.ScoreMode.MAX)
.boostMode(CombineFunction.REPLACE) .boostMode(CombineFunction.REPLACE)
); );
} }
public EsQueryBuilder notEq(GFunction<? extends T, Object> gFunction, Object value) { public EsQueryBuilder notEq(GFunction<? extends T, Object> gFunction, Object value) {
String field = gFunction.field(); String field = gFunction.field();
this.searchSourceBuilder.postFilter(QueryBuilders.boolQuery().mustNot(QueryBuilders.termQuery(field, value))); this.searchSourceBuilder.postFilter(QueryBuilders.boolQuery().mustNot(QueryBuilders.termQuery(field, value)));
@ -865,8 +876,8 @@ public abstract class EsService<T> {
return search(this.searchSourceBuilder); return search(this.searchSourceBuilder);
} }
public Page<T> queryPage(boolean searchfrom){ public Page<T> queryPage(boolean searchfrom) {
return searchPage(this.searchSourceBuilder,searchfrom); return searchPage(this.searchSourceBuilder, searchfrom);
} }
public void delete() { public void delete() {
@ -891,7 +902,7 @@ public abstract class EsService<T> {
} }
//分词 返回分词结果 //分词 返回分词结果
public List<String> analyzeTitle(DocInfo docInfo){ public List<String> analyzeTitle(DocInfo docInfo) {
try { try {
RestHighLevelClient client = ElasticSearchPoolUtil.getClient(); RestHighLevelClient client = ElasticSearchPoolUtil.getClient();
@ -903,7 +914,7 @@ public abstract class EsService<T> {
Response response = client.getLowLevelClient().performRequest(request); Response response = client.getLowLevelClient().performRequest(request);
JSONObject tokens = JSONObject.parseObject(EntityUtils.toString(response.getEntity())); JSONObject tokens = JSONObject.parseObject(EntityUtils.toString(response.getEntity()));
JSONArray arrays = tokens.getJSONArray("tokens"); JSONArray arrays = tokens.getJSONArray("tokens");
List<String> list=new ArrayList<>(); List<String> list = new ArrayList<>();
for (int i = 0; i < arrays.size(); i++) { for (int i = 0; i < arrays.size(); i++) {
JSONObject obj = JSON.parseObject(arrays.getString(i)); JSONObject obj = JSON.parseObject(arrays.getString(i));
list.add(obj.getString("token")); list.add(obj.getString("token"));

Write Preview
Loading…
Cancel
Save