ES定时同步数据

2023-11-27 19:04:11 +08:00
parent fc684156d4
commit ced29459af
2 changed files with 247 additions and 0 deletions
@@ -477,5 +477,19 @@ public class JeroElasticsearchTemplate {
        return json;
    }

+    /**
+     * 按条件删除数据
+     * <p>
+     * 请求地址：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;
+    }
+
 }

@@ -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<String> listType = getList(resourceType);
+        Map<String,Integer> 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<String, Integer> mapHash) {
+        List<LawsNewsFeed> 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<String> listIds = listLawsNewsFeed.stream().map(LawsNewsFeed::getId).collect(Collectors.toList());
+                addES(listIds, LawsHomeSearchCommon.LAWS_NEWS_FEED);
+            }
+        }
+    }
+
+    private void checkLawsEnterpriseStandard(String resourceType, Map<String, Integer> mapHash) {
+        List<LawsEnterpriseStandard> 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<String> listIds = listLawsEnterpriseStandard.stream().map(LawsEnterpriseStandard::getId).collect(Collectors.toList());
+                addES(listIds, LawsHomeSearchCommon.LAWS_ENTERPRISE_STANDARD);
+            }
+        }
+    }
+
+    private void checkLawsOverseasStandard(String resourceType, Map<String, Integer> mapHash) {
+        List<LawsOverseasStandard> 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<String> listIds = listLawsOverseasStandard.stream().map(LawsOverseasStandard::getId).collect(Collectors.toList());
+                addES(listIds, LawsHomeSearchCommon.LAWS_OVERSEAS_STANDARD);
+            }
+        }
+    }
+
+    private void checkLawsDomesticStandard(String resourceType, Map<String, Integer> mapHash) {
+        List<LawsDomesticStandard> 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<String> listIds = listLawsDomesticStandard.stream().map(LawsDomesticStandard::getId).collect(Collectors.toList());
+                addES(listIds, LawsHomeSearchCommon.LAWS_DOMESTIC_STANDARD);
+            }
+        }
+    }
+
+    @NotNull
+    private List<String> getList(String resourceType) {
+        List<String> 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<String> 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<String> 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<String,Integer> getTypeCount(){
+        Map<String,Integer> mapHash = new HashMap<>();
+        JSONObject jsonObject = new JSONObject();
+        Map<String, Object> map = new HashMap<>();
+        Map<String, Object> map1 = new HashMap<>();
+        Map<String, Object> 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<Map<String, Object>> list1 = (List<Map<String, Object>>) ((Map) ((Map) search.get("aggregations")).get("resource_type_aggs")).get("buckets");
+        if (!CollectionUtils.isEmpty(list1)) {
+            for (Map<String, Object> 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<String, Object> map1 = new HashMap<>();
+        Map<String, Object> 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);
+    }
+}