diff --git a/adc-da-search/src/main/java/com/adc/da/search/service/ElasticsearchService.java b/adc-da-search/src/main/java/com/adc/da/search/service/ElasticsearchService.java index 07c74b31..baedb9e1 100644 --- a/adc-da-search/src/main/java/com/adc/da/search/service/ElasticsearchService.java +++ b/adc-da-search/src/main/java/com/adc/da/search/service/ElasticsearchService.java @@ -28,6 +28,11 @@ public interface ElasticsearchService { */ boolean deleteIndex(String index); + /** + * Elasticsearch 批量删除数据 + */ + void deleteBatchId(List list,String index); + /** * Elasticsearch 判断索引是否存在定义 * @param index 要判断的索引名称 diff --git a/adc-da-search/src/main/java/com/adc/da/search/service/impl/ElasticsearchServiceImpl.java b/adc-da-search/src/main/java/com/adc/da/search/service/impl/ElasticsearchServiceImpl.java index 8879dfb8..0b86db69 100644 --- a/adc-da-search/src/main/java/com/adc/da/search/service/impl/ElasticsearchServiceImpl.java +++ b/adc-da-search/src/main/java/com/adc/da/search/service/impl/ElasticsearchServiceImpl.java @@ -8,8 +8,10 @@ import org.elasticsearch.action.admin.indices.create.CreateIndexRequestBuilder; import org.elasticsearch.action.admin.indices.create.CreateIndexResponse; import org.elasticsearch.action.admin.indices.exists.indices.IndicesExistsRequest; import org.elasticsearch.action.admin.indices.exists.indices.IndicesExistsResponse; +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.GetRequestBuilder; import org.elasticsearch.action.get.GetResponse; @@ -178,6 +180,19 @@ public class ElasticsearchServiceImpl implements ElasticsearchService { return response.getResult().toString(); } + @Override + public void deleteBatchId( List lists,String index) { + logger.info("批量删除索引: "+index +", 正在删除:" + lists.size() + "条数据"); + BulkRequest request = new BulkRequest(); + for (String smsStr : lists) { + DeleteRequest deleteRequest=new DeleteRequest(index,smsStr); + request.add(deleteRequest); + } + client.bulk(request); + logger.info("批量删除索引: "+index +", 删除成功共:" + lists.size() + "条数据"); + } + + /** * Elasticsearch 删除某个index所有的文档接口实现 * @param index 要插入的索引 diff --git a/adc-da-slrs/src/main/java/com/adc/da/search/ResetSearchCenterController.java b/adc-da-slrs/src/main/java/com/adc/da/search/ResetSearchCenterController.java index 76c49725..73b8b274 100644 --- a/adc-da-slrs/src/main/java/com/adc/da/search/ResetSearchCenterController.java +++ b/adc-da-slrs/src/main/java/com/adc/da/search/ResetSearchCenterController.java @@ -22,7 +22,7 @@ import java.util.Map; @RestController @RequestMapping("/${restPath}/search/resetSearchCenter") -@Api(description = "搜索中心重置索引及数据+实施预警触发") +@Api(tags = "搜索中心重置索引及数据") public class ResetSearchCenterController extends BaseController> { private static final Logger logger = LoggerFactory.getLogger(ResetSearchCenterController.class); @@ -41,7 +41,7 @@ public class ResetSearchCenterController extends BaseController rowsStand = iSarStandardsInfoService.getSarStandardsInfoPage(page); List insList = new ArrayList<>(); List updList = new ArrayList<>(); + List delList = new ArrayList<>(); + List standIdList = rowsStand.stream().map(SarStandardsInfo::getId).collect(Collectors.toList()); if(!esIdList.isEmpty()){ // 差集 (list1 - list2) 新增 @@ -134,12 +136,17 @@ public class ResetSearchCenterService { .collect(Collectors.toList()); updList.addAll(intersection); + // 差集 (list2 - list1) 删除 + List delDataList = esIdList.stream().filter(num -> !standIdList.contains(num)) + .collect(Collectors.toList()); + delList.addAll(delDataList); + }else { // 新增 insList.addAll(rowsStand); } - if(!updList.isEmpty() && ("ALL".equals(searchCenter.getExecType()) || "UPD".equals(searchCenter.getExecType()))){ + if(!updList.isEmpty() && ("ALL".equals(searchCenter.getExecType().toUpperCase()) || "UPD".equals(searchCenter.getExecType().toUpperCase()))){ for(SarStandardsInfo sarStandardsInfoEO : updList){ countUpdateSuccess++; if (sarStandardsInfoEO.getAttrInfoMap() != null) { @@ -149,7 +156,7 @@ public class ResetSearchCenterService { } } - if(!insList.isEmpty() && ("ALL".equals(searchCenter.getExecType()) || "ADD".equals(searchCenter.getExecType()))){ + if(!insList.isEmpty() && ("ALL".equals(searchCenter.getExecType().toUpperCase()) || "ADD".equals(searchCenter.getExecType().toUpperCase()))){ for(SarStandardsInfo sarStandardsInfoEO : insList){ countAddSuccess++; if (sarStandardsInfoEO.getAttrInfoMap() != null) { @@ -158,8 +165,13 @@ public class ResetSearchCenterService { createStandMQService.sendStandMQ(sarStandardsInfoEO,"add"); } } - logger.info("重置国内外标准:新增-"+ countAddSuccess + "条 更新-"+countUpdateSuccess+"条(国内外标准共:"+rowsStand.size()+"条)"); - return Result.success("重置国内外标准:新增-"+ countAddSuccess + "条 更新-"+countUpdateSuccess+"条(国内外标准共:"+rowsStand.size()+"条)"); + + if(!delList.isEmpty() && ("ALL".equals(searchCenter.getExecType().toUpperCase()) || "DEL".equals(searchCenter.getExecType().toUpperCase()))){ + elasticsearchService.deleteBatchId(delList,"stand"); + elasticsearchService.deleteBatchId(delList,"fulltextserch"); + } + logger.info("重置国内外标准:新增-"+ countAddSuccess + "条 更新-"+countUpdateSuccess+"条(国内外标准共:"+rowsStand.size()+"条)删除-共"+delList.size()+"条数据"); + return Result.success("重置国内外标准:新增-"+ countAddSuccess + "条 更新-"+countUpdateSuccess+"条(国内外标准共:"+rowsStand.size()+"条)删除-共"+delList.size()+"条数据"); } @Async @@ -181,6 +193,9 @@ public class ResetSearchCenterService { List rowsStand = iSarLawsStandInfoService.getSarStandardsInfoPage(sarLawsInfoEOPage); List insList = new ArrayList<>(); List updList = new ArrayList<>(); + List delList = new ArrayList<>(); + List standIdList = rowsStand.stream().map(SarLawsStandInfo::getId).collect(Collectors.toList()); + if(!esIdList.isEmpty()){ // 差集 (list1 - list2) 新增 @@ -197,12 +212,17 @@ public class ResetSearchCenterService { .collect(Collectors.toList()); updList.addAll(intersection); + // 差集 (list2 - list1) 删除 + List delDataList = esIdList.stream().filter(num -> !standIdList.contains(num)) + .collect(Collectors.toList()); + delList.addAll(delDataList); + }else { // 新增 insList.addAll(rowsStand); } - if(!updList.isEmpty() && ("ALL".equals(searchCenter.getExecType()) || "UPD".equals(searchCenter.getExecType()))){ + if(!updList.isEmpty() && ("ALL".equals(searchCenter.getExecType().toUpperCase()) || "UPD".equals(searchCenter.getExecType().toUpperCase()))){ for(SarLawsStandInfo sarLawsInfoEO : updList){ countUpdateSuccess++; if (sarLawsInfoEO.getAttrInfoMap() != null) { @@ -212,7 +232,7 @@ public class ResetSearchCenterService { } } - if(!insList.isEmpty() && ("ALL".equals(searchCenter.getExecType()) || "ADD".equals(searchCenter.getExecType()))){ + if(!insList.isEmpty() && ("ALL".equals(searchCenter.getExecType().toUpperCase()) || "ADD".equals(searchCenter.getExecType().toUpperCase()))){ for(SarLawsStandInfo sarLawsInfoEO : insList){ countAddSuccess++; if (sarLawsInfoEO.getAttrInfoMap() != null) { @@ -221,8 +241,14 @@ public class ResetSearchCenterService { createStandMQService.sendLawsMQ(sarLawsInfoEO,"add"); } } - logger.info("重置国内外政策:新增-"+ countAddSuccess + "条 更新-"+countUpdateSuccess+"条(国内外政策共:"+rowsStand.size()+"条)"); - return Result.success("重置国内外政策:新增-"+ countAddSuccess + "条 更新-"+countUpdateSuccess+"条(国内外政策共:"+rowsStand.size()+"条)"); + + if(!delList.isEmpty() && ("ALL".equals(searchCenter.getExecType().toUpperCase()) || "DEL".equals(searchCenter.getExecType().toUpperCase()))){ + elasticsearchService.deleteBatchId(delList,"laws"); + elasticsearchService.deleteBatchId(delList,"fulltextserch"); + } + + logger.info("重置国内外政策:新增-"+ countAddSuccess + "条 更新-"+countUpdateSuccess+"条(国内外政策共:"+rowsStand.size()+"条)删除-共"+delList.size()+"条数据"); + return Result.success("重置国内外政策:新增-"+ countAddSuccess + "条 更新-"+countUpdateSuccess+"条(国内外政策共:"+rowsStand.size()+"条)删除-共"+delList.size()+"条数据"); } private void idListFunc(SarLawsStandInfoPage sarLawsInfoEOPage, SearchCenter searchCenter) { @@ -256,6 +282,9 @@ public class ResetSearchCenterService { List rowsStand = iSarBussionessStandService.getSarBussionStandPage(sarBussionessStandEOPage); List insList = new ArrayList<>(); List updList = new ArrayList<>(); + List delList = new ArrayList<>(); + List standIdList = rowsStand.stream().map(SarBussionessStand::getId).collect(Collectors.toList()); + if(!esIdList.isEmpty()){ // 差集 (list1 - list2) 新增 @@ -272,12 +301,17 @@ public class ResetSearchCenterService { .collect(Collectors.toList()); updList.addAll(intersection); + // 差集 (list2 - list1) 删除 + List delDataList = esIdList.stream().filter(num -> !standIdList.contains(num)) + .collect(Collectors.toList()); + delList.addAll(delDataList); + }else { // 新增 insList.addAll(rowsStand); } - if(!updList.isEmpty() && ("ALL".equals(searchCenter.getExecType()) || "UPD".equals(searchCenter.getExecType()))){ + if(!updList.isEmpty() && ("ALL".equals(searchCenter.getExecType().toUpperCase()) || "UPD".equals(searchCenter.getExecType().toUpperCase()))){ for(SarBussionessStand sarBussionessStandEO : updList){ countUpdateSuccess++; if (sarBussionessStandEO.getAttrInfoMap() != null) { @@ -287,7 +321,7 @@ public class ResetSearchCenterService { } } - if(!insList.isEmpty() && ("ALL".equals(searchCenter.getExecType()) || "ADD".equals(searchCenter.getExecType()))){ + if(!insList.isEmpty() && ("ALL".equals(searchCenter.getExecType().toUpperCase()) || "ADD".equals(searchCenter.getExecType().toUpperCase()))){ for(SarBussionessStand sarBussionessStandEO : insList){ countAddSuccess++; if (sarBussionessStandEO.getAttrInfoMap() != null) { @@ -296,8 +330,13 @@ public class ResetSearchCenterService { createStandMQService.sendBussStandMQ(sarBussionessStandEO,"add"); } } - logger.info("重置企标:新增-"+ countAddSuccess + "条 更新-"+countUpdateSuccess+"条(企标共:"+rowsStand.size()+"条)"); - return Result.success("重置企标:新增-"+ countAddSuccess + "条 更新-"+countUpdateSuccess+"条(企标共:"+rowsStand.size()+"条)"); + + if(!delList.isEmpty() && ("ALL".equals(searchCenter.getExecType().toUpperCase()) || "DEL".equals(searchCenter.getExecType().toUpperCase()))){ + elasticsearchService.deleteBatchId(delList,"bussstand"); + elasticsearchService.deleteBatchId(delList,"fulltextserch"); + } + logger.info("重置企标:新增-"+ countAddSuccess + "条 更新-"+countUpdateSuccess+"条(企标共:"+rowsStand.size()+"条)删除-共"+delList.size()+"条数据"); + return Result.success("重置企标:新增-"+ countAddSuccess + "条 更新-"+countUpdateSuccess+"条(企标共:"+rowsStand.size()+"条)删除-共"+delList.size()+"条数据"); } /**