Compare commits
No commits in common. "6c17cf12a7dfe32d219bd014993a991ed279956b" and "564b35a68d6f0edb13e9db54e30e131918ab58ae" have entirely different histories.
6c17cf12a7
...
564b35a68d
|
|
@ -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.substring(0, 100), e.getMessage());
|
||||
} else {
|
||||
log.error("executeJson ql[{}] error, msg [{}]", gql, e.getMessage());
|
||||
}
|
||||
log.error("executeJson ql[{}] error, msg [{}]", gql, e.getMessage());
|
||||
throw new GraphExecuteException("execute gql error, gql: " + gql, e);
|
||||
} finally {
|
||||
session.release();
|
||||
|
|
|
|||
|
|
@ -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) {
|
||||
if (isComplete) {
|
||||
log.info("关系补全全部结束...");
|
||||
|
||||
for (TagEnum tagEnum : TagEnum.values()) {
|
||||
|
||||
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();
|
||||
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);
|
||||
String tag = tagEnum.getTag();
|
||||
|
||||
List<NebulaMultiMatchJsonResult> idJsonResults =
|
||||
graphCommonService.executeJson(NebulaUtil.getTagIdsWithLimit(spaceName, tag, skip, this.limitNum),
|
||||
NebulaMultiMatchJsonResult.class);
|
||||
Integer num = FileCacheManager.get(tag + "_num", Integer.class);
|
||||
// Object num = redisClient.get(tag + "_num");
|
||||
|
||||
if (CollectionUtils.isEmpty(idJsonResults)) {
|
||||
break;
|
||||
}
|
||||
Integer skip = (num == null ? 0 : num);
|
||||
|
||||
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);
|
||||
if (tagNumsMap.get(tag) < skip) {
|
||||
FileCacheManager.increment(ALL_NUM);
|
||||
// redisClient.increment(ALL_NUM);
|
||||
continue;
|
||||
}
|
||||
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(
|
||||
NebulaUtil.insertEdges(spaceName, links, reverseRelation),
|
||||
CommonVo.class,
|
||||
"",
|
||||
sessionPool,
|
||||
true
|
||||
);
|
||||
} 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);
|
||||
}
|
||||
graphCommonService.executeJsonWithoutLogs(
|
||||
NebulaUtil.insertEdges(spaceName, links, reverseRelation),
|
||||
CommonVo.class,
|
||||
"",
|
||||
sessionPool,
|
||||
true
|
||||
);
|
||||
log.info("reverseRelation {} 存在边关系，已关系补全", reverseRelation);
|
||||
}
|
||||
|
||||
public ShowStatsVo showStats() {
|
||||
|
|
|
|||
Loading…
Reference in New Issue
Block a user