clay/fateverse
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix : 流程解析存在多线程并发问题

Browse Source
This commit is contained in:
clay
2024-04-11 14:52:41 +08:00
parent dec31fb6ed
commit 695d21793e
1 changed files with 55 additions and 57 deletions
Download Patch File Download Diff File Expand all files Collapse all files

112
workflow/src/main/java/cn/fateverse/workflow/service/ProcessNodeService.java
View File

@@ -37,17 +37,6 @@ public class ProcessNodeService {
@DubboReference @DubboReference
private DubboUserService userService; private DubboUserService userService;
/**
* 用户id缓存
*/
private final Map<String, List<Long>> cacheUserIdMap;
//角色用户缓存
private final Map<Long, List<UserVo>> cacheRoleCacheUserMap;
//部门用户缓存
private final Map<Long, UserInfo> cacheDeptCacheUserMap;
//用户缓存信息
private final Map<Long, UserInfo> cacheUserMap;
public ProcessNodeService() { public ProcessNodeService() {
@@ -57,10 +46,6 @@ public class ProcessNodeService {
TimeUnit.SECONDS, new LinkedBlockingDeque<>(128), TimeUnit.SECONDS, new LinkedBlockingDeque<>(128),
new ThreadFactoryBuilder().setNameFormat("process_%d").build(), new ThreadFactoryBuilder().setNameFormat("process_%d").build(),
new ThreadPoolExecutor.CallerRunsPolicy()); new ThreadPoolExecutor.CallerRunsPolicy());
cacheUserIdMap = new ConcurrentHashMap<>();
cacheRoleCacheUserMap = new ConcurrentHashMap<>();
cacheDeptCacheUserMap = new ConcurrentHashMap<>();
cacheUserMap = new ConcurrentHashMap<>();
} }
@@ -92,29 +77,11 @@ public class ProcessNodeService {
} catch (InterruptedException e) { } catch (InterruptedException e) {
throw new RuntimeException(e); throw new RuntimeException(e);
} }
clearCache(); wrapper.clearCache();
System.out.println("用时 :" + (new Date().getTime() - startTime.getTime())); System.out.println("用时 :" + (new Date().getTime() - startTime.getTime()));
return wrapper.result; return wrapper.result;
} }
/**
* 初始化
*/
/**
* 清除缓存,当空间大于16时进行清除
*/
private void clearCache() {
if (cacheRoleCacheUserMap.size() > 16) {
cacheRoleCacheUserMap.clear();
}
if (cacheUserIdMap.size() > 16) {
cacheUserIdMap.clear();
}
if (cacheUserMap.size() > 16) {
cacheUserMap.clear();
}
}
/** /**
* 根据角色查询 * 根据角色查询
@@ -122,12 +89,12 @@ public class ProcessNodeService {
* @param roleIds 角色id列表 * @param roleIds 角色id列表
* @return 用户信息 * @return 用户信息
*/ */
private List<UserInfo> searchUserListByRoleIds(List<Long> roleIds) { private List<UserInfo> searchUserListByRoleIds(List<Long> roleIds, ProcessNodeWrapper wrapper) {
List<UserVo> result = new ArrayList<>(); List<UserVo> result = new ArrayList<>();
List<Long> roleIdsNew = new ArrayList<>(); List<Long> roleIdsNew = new ArrayList<>();
roleIds.forEach(roleId -> { roleIds.forEach(roleId -> {
if (cacheRoleCacheUserMap.containsKey(roleId)) { if (wrapper.cacheRoleCacheUserMap.containsKey(roleId)) {
result.addAll(cacheRoleCacheUserMap.get(roleId)); result.addAll(wrapper.cacheRoleCacheUserMap.get(roleId));
} else { } else {
roleIdsNew.add(roleId); roleIdsNew.add(roleId);
} }
@@ -135,7 +102,7 @@ public class ProcessNodeService {
if (!roleIdsNew.isEmpty()) { if (!roleIdsNew.isEmpty()) {
List<UserVo> userList = userService.searchUserListByRoleIds(roleIdsNew); List<UserVo> userList = userService.searchUserListByRoleIds(roleIdsNew);
Map<Long, List<UserVo>> userRole = userList.stream().collect(Collectors.groupingBy(UserVo::getRoleId)); Map<Long, List<UserVo>> userRole = userList.stream().collect(Collectors.groupingBy(UserVo::getRoleId));
cacheRoleCacheUserMap.putAll(userRole); wrapper.cacheRoleCacheUserMap.putAll(userRole);
result.addAll(userList); result.addAll(userList);
} }
return result.stream().map(userVo -> return result.stream().map(userVo ->
@@ -148,12 +115,12 @@ public class ProcessNodeService {
* @param deptIds 部门id列表 * @param deptIds 部门id列表
* @return 用户信息 * @return 用户信息
*/ */
private List<UserInfo> searchUserByDeptIds(List<Long> deptIds) { private List<UserInfo> searchUserByDeptIds(List<Long> deptIds, ProcessNodeWrapper wrapper) {
List<UserInfo> result = new ArrayList<>(); List<UserInfo> result = new ArrayList<>();
List<Long> deptIdsNew = new ArrayList<>(); List<Long> deptIdsNew = new ArrayList<>();
deptIds.forEach(deptId -> { deptIds.forEach(deptId -> {
if (cacheDeptCacheUserMap.containsKey(deptId)) { if (wrapper.cacheDeptCacheUserMap.containsKey(deptId)) {
result.add(cacheDeptCacheUserMap.get(deptId)); result.add(wrapper.cacheDeptCacheUserMap.get(deptId));
} else { } else {
deptIdsNew.add(deptId); deptIdsNew.add(deptId);
} }
@@ -163,7 +130,7 @@ public class ProcessNodeService {
userList.forEach(userVo -> { userList.forEach(userVo -> {
UserInfo value = UserInfo.toUserInfo(userVo, OperationStateEnums.UNACTIVATED); UserInfo value = UserInfo.toUserInfo(userVo, OperationStateEnums.UNACTIVATED);
result.add(value); result.add(value);
cacheDeptCacheUserMap.put(userVo.getLeaderDeptId(), value); wrapper.cacheDeptCacheUserMap.put(userVo.getLeaderDeptId(), value);
}); });
} }
return result; return result;
@@ -175,12 +142,12 @@ public class ProcessNodeService {
* @param userIds * @param userIds
* @return 用户信息列表 * @return 用户信息列表
*/ */
private List<UserInfo> searchUserListByUserIds(List<Long> userIds) { private List<UserInfo> searchUserListByUserIds(List<Long> userIds, ProcessNodeWrapper wrapper) {
List<UserInfo> result = new ArrayList<>(); List<UserInfo> result = new ArrayList<>();
List<Long> userIdsNew = new ArrayList<>(); List<Long> userIdsNew = new ArrayList<>();
userIds.forEach(userId -> { userIds.forEach(userId -> {
if (cacheUserMap.containsKey(userId)) { if (wrapper.cacheUserMap.containsKey(userId)) {
result.add(cacheUserMap.get(userId)); result.add(wrapper.cacheUserMap.get(userId));
} else { } else {
userIdsNew.add(userId); userIdsNew.add(userId);
} }
@@ -190,7 +157,7 @@ public class ProcessNodeService {
userList.forEach(userVo -> { userList.forEach(userVo -> {
UserInfo value = UserInfo.toUserInfo(userVo, OperationStateEnums.UNACTIVATED); UserInfo value = UserInfo.toUserInfo(userVo, OperationStateEnums.UNACTIVATED);
result.add(value); result.add(value);
cacheUserMap.put(userVo.getUserId(), value); wrapper.cacheUserMap.put(userVo.getUserId(), value);
}); });
} }
return result; return result;
@@ -304,7 +271,7 @@ public class ProcessNodeService {
List<Long> parentIds = Arrays.stream(dept.getAncestors().split(",")).map(Long::valueOf).collect(Collectors.toList()); List<Long> parentIds = Arrays.stream(dept.getAncestors().split(",")).map(Long::valueOf).collect(Collectors.toList());
switch (endCondition) { switch (endCondition) {
case "TOP"://直接到最上层 case "TOP"://直接到最上层
assignedUser = searchUserByDeptIds(parentIds.subList(1, parentIds.size())); assignedUser = searchUserByDeptIds(parentIds.subList(1, parentIds.size()), wrapper);
break; break;
case "LEAVE"://到指定级别 case "LEAVE"://到指定级别
int level = Integer.parseInt(leaderTop.get("level").toString()); int level = Integer.parseInt(leaderTop.get("level").toString());
@@ -315,7 +282,7 @@ public class ProcessNodeService {
} else { } else {
leaderDeptIds = parentIds.subList(parentIds.size() - level, parentIds.size()); leaderDeptIds = parentIds.subList(parentIds.size() - level, parentIds.size());
} }
assignedUser = searchUserByDeptIds(leaderDeptIds); assignedUser = searchUserByDeptIds(leaderDeptIds, wrapper);
break; break;
default: default:
throw new CustomException("当前选项还未实现!"); throw new CustomException("当前选项还未实现!");
@@ -331,7 +298,7 @@ public class ProcessNodeService {
int level = Integer.parseInt(props.getLeader().get("level").toString()); int level = Integer.parseInt(props.getLeader().get("level").toString());
Dept dept = wrapper.user.getDept(); Dept dept = wrapper.user.getDept();
if (1 == level) { if (1 == level) {
assignedUser = searchUserListByUserIds(Collections.singletonList(dept.getDeptId())); assignedUser = searchUserListByUserIds(Collections.singletonList(dept.getDeptId()), wrapper);
} else { } else {
List<Long> parentIds = Arrays.stream(dept.getAncestors().split(",")).map(Long::valueOf).collect(Collectors.toList()); List<Long> parentIds = Arrays.stream(dept.getAncestors().split(",")).map(Long::valueOf).collect(Collectors.toList());
Long leaderDeptId = null; Long leaderDeptId = null;
@@ -341,7 +308,7 @@ public class ProcessNodeService {
} else { } else {
leaderDeptId = parentIds.get(parentIds.size() - level); leaderDeptId = parentIds.get(parentIds.size() - level);
} }
assignedUser = searchUserByDeptIds(Collections.singletonList(leaderDeptId)); assignedUser = searchUserByDeptIds(Collections.singletonList(leaderDeptId), wrapper);
} }
props.setAssignedUser(assignedUser); props.setAssignedUser(assignedUser);
props.setAssignedType(AssigneeTypeEnums.ASSIGN_USER); props.setAssignedType(AssigneeTypeEnums.ASSIGN_USER);
@@ -374,7 +341,7 @@ public class ProcessNodeService {
List<UserInfo> assignedUser; List<UserInfo> assignedUser;
List<Long> roleIds = props.getRoleList().stream().map(RoleInfo::getRoleId).collect(Collectors.toList()); List<Long> roleIds = props.getRoleList().stream().map(RoleInfo::getRoleId).collect(Collectors.toList());
//rpc远程通过角色id获取到当前角色下的所有用户数据 //rpc远程通过角色id获取到当前角色下的所有用户数据
assignedUser = searchUserListByRoleIds(roleIds); assignedUser = searchUserListByRoleIds(roleIds, wrapper);
props.setAssignedUser(assignedUser); props.setAssignedUser(assignedUser);
props.setAssignedType(AssigneeTypeEnums.ASSIGN_USER); props.setAssignedType(AssigneeTypeEnums.ASSIGN_USER);
setResult(processNode, wrapper); setResult(processNode, wrapper);
@@ -383,12 +350,15 @@ public class ProcessNodeService {
private static class ProcessNodeWrapper { private static class ProcessNodeWrapper {
public ProcessNodeWrapper(Map<String, List<UserInfo>> userOperationInfoMap, JSONObject optionalUser, User user, Boolean isStart) {
this.userOperationInfoMap = userOperationInfoMap; //用户id缓存
this.optionalUser = optionalUser; private final Map<String, List<Long>> cacheUserIdMap;
this.user = user; //角色用户缓存
this.isStart = isStart; private final Map<Long, List<UserVo>> cacheRoleCacheUserMap;
} //部门用户缓存
private final Map<Long, UserInfo> cacheDeptCacheUserMap;
//用户缓存信息
private final Map<Long, UserInfo> cacheUserMap;
//操作历史缓存 //操作历史缓存
private final Map<String, List<UserInfo>> userOperationInfoMap; private final Map<String, List<UserInfo>> userOperationInfoMap;
@@ -397,5 +367,33 @@ public class ProcessNodeService {
private final List<ProcessNode> result = new ArrayList<>(); private final List<ProcessNode> result = new ArrayList<>();
private final User user; private final User user;
private final Boolean isStart; private final Boolean isStart;
public ProcessNodeWrapper(Map<String, List<UserInfo>> userOperationInfoMap, JSONObject optionalUser, User user, Boolean isStart) {
this.userOperationInfoMap = userOperationInfoMap;
this.optionalUser = optionalUser;
this.user = user;
this.isStart = isStart;
this.cacheUserIdMap = new ConcurrentHashMap<>();
this.cacheRoleCacheUserMap = new ConcurrentHashMap<>();
this.cacheDeptCacheUserMap = new ConcurrentHashMap<>();
this.cacheUserMap = new ConcurrentHashMap<>();
}
/**
* 清除缓存,当空间大于16时进行清除
*/
private void clearCache() {
if (cacheRoleCacheUserMap.size() > 16) {
cacheRoleCacheUserMap.clear();
}
if (cacheUserIdMap.size() > 16) {
cacheUserIdMap.clear();
}
if (cacheUserMap.size() > 16) {
cacheUserMap.clear();
}
}
} }
} }