diff --git a/laws-base/laws-base-core/src/main/java/com/jero/common/es/JeroElasticsearchTemplate.java b/laws-base/laws-base-core/src/main/java/com/jero/common/es/JeroElasticsearchTemplate.java index d1a9a8d3..b0f3d35c 100644 --- a/laws-base/laws-base-core/src/main/java/com/jero/common/es/JeroElasticsearchTemplate.java +++ b/laws-base/laws-base-core/src/main/java/com/jero/common/es/JeroElasticsearchTemplate.java @@ -477,5 +477,19 @@ public class JeroElasticsearchTemplate { return json; } + /** + * 按条件删除数据 + *

+ * 请求地址：POST http://{baseUrl}/{indexName}/{typeName}/_delete_by_query + */ + public JSONObject delete(String indexName, String typeName, JSONObject queryObject) { + String url = this.getBaseUrl(indexName, typeName).append("/_delete_by_query").toString(); + + log.info("url:" + url + " ,delete: " + queryObject.toJSONString()); + JSONObject res = RestUtil.post(url, queryObject); + log.info("url:" + url + " ,return res: \n" + res.toJSONString()); + return res; + } + } diff --git a/laws-modules/src/main/java/com/jero/modules/laws/home/job/HomeSearchCheckEsPlanJob.java b/laws-modules/src/main/java/com/jero/modules/laws/home/job/HomeSearchCheckEsPlanJob.java new file mode 100644 index 00000000..e2b96a9d --- /dev/null +++ b/laws-modules/src/main/java/com/jero/modules/laws/home/job/HomeSearchCheckEsPlanJob.java @@ -0,0 +1,233 @@ +package com.jero.modules.laws.home.job; + +import com.alibaba.fastjson.JSONArray; +import com.alibaba.fastjson.JSONObject; +import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper; +import com.jero.common.es.JeroElasticsearchTemplate; +import com.jero.modules.laws.home.common.LawsHomeSearchCommon; +import com.jero.modules.laws.home.enums.HomeSearchEnum; +import com.jero.modules.laws.home.service.ILawsHomeSearchService; +import com.jero.modules.laws.standard.entity.LawsDomesticStandard; +import com.jero.modules.laws.standard.entity.LawsEnterpriseStandard; +import com.jero.modules.laws.standard.entity.LawsNewsFeed; +import com.jero.modules.laws.standard.entity.LawsOverseasStandard; +import com.jero.modules.laws.standard.service.ILawsDomesticStandardService; +import com.jero.modules.laws.standard.service.ILawsEnterpriseStandardService; +import com.jero.modules.laws.standard.service.ILawsNewsFeedService; +import com.jero.modules.laws.standard.service.ILawsOverseasStandardService; +import io.minio.messages.DeleteRequest; +import lombok.extern.slf4j.Slf4j; +import org.apache.commons.collections4.CollectionUtils; +import org.apache.commons.lang3.StringUtils; +import org.jetbrains.annotations.NotNull; +import org.quartz.Job; +import org.quartz.JobDataMap; +import org.quartz.JobExecutionContext; +import org.quartz.JobExecutionException; + +import javax.annotation.Resource; +import java.io.IOException; +import java.util.*; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ForkJoinPool; +import java.util.stream.Collectors; + +/** + * 每天定时比对ES数据条数与业务数据条数是否一致，不一致重新更新 + * @author lqt + * @version 1.0 + * @date 2023/11/27 10:30 + */ +@Slf4j +public class HomeSearchCheckEsPlanJob implements Job { + @Resource + private ILawsDomesticStandardService lawsDomesticStandardService; + @Resource + private ILawsOverseasStandardService lawsOverseasStandardService; + @Resource + private ILawsEnterpriseStandardService lawsEnterpriseStandardService; + @Resource + private ILawsNewsFeedService lawsNewsFeedService; + @Resource + private JeroElasticsearchTemplate jeroElasticsearchTemplate; + @Resource + private ILawsHomeSearchService lawsHomeSearchService; + + + @Override + public void execute(JobExecutionContext context) throws JobExecutionException { + // 获取外部传来的参数 + JobDataMap map = context.getJobDetail().getJobDataMap(); + String resourceType = map.getString("parameter"); + List listType = getList(resourceType); + Map mapHash = getTypeCount(); + listType.parallelStream().forEach(p->{ + switch (p){ + // 比对国内标准 + case LawsHomeSearchCommon.LAWS_DOMESTIC_STANDARD: + checkLawsDomesticStandard(resourceType, mapHash); + break; + // 比对国外标准 + case LawsHomeSearchCommon.LAWS_OVERSEAS_STANDARD: + checkLawsOverseasStandard(resourceType, mapHash); + break; + // 比对企业标准 + case LawsHomeSearchCommon.LAWS_ENTERPRISE_STANDARD: + checkLawsEnterpriseStandard(resourceType, mapHash); + break; + // 比对动态消息 + case LawsHomeSearchCommon.LAWS_NEWS_FEED: + checkLawsNewsFeed(resourceType, mapHash); + break; + default:break; + } + }); + } + + private void checkLawsNewsFeed(String resourceType, Map mapHash) { + List listLawsNewsFeed = lawsNewsFeedService.list(); + int count4 = mapHash.getOrDefault(LawsHomeSearchCommon.LAWS_NEWS_FEED, 0); + int m4 = 0; + if(!CollectionUtils.isEmpty(listLawsNewsFeed)){ + m4 = listLawsNewsFeed.size(); + } + if(count4 != m4){ + // 清理数据重新更新 + delete(resourceType); + if(!CollectionUtils.isEmpty(listLawsNewsFeed)){ + List listIds = listLawsNewsFeed.stream().map(LawsNewsFeed::getId).collect(Collectors.toList()); + addES(listIds, LawsHomeSearchCommon.LAWS_NEWS_FEED); + } + } + } + + private void checkLawsEnterpriseStandard(String resourceType, Map mapHash) { + List listLawsEnterpriseStandard = lawsEnterpriseStandardService.list(); + int count3 = mapHash.getOrDefault(LawsHomeSearchCommon.LAWS_ENTERPRISE_STANDARD, 0); + int m3 = 0; + if(!CollectionUtils.isEmpty(listLawsEnterpriseStandard)){ + m3 = listLawsEnterpriseStandard.size(); + } + if(count3 != m3){ + // 清理数据重新更新 + delete(resourceType); + if(!CollectionUtils.isEmpty(listLawsEnterpriseStandard)){ + List listIds = listLawsEnterpriseStandard.stream().map(LawsEnterpriseStandard::getId).collect(Collectors.toList()); + addES(listIds, LawsHomeSearchCommon.LAWS_ENTERPRISE_STANDARD); + } + } + } + + private void checkLawsOverseasStandard(String resourceType, Map mapHash) { + List listLawsOverseasStandard = lawsOverseasStandardService.list(); + int count2 = mapHash.getOrDefault(LawsHomeSearchCommon.LAWS_OVERSEAS_STANDARD, 0); + int m2 = 0; + if(!CollectionUtils.isEmpty(listLawsOverseasStandard)){ + m2 = listLawsOverseasStandard.size(); + } + if(count2 != m2){ + // 清理数据重新更新 + delete(resourceType); + if(!CollectionUtils.isEmpty(listLawsOverseasStandard)){ + List listIds = listLawsOverseasStandard.stream().map(LawsOverseasStandard::getId).collect(Collectors.toList()); + addES(listIds, LawsHomeSearchCommon.LAWS_OVERSEAS_STANDARD); + } + } + } + + private void checkLawsDomesticStandard(String resourceType, Map mapHash) { + List listLawsDomesticStandard = lawsDomesticStandardService.list(); + // 比对数据 + int count1 = mapHash.getOrDefault(LawsHomeSearchCommon.LAWS_DOMESTIC_STANDARD, 0); + int m1 = 0; + if(!CollectionUtils.isEmpty(listLawsDomesticStandard)){ + m1 = listLawsDomesticStandard.size(); + } + if(count1 != m1){ + // 清理数据重新更新 + delete(resourceType); + if(!CollectionUtils.isEmpty(listLawsDomesticStandard)){ + List listIds = listLawsDomesticStandard.stream().map(LawsDomesticStandard::getId).collect(Collectors.toList()); + addES(listIds, LawsHomeSearchCommon.LAWS_DOMESTIC_STANDARD); + } + } + } + + @NotNull + private List getList(String resourceType) { + List listType = new ArrayList<>(); + if(StringUtils.isBlank(resourceType)){ + listType.add(LawsHomeSearchCommon.LAWS_DOMESTIC_STANDARD); + listType.add(LawsHomeSearchCommon.LAWS_OVERSEAS_STANDARD); + listType.add(LawsHomeSearchCommon.LAWS_ENTERPRISE_STANDARD); + listType.add(LawsHomeSearchCommon.LAWS_NEWS_FEED); + }else{ + listType.add(resourceType); + } + return listType; + } + + private void addES(List listIds, String lawsNewsFeed) { + int startIndex = 0; // 从第0个下标开始 + int batchCount = 1000; + while (startIndex < listIds.size()) { + int endIndex = 0; + if (listIds.size() - batchCount < startIndex) { + endIndex = listIds.size(); + } else { + endIndex = startIndex + batchCount; + } + List ids = listIds.subList(startIndex, endIndex); + // 异步生成记录 + ForkJoinPool pool = new ForkJoinPool(); + // 异步调用 + CompletableFuture.runAsync(() -> { + try { + lawsHomeSearchService.updateEsData(lawsNewsFeed, String.join(",", ids), LawsHomeSearchCommon.LAWS_ES_ADD); + } catch (Exception e) { + log.error("ES同步异常：", e.getMessage()); + } + }, pool); + startIndex = startIndex + batchCount; // 下一批 + } + } + + private Map getTypeCount(){ + Map mapHash = new HashMap<>(); + JSONObject jsonObject = new JSONObject(); + Map map = new HashMap<>(); + Map map1 = new HashMap<>(); + Map map2 = new HashMap<>(); + map1.put("field","resource_type.keyword"); + map2.put("terms",map1); + map.put("resource_type_aggs",map2); + jsonObject.put("aggs",map); + jsonObject.put("size",0); + //2. 数据查询 + JSONObject search = jeroElasticsearchTemplate.search(HomeSearchEnum.INDEX_NAME_DOCUMENT_ZQ.getValue(), HomeSearchEnum.TYPE_NAME_DOCUMENT_ZQ.getValue(), jsonObject); + List> list1 = (List>) ((Map) ((Map) search.get("aggregations")).get("resource_type_aggs")).get("buckets"); + if (!CollectionUtils.isEmpty(list1)) { + for (Map stringObjectMap : list1) { + if (Objects.isNull(stringObjectMap.get("key"))) { + continue; + } + mapHash.put(stringObjectMap.get("key").toString(),Objects.isNull(stringObjectMap.get("doc_count")) ? 0 : Integer.parseInt(stringObjectMap.get("doc_count").toString())); + } + } + return mapHash; + } + + private void delete(String resourceType){ + JSONObject obj = new JSONObject(); + JSONArray queryMapJsonDict = new JSONArray(); + Map map1 = new HashMap<>(); + Map map2 = new HashMap<>(); + map2.put("resource_type",resourceType); + map1.put("match_phrase",map2); + queryMapJsonDict.add(map1); + JSONObject jsonObject = jeroElasticsearchTemplate.buildBoolQuery(queryMapJsonDict, null, null); + obj.put("query",jsonObject); + //2. 数据查询 + jeroElasticsearchTemplate.delete(HomeSearchEnum.INDEX_NAME_DOCUMENT_ZQ.getValue(), HomeSearchEnum.TYPE_NAME_DOCUMENT_ZQ.getValue(), obj); + } +}