relation-completion/src/main/java/com/idata/tools/relationcompletion/completion/GraphCommonService.java

@@ -37,6 +37,7 @@ public class GraphCommonService {
     Session session;
 
     public <T> List<T> executeJson(String gql, Class<T> voClass) {
+
         try {
             log.info("执行 executeJson 方法，gql={}", gql);
             final String data = session.executeJson(gql);
@@ -59,6 +60,7 @@ public class GraphCommonService {
             log.error("executeJson ql[{}] error, msg [{}]", gql, e.getMessage());
             throw e;
         }
+
     }
 
 
@@ -79,6 +81,7 @@ public class GraphCommonService {
         Session session = null;
         try {
             session = sessionPool.borrow(false);
+//
             final String data = session.executeJson(gql);
             if (isLog) {
                 log.info("执行 executeJson 方法，gql={}", gql.substring(0, 100));
@@ -96,11 +99,7 @@ public class GraphCommonService {
             JSONArray results = JSONUtil.parseArray(jsonObject.get("results"));
             return JSONUtil.toList(results, voClass);
         } catch (IOErrorException e) {
-            if (isLog) {
+            log.error("executeJson ql[{}] error, msg [{}]", gql, e.getMessage());
-                log.error("executeJson ql[{}] error, msg [{}]", gql.substring(0, 100), e.getMessage());
-            } else {
-                log.error("executeJson ql[{}] error, msg [{}]", gql, e.getMessage());
-            }
             throw new GraphExecuteException("execute gql error, gql: " + gql, e);
         } finally {
             session.release();

relation-completion/src/main/java/com/idata/tools/relationcompletion/completion/RelationCompletion.java

@@ -42,7 +42,6 @@ public class RelationCompletion {
     private final String spaceName;
     private final int limitNum;
     private static final String ALL_NUM = "allGetNum";
-    private static final Logger log = LoggerFactory.getLogger(RelationCompletion.class);
 
     public RelationCompletion(GraphCommonService graphCommonService,
                               NebulaGraphProperties nebulaGraphProperties) {
@@ -53,6 +52,8 @@ public class RelationCompletion {
                 10, 10, 5L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<>());
     }
 
+    private static Logger log = LoggerFactory.getLogger(RelationCompletion.class);
+
     /**
      * 关系反转入口函数
      */
@@ -64,85 +65,90 @@ public class RelationCompletion {
         Map<String, Integer> tagNumsMap = getTagAndNums(stats);
         // 新起一个nebula session池
         SessionPool sessionPool = graphCommonService.newPoolConfig();
-        boolean isComplete = false;
+
-        while (true) {
+        for (TagEnum tagEnum : TagEnum.values()) {
-            if (isComplete) {
+
-                log.info("关系补全全部结束...");
+            Object allGetNum = FileCacheManager.get(ALL_NUM, Integer.class);
+            if (allGetNum != null && (Integer) allGetNum >= 5) {
                 break;
             }
-            for (TagEnum tagEnum : TagEnum.values()) {
-                Object allGetNum = FileCacheManager.get(ALL_NUM, Integer.class);
-                if (allGetNum != null && (Integer) allGetNum >= 5) {
-                    isComplete = true;
-                    break;
-                }
 
-                String tag = tagEnum.getTag();
+            String tag = tagEnum.getTag();
-                Integer num = FileCacheManager.get(tag + "_num", Integer.class);
-                //  Object num = redisClient.get(tag + "_num");
-                Integer skip = (num == null ? 0 : num);
-                if (tagNumsMap.get(tag) < skip) {
-                    FileCacheManager.increment(ALL_NUM);
-                    //  redisClient.increment(ALL_NUM);
-                    continue;
-                }
-                log.info("开启{}类型的点为起始节点的第{}条数据到第{}条数据反向关系补全", tag, skip, skip + this.limitNum);
 
-                List<NebulaMultiMatchJsonResult> idJsonResults =
+            Integer num = FileCacheManager.get(tag + "_num", Integer.class);
-                        graphCommonService.executeJson(NebulaUtil.getTagIdsWithLimit(spaceName, tag, skip, this.limitNum),
+            //  Object num = redisClient.get(tag + "_num");
-                                NebulaMultiMatchJsonResult.class);
 
-                if (CollectionUtils.isEmpty(idJsonResults)) {
+            Integer skip = (num == null ? 0 : num);
-                    break;
-                }
 
-                List<String> idLists = NebulaMultiMatchJsonResult.getIdLists(idJsonResults.get(0));
+            if (tagNumsMap.get(tag) < skip) {
-                if (CollectionUtils.isEmpty(idLists)) {
+                FileCacheManager.increment(ALL_NUM);
-                    log.info("idLists为空...");
+                //  redisClient.increment(ALL_NUM);
-                    break;
+                continue;
-                } else {
-                    log.info("查询到的idLists大小: {}", idLists.size());
-                }
-                List<CompletableFuture<?>> futures = new ArrayList<>();
-                // 遍历边
-                EdgeEnum[] edgeEnums = EdgeEnum.values();
-                for (EdgeEnum edgeEnum : edgeEnums) {
-                    Integer finalSkip = skip;
-                    CompletableFuture<Void> setVertexAndEdgeTotalsMap = CompletableFuture.supplyAsync(() -> {
-                        try {
-                            String srcTag = edgeEnum.getSrcTag();
-                            String edgeType = edgeEnum.getEdgeType();
-                            String reverseRelation = edgeEnum.getReverseRelation();
-                            Instant now = Instant.now();
-
-                            List<NebulaVertexJsonResult> vertexJsonResults = graphCommonService.executeJsonWithoutLogs(
-                                    NebulaUtil.getDstVertices(spaceName, srcTag, idLists, edgeType),
-                                    NebulaVertexJsonResult.class,
-                                    "",
-                                    sessionPool);
-
-                            SearchSubgraphVo searchSubgraphVo = NebulaUtil.getSearchSubgraphVo(vertexJsonResults.get(0));
-                            // 构造反向关系并插入nebula
-                            convertEdgeDirection(searchSubgraphVo, sessionPool, reverseRelation);
-                            log.info("{}类型的点; {}类型的边的反向关系补全, 反向关系为: {} 耗时为:{}", tag, edgeEnum.getEdgeType()
-                                    , edgeEnum.getReverseRelation(), Duration.between(now, Instant.now()));
-                        } catch (Exception e) {
-                            log.info("反向关系补全 edgeEnum:{} 失败", edgeEnum, e);
-                        }
-                        log.info("{}类型的点为起始节点{}到第{}条数据反向关系补全完成", tag, finalSkip, finalSkip + this.limitNum);
-                        return null;
-                    }, threadPoolExecutor);
-                    futures.add(setVertexAndEdgeTotalsMap);
-                }
-                CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).join();
-
-                // 更新特定标签的数量
-                skip += this.limitNum;
-                FileCacheManager.set(tag + "_num", skip);
             }
+            log.info("开启{}类型的点为起始节点的第{}条数据到第{}条数据反向关系补全", tag, skip, skip + this.limitNum);
+
+            List<NebulaMultiMatchJsonResult> idJsonResults =
+                    graphCommonService.executeJson(NebulaUtil.getTagIdsWithLimit(spaceName, tag, skip, this.limitNum),
+                            NebulaMultiMatchJsonResult.class);
+
+            if (CollectionUtils.isEmpty(idJsonResults)) {
+                break;
+            }
+
+            List<String> idLists = NebulaMultiMatchJsonResult.getIdLists(idJsonResults.get(0));
+            if (CollectionUtils.isEmpty(idLists)) {
+                log.info("idLists为空...");
+                break;
+            } else {
+                log.info("查询到的idLists大小: {}", idLists.size());
+            }
+            List<CompletableFuture<?>> futures = new ArrayList<>();
+            // 遍历边
+            EdgeEnum[] edgeEnums = EdgeEnum.values();
+            for (EdgeEnum edgeEnum : edgeEnums) {
+                Integer finalSkip = skip;
+                CompletableFuture<Void> setVertexAndEdgeTotalsMap = CompletableFuture.supplyAsync(() -> {
+                    try {
+
+
+                        String srcTag = edgeEnum.getSrcTag();
+                        String edgeType = edgeEnum.getEdgeType();
+                        String reverseRelation = edgeEnum.getReverseRelation();
+
+                        Instant now = Instant.now();
+
+
+                        List<NebulaVertexJsonResult> vertexJsonResults = graphCommonService.executeJsonWithoutLogs(
+                                NebulaUtil.getDstVertices(spaceName, srcTag, idLists, edgeType),
+                                NebulaVertexJsonResult.class,
+                                "",
+                                sessionPool);
+
+                        SearchSubgraphVo searchSubgraphVo = NebulaUtil.getSearchSubgraphVo(vertexJsonResults.get(0));
+                        // 构造反向关系并插入nebula
+                        convertEdgeDirection(searchSubgraphVo, sessionPool, reverseRelation);
+                        log.info("{}类型的点; {}类型的边的反向关系补全, 反向关系为: {} 耗时为:{}", tag, edgeEnum.getEdgeType()
+                                , edgeEnum.getReverseRelation(), Duration.between(now, Instant.now()));
+
+                    } catch (Exception e) {
+
+                        log.info("反向关系补全 edgeEnum:{} 失败", edgeEnum, e);
+                    }
+
+                    log.info("{}类型的点为起始节点{}到第{}条数据反向关系补全完成", tag, finalSkip, finalSkip + this.limitNum);
+                    return null;
+                }, threadPoolExecutor);
+                futures.add(setVertexAndEdgeTotalsMap);
+            }
+            CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).join();
+
+            // 更新特定标签的数量
+            skip += this.limitNum;
+            FileCacheManager.set(tag + "_num", skip);
         }
     }
 
+
     private Map<String, Integer> getTagAndNums(ShowStatsVo stats) {
         Map<String, Integer> map = new HashMap<>();
         for (ShowStatsVo.Data statsNum : stats.getNums()) {
@@ -164,27 +170,16 @@ public class RelationCompletion {
             log.info("reverseRelation {} 仅有节点数据，无需关系补全", reverseRelation);
             return;
         }
-        // 插入边
-        insertEdge(links, sessionPool, reverseRelation);
-        log.info("reverseRelation {} 存在边关系，已关系补全", reverseRelation);
-    }
 
-    private void insertEdge(List< SearchSubgraphVo.EdgeVo> links, SessionPool sessionPool, String reverseRelation) {
         // TODO 处理请求参数超出服务端设置的大小时的情况，服务端默认4MB，超出会报异常
-        try {
+        graphCommonService.executeJsonWithoutLogs(
-            graphCommonService.executeJsonWithoutLogs(
+                NebulaUtil.insertEdges(spaceName, links, reverseRelation),
-                    NebulaUtil.insertEdges(spaceName, links, reverseRelation),
+                CommonVo.class,
-                    CommonVo.class,
+                "",
-                    "",
+                sessionPool,
-                    sessionPool,
+                true
-                    true
+        );
-            );
+        log.info("reverseRelation {} 存在边关系，已关系补全", reverseRelation);
-        } catch (Exception e) {
-            log.warn("请求参数大小超出服务端限制，减少参数重新发送...");
-            int mid = links.size() / 2;
-            insertEdge(links.subList(0, mid), sessionPool, reverseRelation);
-            insertEdge(links.subList(mid, links.size()), sessionPool, reverseRelation);
-        }
     }
 
     public ShowStatsVo showStats() {