This commit is contained in:
zhengziru
2024-07-25 15:44:18 +08:00
parent 3a03b4eb9f
commit 53f1a70ed0
13 changed files with 333 additions and 434 deletions
@@ -33,19 +33,6 @@ public class UserUpdateJobTask implements Job {
@Override
@Aop(TransAop.READ_COMMITTED)
public void execute(JobExecutionContext context){
try {
sourceUserService.pullSourceData(true);
Date pullTime = (Date) dao.func2(UserSource.class, "max", "pullTime");
sourceUserService.updateSysUser(DateUtil.formatDateTime(pullTime), "add",null,null,false);
DateTime agoWeekDate = DateUtil.offsetWeek(new Date(), -1);
String formatAgoWeekDate = DateUtil.format(agoWeekDate, "yyyy-MM-dd");
sourceUserService.clear(Cnd.where("DATE(pullTime)","<",formatAgoWeekDate));
} catch (Exception e) {
e.printStackTrace();
}
sourceUserService.dailyScheduledUpdates();
}
}
@@ -53,15 +53,15 @@ public class Globals {
//项目短名称
public static String AppShrotName = "";
//项目域名 (设置一个默认的 为优先级考虑)
public static String AppDomain = "https://zhghydd.cpu.edu.cn";
public static String AppDomain = "http://101.126.19.57";
//cas地址
public static String CasAddress = "https://id.cpu.edu.cn/sso";
public static String CasAddress = "";
//文件访问域名
public static String AppFileDomain = "";
//文件上传路径
public static String AppUploadBase = "";
//学校代码
public static String schoolCode = "cpu";
public static String schoolCode = "hrbnu";
//系统自定义参数
public static NutMap MyConfig = NutMap.NEW();
//自定义路由
@@ -69,7 +69,7 @@ public class Globals {
//微信map
public static NutMap WxMap = NutMap.NEW();
//高校名称
public static String schoolName = "中国药科大学";
public static String schoolName = "哈尔滨师范大学";
public static String BrandActivityName = "";
@@ -235,12 +235,4 @@ public class UpdateUserController {
}
@At
@ViReturn
@RequiresPermissions("data.user.update")
@SLog(type = "数据管理",tag = "更新系统用户数据",msg = "退休(发放福利)人员重置单位信息",param = true,result = true)
public Object retireWelfareChange(){
sourceUserService.retireWelfareChange();
return null;
}
}
@@ -27,7 +27,7 @@ public interface SourceUserService extends ViService<UserSource> {
/**
* 退休(发放福利)人员变更
* 每日定时更新
*/
void retireWelfareChange();
void dailyScheduledUpdates();
}
@@ -1,6 +1,7 @@
package io.v.nutz.zhgh.data.service.impl;
import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.core.date.DateTime;
import cn.hutool.core.date.DateUtil;
import cn.hutool.core.util.ObjectUtil;
import cn.hutool.core.util.StrUtil;
@@ -49,9 +50,7 @@ import org.springframework.beans.BeanUtils;
import org.springframework.util.StringUtils;
import java.util.*;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicReference;
import java.util.stream.Collectors;
@@ -89,38 +88,6 @@ public class SourceUserServiceImpl extends ViServiceImpl<UserSource> implements
@Override
@Aop(TransAop.READ_COMMITTED)
public void pullSourceData(Boolean isIncrement) {
/* Date now = new Date();
List<UserSource> users = SourceData.users();
//找所有单位
Map<String, String> unitIdMap = sysUnitService.query().stream().collect(Collectors.toMap(Sys_unit::getId, Sys_unit::getName));
//获取人员的三级单位使用的是二级单位id的教职工,分组,将这些单位id+000后新增
List<UserSource> userSourceList = users.stream().filter(v -> v.getThreeUnitId().length() == 4).collect(Collectors.toList());
List<String> needUpdateUnitList = userSourceList.stream().map(Sys_user::getThreeUnitId).distinct().collect(Collectors.toList());
List<Sys_unit> insertUnitList = new ArrayList<>();
needUpdateUnitList.forEach(v->{
String unitId = v + "000";
if (!unitIdMap.containsKey(unitId)){
Sys_unit unit = new Sys_unit();
unit.setId(unitId);
unit.setUnitcode(unitId);
unit.setName(unitIdMap.get(v));
unit.setParentId(v);
unit.setUnitlevel(3);
insertUnitList.add(unit);
}
});
userSourceList.forEach(v->v.setThreeUnitId(v.getThreeUnitId()+"000"));
asyncService.exe2(users, (user) -> {
user.setPullTime(now);
UserMode.initUser(user);
});
dao().insert(users);
sysUserService.updateIgnoreNull(userSourceList);
sysUnitService.insert(insertUnitList);*/
Date now = new Date();
List<UserSource> users = SourceData.users();
asyncService.exe2(users, (user) -> {
@@ -135,10 +102,9 @@ public class SourceUserServiceImpl extends ViServiceImpl<UserSource> implements
public void updateSysUser(String pullTime, String sourceType, String[] columnNames, String partGroupId, boolean isInvert) {
try {
List<Sys_user> users = sysUserService.query(Cnd.where("loginname", "is not", null));
Map<String, Sys_user> userMap = users.stream().collect(Collectors.toMap(sys_user -> sys_user.getLoginname(), sys_user -> sys_user));
Map<String, Sys_user> userMap = users.stream().collect(Collectors.toMap(sys_user -> sys_user.getLoginname().toLowerCase(), sys_user -> sys_user));
List<UserSource> sources = new ArrayList<>();
Date now = new Date();
List<UserHistory> histories = new ArrayList<>();
@@ -160,76 +126,33 @@ public class SourceUserServiceImpl extends ViServiceImpl<UserSource> implements
AtomicReference<Dao> filterColumnDao = new AtomicReference<>(dao());
//比对出退休人员后,添加到更新表中
List<UserPartUp> addUserPartUpList = new ArrayList<>();
//退休人员
List<Sys_user> retireUserList = new ArrayList<>();
//退休发放福利的人
List<String> userIds = userPatUpService.dao().query(UserPartUp.class, Cnd.where("groupId", "=", 66)).stream().map(UserPartUp::getUserId).collect(Collectors.toList());
//全部更新时,要修改的用户表
List<Sys_user> needDoUpdateList = new ArrayList<>();
//新增的用户
List<Sys_user> needInitUserList = new ArrayList<>();
//添加角色的用户
List<Sys_user_role> userRoles = new ArrayList<>();
List<Sys_user> unitChangeUserList = new ArrayList<>();
// 创建一个固定大小的线程池 可以根据服务器性能调整线程池大小
int numberOfThreads = Runtime.getRuntime().availableProcessors() * 2;
List<CompletableFuture<Void>> futures = new ArrayList<>();
// 每批处理的数据量,可以根据实际情况调整
int batchSize = 200;
// 创建一个固定大小的线程池
int numberOfThreads = 20; // 可以根据服务器性能调整线程池大小
ExecutorService executorService = Executors.newFixedThreadPool(numberOfThreads);
int batchSize = 200; // 每批处理的数据量,可以根据实际情况调整
for (int i = 0; i < sources.size(); i += batchSize) {
final int end = Math.min(i + batchSize, sources.size());
List<UserSource> batch = sources.subList(i, end);
// 提交任务到线程池
Map<String, String> finalUserPartMap = userPartMap;
executorService.submit(()->{
CompletableFuture<Void> future = CompletableFuture.runAsync(() -> {
for (UserSource source : batch) {
//不更新会员和福利会员字段
source.setMember(null);
source.setWelfareMember(null);
Sys_user user = userMap.get(source.getLoginname());
UserChangeType changeType = getChangeType(source, user);
Sys_user user = userMap.get(source.getLoginname().toLowerCase());
Sys_user u = new Sys_user();
BeanUtils.copyProperties(source, u);
//如果变更类型是退休,加到更新表中
if (Lang.isNotEmpty(user) && userIds.contains(user.getId())) {
continue;
}
if (UserChangeType.RETIREMENT.equals(changeType)) {
UserPartUp part = new UserPartUp();
//同时将在职状态改为退休发放福利
user.setUserState("退休(发放福利)");
retireUserList.add(user);
UserHistory history = new UserHistory();
BeanUtils.copyProperties(user, history);
history.setChangeTime(now);
history.setChangeType(changeType.getCode());
histories.add(history);
if (!userIds.contains(user.getId())) {
part.setUserId(user.getId());
part.setLoginname(user.getLoginname());
part.setGroupId(66);
part.setCreator(ShiroUtil.getUserId());
part.setGroupName("退休(发放福利)");
addUserPartUpList.add(part);
}
continue;
} else if (UserChangeType.UNIT_CHANGE.equals(changeType)) {
unitChangeUserList.add(u);
}
switch (sourceType) {
case "all" -> {
//复制属性到一个新的user对象
@@ -255,7 +178,6 @@ public class SourceUserServiceImpl extends ViServiceImpl<UserSource> implements
}
List<UserChangeType> changeTypes = getChangeTypes(source, user);
if (changeTypes.contains(UserChangeType.NEW)) {
user = UserMode.initUser(u);
needInitUserList.add(user);
@@ -264,16 +186,15 @@ public class SourceUserServiceImpl extends ViServiceImpl<UserSource> implements
userRole.setRoleId(Roles.PUBLIC);
userRoles.add(userRole);
histories.add(createUserHistory(user, changeTypes));
} else if (Lang.isNotEmpty(changeTypes)) {
} else if (changeTypes.size() > 0) {
histories.add(createUserHistory(user, changeTypes));
}
}
});
}, Executors.newFixedThreadPool(numberOfThreads));
futures.add(future);
}
executorService.shutdown();
executorService.awaitTermination(Long.MAX_VALUE, TimeUnit.NANOSECONDS);
CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).join();
//如果有新用户,增加到用户表同时增加角色
if (Lang.isNotEmpty(needInitUserList)) {
@@ -283,119 +204,40 @@ public class SourceUserServiceImpl extends ViServiceImpl<UserSource> implements
sysUserRoleService.insert(userRoles);
}
// Collection<String[]> updateColumns = SourceData.USER_FIELD_RELATION.values();
for (List<Sys_user> userList : CollectionUtil.split(needDoUpdateList, 50)) {
Dao finalFilterColumnDao = filterColumnDao.get();
new Thread(() -> {
log.info("{}在更新数据", Thread.currentThread().getName());
if ("part".equals(sourceType)) {
finalFilterColumnDao.updateIgnoreNull(userList);
} else {
dao().updateIgnoreNull(userList);
}
}).start();
if ("part".equals(sourceType)) {
massUpdatesAsync(needDoUpdateList, true, filterColumnDao);
} else {
massUpdatesAsync(needDoUpdateList, false, null);
}
if (Lang.isNotEmpty(addUserPartUpList)) userPatUpService.dao().insert(addUserPartUpList);
if (Lang.isNotEmpty(retireUserList)) userPatUpService.dao().updateIgnoreNull(retireUserList);
//历史记录表插入数据
historyUserService.dao().fastInsert(histories);
//人员状态及在职状态更新
userPatUpService.renewUserState();
//单位异动,清除三级单位,重新赋三级单位
resetThreeUnitId(unitChangeUserList);
//清除缓存
sysUserService.clearCache();
sysRoleService.clearCache();
// sysUserService.clearCache();
// sysRoleService.clearCache();
} catch (Exception e) {
e.printStackTrace();
}
}
/**
* 退休(发放福利)人员变更
*/
@Override
@Aop(TransAop.READ_COMMITTED)
public void retireWelfareChange() {
//获取变更表的退休发放福利人员
List<UserPartUp> userPartUpList = userPatUpService.query(Cnd.where("groupId", "=", 66));
List<String> loginNameList = userPartUpList.stream().map(UserPartUp::getLoginname).collect(Collectors.toList());
//在userSource表找出 部分更新表的 退休人员记录
List<UserSource> sourceList = userPatUpService.dao().query(UserSource.class, Cnd.where("loginname", "in", loginNameList).and("userState", "=", UserChangeType.RETIREMENT.getDescription()));
public void dailyScheduledUpdates() {
pullSourceData(true);
Date pullTime = (Date) dao().func2(UserSource.class, "max", "pullTime");
updateSysUser(DateUtil.formatDateTime(pullTime), "add",null,null,false);
//根据工号去重 (更新数据源)
List<UserSource> sourceDistinctList = sourceList.stream().collect(Collectors.collectingAndThen(Collectors.toCollection(() -> new TreeSet<>(Comparator.comparing(UserSource::getLoginname))), ArrayList::new));
//在用户表中找出 部分更新表的 退休人员记录 (系统数据)
Map<String, Sys_user> userMap = userPatUpService.dao().query(Sys_user.class, Cnd.where("loginname", "in", loginNameList)).stream().collect(Collectors.toMap(sys_user -> sys_user.getLoginname(), sys_user -> sys_user));
DateTime agoWeekDate = DateUtil.offsetWeek(new Date(), -1);
String formatAgoWeekDate = DateUtil.format(agoWeekDate, "yyyy-MM-dd");
List<Sys_user> needUpdateUserList = new ArrayList<>();
List<UserHistory> histories = new ArrayList<>();
sourceDistinctList.forEach(source -> {
Sys_user userSystem = userMap.get(source.getLoginname());
UserChangeType changeType = getChangeType(source, userSystem);
Sys_user user = new Sys_user();
BeanUtils.copyProperties(source, user);
user.setId(userMap.get(source.getLoginname()).getId());
needUpdateUserList.add(user);
//添加人员变更记录
if (changeType != null) {
UserHistory history = new UserHistory();
BeanUtils.copyProperties(userSystem, history);
history.setChangeTime(new Date());
history.setChangeType(changeType.getCode());
histories.add(history);
}
//添加会员变更记录
MemberChangeRecord record = new MemberChangeRecord();
record.setUserId(user.getId());
record.setType(UserChangeType.RETIREMENT.getCode());
record.setOriginUnitId(userSystem.getUnitid());
record.setCurrentUnitId(user.getUnitid());
record.setUpdateWelfareMember(true);
record.setApplyTime(new Date());
record.setApplyOrigin(MemberChangeOrigin.RETIREMENT_CHANGE.getCode());
record.setRemark("【退休(发放福利)】人员变更为【退休】,不享受福利,单位归档离退休工作处");
record.setStateId(70);
sysUserService.dao().insert(record);
MemberChangeHandle handle = Enum.instance(MemberChangeHandle.class, UserChangeType.RETIREMENT.getCode());
handle.change(record);
});
if (Lang.isNotEmpty(needUpdateUserList)) {
sysUserService.updateIgnoreNull(needUpdateUserList);
}
historyUserService.dao().insert(histories);
userPatUpService.clear(Cnd.where("groupId", "=", 66));
sysUserService.clearCache();
dao().clear(UserSource.class, Cnd.where("DATE(pullTime)", "<", formatAgoWeekDate));
}
private UserChangeType getChangeType(Sys_user source, Sys_user user) {
//新入职
if (user == null) {
return UserChangeType.NEW;
}
//在职状态变化
if (StrUtil.isNotBlank(user.getUserState()) && StrUtil.isNotEmpty(source.getUserState()) && !source.getUserState().equals(user.getUserState())) {
return Enum.instance(UserChangeType.class, source.getUserState());
}
//单位变化
if (StrUtil.isNotBlank(source.getUnitid()) && StrUtil.isNotBlank(source.getUnitid()) && !source.getUnitid().equals(user.getUnitid())) {
return UserChangeType.UNIT_CHANGE;
}
return null;
}
private List<UserChangeType> getChangeTypes(Sys_user source, Sys_user user) {
List<UserChangeType> userChangeTypes = new ArrayList<>();
if (user == null) {
@@ -405,7 +247,7 @@ public class SourceUserServiceImpl extends ViServiceImpl<UserSource> implements
if (!StringUtils.isEmpty(source.getUnitid()) && !source.getUnitid().equals(user.getUnitid())) {
userChangeTypes.add(UserChangeType.UNIT_CHANGE);
}
if (StrUtil.isNotBlank(user.getUserState()) && StrUtil.isNotEmpty(source.getUserState()) && !source.getUserState().equals(user.getUserState())) {
if (!StringUtils.isEmpty(user.getUserState()) && StrUtil.isNotEmpty(source.getUserState()) && !source.getUserState().equals(user.getUserState())) {
for (UserChangeType value : UserChangeType.values()) {
if (value.getDescription().equals(source.getUserState())) {
userChangeTypes.add(value);
@@ -413,7 +255,7 @@ public class SourceUserServiceImpl extends ViServiceImpl<UserSource> implements
}
}
List<String> columns = List.of("nation", "birthday", "personType", "political", "schoolTime", "jobTitle", "education", "academicDegree", "position");
List<String> columns = List.of("nation", "birthday", "personType", "political", "schoolTime", "jobTitle", "education", "academicDegree", "position", "leaveSchoolDate", "retireDate");
for (String column : columns) {
NutMap userSourceMap = Lang.obj2nutmap(source);
NutMap userMap = Lang.obj2nutmap(user);
@@ -428,25 +270,6 @@ public class SourceUserServiceImpl extends ViServiceImpl<UserSource> implements
return userChangeTypes;
}
/**
* 生成一条用户历史记录
*
* @param user
* @param changeType
* @return
*/
private UserHistory createUserHistory(Sys_user user, UserChangeType changeType) {
//除仅更新新增人员外的方式都需要往历史记录表插数据
UserHistory history = new UserHistory();
history.setId(R.UU32());
BeanUtils.copyProperties(user, history);
history.setChangeTime(new Date());
history.setChangeType(changeType.getCode());
return history;
}
/**
* 生成一条用户历史记录
*
@@ -475,23 +298,62 @@ public class SourceUserServiceImpl extends ViServiceImpl<UserSource> implements
}
private void resetThreeUnitId(List<Sys_user> unitChangeUserList){
List<Sys_unit> unitList = dao().query(Sys_unit.class, Cnd.where("unitlevel","=",2));
/**
* 数据大批量更新
* @param needDoUpdateList 需要更新的数据
* @param isPart 是否部分更新
* @param filterColumnDao 部分更新传递(部分更新存在过滤字段)
*/
private void massUpdatesAsync(List<Sys_user> needDoUpdateList, Boolean isPart, AtomicReference<Dao> filterColumnDao) {
// 根据服务器性能和任务特性动态确定线程池大小;通常,线程池大小为CPU核心数的2倍
int numberOfThreads = Runtime.getRuntime().availableProcessors() * 2;
ExecutorService executorService = Executors.newFixedThreadPool(numberOfThreads);
ExecutorCompletionService<Void> ecs = new ExecutorCompletionService<>(executorService);
unitChangeUserList.forEach(v->{
Sys_unit sysUnit = unitList.stream().filter(item -> v.getUnitid().equals(item.getParentId()))
.collect(Collectors.toList()).stream().findFirst().orElse(null);
if (Lang.isNotEmpty(sysUnit)) {
v.setThreeUnitId(sysUnit.getId());
} else {
v.setThreeUnitId(null);
// 每批处理的数据量
int batchSize = 100;
try {
// 将sources分成若干批
for (int i = 0; i < needDoUpdateList.size(); i += batchSize) {
final int end = Math.min(i + batchSize, needDoUpdateList.size());
List<Sys_user> batch = needDoUpdateList.subList(i, end);
// 提交任务给线程池执行
ecs.submit(() -> {
Dao daoToUse = isPart ? filterColumnDao.get() : dao();
// 执行数据库更新操作
synchronized (daoToUse) {
daoToUse.updateIgnoreNull(batch);
}
return null; // CompletionService需要一个返回值,这里返回null
});
}
});
if (unitChangeUserList.size() > 200) {
sysUserService.largeQuantitiesInsert(unitChangeUserList,2,false);
} else {
dao().updateIgnoreNull(unitChangeUserList);
// 等待所有任务完成
for (int i = 0; i < needDoUpdateList.size(); i += batchSize) {
try {
Future<Void> future = ecs.take();
future.get(); // 等待任务完成
} catch (InterruptedException | ExecutionException e) {
Thread.currentThread().interrupt(); // 恢复中断状态
throw new RuntimeException(e);
}
}
} catch (Exception e) {
e.printStackTrace();
} finally {
// 关闭线程池
executorService.shutdown();
try {
// 设置合理的超时时间,防止无限期阻塞
if (!executorService.awaitTermination(10, TimeUnit.SECONDS)) {
executorService.shutdownNow(); // 如果超时,强制关闭
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt(); // 恢复中断状态
}
}
}
}
@@ -110,7 +110,7 @@ public class MemberCheckUnionController {
cnd.and(new Static("his.changeType is null"));
}
}
cnd.desc("his.changeTime");
sql.setCondition(cnd);
sql.setParam("success", MemberChangeApplyState.SUCCESS);
sql.setVar("failStates", Joiner.on(",").join(List.of(MemberChangeApplyState.UNION_FAIL, MemberChangeApplyState.SCHOOL_UNION_FAIL)));