c-zhouhf1/zhgh_hmc
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

commit

Browse Source
This commit is contained in:
Paidax
2024-12-09 15:57:22 +08:00
parent baa291b8d5
commit 131c9b1082
9 changed files with 235 additions and 126 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/main/java/io/v/nutz/zhgh/data/constant/SourceData.java
+5 -7
View File
@@ -42,7 +42,7 @@ public interface SourceData {
* @param jsonResult 接口返回数据
*/
private static void checkSuccess(NutMap jsonResult) {
Assert.isTrue(jsonResult.getBoolean("success"), jsonResult.getString("data"));
// Assert.isTrue(jsonResult.getBoolean("success"), jsonResult.getString("data"));
}
/* *//**
@@ -157,10 +157,8 @@ public interface SourceData {
while (hasRecords) {
String body = HttpUtil.createPost(DATA_URL).body(JSON.toJSONString(new NutMap().addv("address", "jzgjbxx").addv("params", new NutMap())
.addv("token", "3318a5a9-c2f2-4c8b-a8ea-e99dd68c165c").addv("pageIndex", page).addv("pageSize", 1000))).execute().body();
// String body = HttpUtil.get(DATA_URL, new NutMap().addv("address", "jzgjbxx").addv("params", new NutMap())
// .addv("token", "3318a5a9-c2f2-4c8b-a8ea-e99dd68c165c").addv("pageIndex", page).addv("pageSize", 1000));
NutMap map = Json.fromJson(NutMap.class, body);
checkSuccess(map);
// checkSuccess(map);
List<NutMap> data = map.getAsList("data", NutMap.class);
for (NutMap row : data) {
Map entity = new HashMap(1000);
@@ -179,7 +177,7 @@ public interface SourceData {
users.add(BeanUtil.mapToBean(entity, UserSource.class, true));
}
int totalCount = map.getInt("totalCount");
maxPage = (int) Math.ceil((double) totalCount / page);
maxPage = (int) Math.ceil((double) totalCount / 1000);
page++;
if (page > maxPage) {
hasRecords = false;
@@ -212,11 +210,11 @@ public interface SourceData {
List<Sys_unit> units = new ArrayList<>();
int page = 1;
String body = HttpUtil.createPost(DATA_URL).body(JSON.toJSONString(new NutMap().addv("address", "rsxt_dwjbsj").addv("params", new NutMap())
.addv("token", "3318a5a9-c2f2-4c8b-a8ea-e99dd68c165c").addv("pageIndex", page).addv("pageSize", 1000))).execute().body();
.addv("token", "3318a5a9-c2f2-4c8b-a8ea-e99dd68c165c").addv("pageIndex", page).addv("pageSize", 100))).execute().body();
// String body = HttpUtil.get(DATA_URL, new NutMap().addv("address", "rsxt_dwjbsj").addv("params", new NutMap())
// .addv("token", "3318a5a9-c2f2-4c8b-a8ea-e99dd68c165c").addv("pageIndex", page).addv("pageSize", 1000));
NutMap map = Json.fromJson(NutMap.class, body);
checkSuccess(map);
// checkSuccess(map);
List<NutMap> data = map.getAsList("data", NutMap.class);
for (NutMap row : data) {
src/main/java/io/v/nutz/zhgh/data/service/impl/SourceUserServiceImpl.java
+45 -51
View File
@@ -1,6 +1,5 @@
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.StrUtil;
@@ -14,7 +13,6 @@ import io.v.nutz.zhgh.data.model.UserHistory;
import io.v.nutz.zhgh.data.model.UserSource;
import io.v.nutz.zhgh.data.service.HistoryUserService;
import io.v.nutz.zhgh.data.service.SourceUserService;
import io.v.nutz.base.lang.Enum;
import io.v.nutz.zhgh.member.UserMode;
import io.v.nutz.base.service.AsyncService;
import io.v.nutz.base.service.impl.ViServiceImpl;
@@ -30,7 +28,6 @@ import org.nutz.dao.Cnd;
import org.nutz.dao.Dao;
import org.nutz.dao.FieldFilter;
import org.nutz.dao.Sqls;
import org.nutz.dao.entity.Record;
import org.nutz.dao.sql.Sql;
import org.nutz.dao.util.Daos;
import org.nutz.ioc.aop.Aop;
@@ -47,9 +44,10 @@ import org.springframework.util.StringUtils;
import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicReference;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
import java.util.stream.Collectors;
/**
* @author 1V
*/
@@ -83,12 +81,6 @@ public class SourceUserServiceImpl extends ViServiceImpl<UserSource> implements
public void pullSourceData(Boolean isIncrement) {
Date now = new Date();
List<UserSource> users = SourceData.users();
/* List<UserSource> userSourceList = users.stream().filter(v -> !v.getUserState().equals("离校")).collect(Collectors.toList());
//离退休劳模,要更新到系统
List<String> userPartUpLoginNameList = userPatUpService.query(Cnd.where("groupId", "=", 2)).stream().map(UserPartUp::getLoginname).collect(Collectors.toList());
//离退休超过两年
List<UserSource> retireUser = userSourceList.stream().filter(v -> v.getRetireDate() != null && DateUtil.betweenYear(v.getRetireDate(), DateUtil.date(), false) > 2L && !userPartUpLoginNameList.contains(v.getLoginname())).collect(Collectors.toList());
users = userSourceList.stream().filter(v -> !retireUser.contains(v)).collect(Collectors.toList());*/
asyncService.exe2(users, (user) -> {
user.setPullTime(now);
UserMode.initUser(user);
@@ -176,7 +168,7 @@ public class SourceUserServiceImpl extends ViServiceImpl<UserSource> implements
}
}
List<UserChangeType> changeTypes = getChangeTypes(source, user);
List<UserChangeType> changeTypes = getChangeTypes(source, user, sourceType);
if (changeTypes.contains(UserChangeType.NEW)) {
user = UserMode.initUser(u);
needInitUserList.add(user);
@@ -220,6 +212,7 @@ public class SourceUserServiceImpl extends ViServiceImpl<UserSource> implements
// sysRoleService.clearCache();
} catch (Exception e) {
e.printStackTrace();
throw new RuntimeException(e.getMessage());
}
}
@@ -237,35 +230,38 @@ public class SourceUserServiceImpl extends ViServiceImpl<UserSource> implements
dao().clear(UserSource.class, Cnd.where("DATE(pullTime)", "<", formatAgoWeekDate));
}
private List<UserChangeType> getChangeTypes(Sys_user source, Sys_user user) {
private List<UserChangeType> getChangeTypes(Sys_user source, Sys_user user, String sourceType) {
List<UserChangeType> userChangeTypes = new ArrayList<>();
if (user == null) {
userChangeTypes.add(UserChangeType.NEW);
return userChangeTypes;
}
if (!StringUtils.isEmpty(source.getUnitid()) && !source.getUnitid().equals(user.getUnitid())) {
userChangeTypes.add(UserChangeType.UNIT_CHANGE);
}
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);
} else {
if ("add".equals(sourceType)) {
return userChangeTypes;
}
if (!StringUtils.isEmpty(source.getUnitid()) && !source.getUnitid().equals(user.getUnitid())) {
userChangeTypes.add(UserChangeType.UNIT_CHANGE);
}
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);
}
}
}
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);
String sourceValue = userSourceMap.getString(column, "");
String value = userMap.getString(column, "");
if (!sourceValue.equals(value) && StrUtil.isNotBlank(sourceValue)) {
userChangeTypes.add(UserChangeType.BASIC_CHANGE);
break;
}
}
}
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);
String sourceValue = userSourceMap.getString(column, "");
String value = userMap.getString(column, "");
if (!sourceValue.equals(value) && StrUtil.isNotBlank(sourceValue)) {
userChangeTypes.add(UserChangeType.BASIC_CHANGE);
break;
}
}
return userChangeTypes;
}
@@ -304,53 +300,51 @@ public class SourceUserServiceImpl extends ViServiceImpl<UserSource> implements
* @param filterColumnDao 部分更新传递（部分更新存在过滤字段）
*/
private void massUpdatesAsync(List<Sys_user> needDoUpdateList, Boolean isPart, AtomicReference<Dao> filterColumnDao) {
// 根据服务器性能和任务特性动态确定线程池大小;通常，线程池大小为CPU核心数的2倍
// 创建一个CPU核心数的2倍的线程池
int numberOfThreads = Runtime.getRuntime().availableProcessors() * 2;
ExecutorService executorService = Executors.newFixedThreadPool(numberOfThreads);
ExecutorCompletionService<Void> ecs = new ExecutorCompletionService<>(executorService);
// 每批处理的数据量
int batchSize = 100;
int batchSize = 200;
// 共享锁
Lock lock = new ReentrantLock();
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) {
lock.lock();
try {
daoToUse.updateIgnoreNull(batch);
} finally {
lock.unlock();
}
return null; // CompletionService需要一个返回值，这里返回null
return null;
});
}
// 等待所有任务完成
for (int i = 0; i < needDoUpdateList.size(); i += batchSize) {
try {
Future<Void> future = ecs.take();
future.get(); // 等待任务完成
future.get();
} catch (InterruptedException | ExecutionException e) {
Thread.currentThread().interrupt(); // 恢复中断状态
throw new RuntimeException(e);
Thread.currentThread().interrupt();
throw new RuntimeException("Error during batch update", e);
}
}
} catch (Exception e) {
e.printStackTrace();
log.error("Error in massUpdatesAsync", e);
throw new RuntimeException("Error in massUpdatesAsync", e);
} finally {
// 关闭线程池
executorService.shutdown();
try {
// 设置合理的超时时间，防止无限期阻塞
if (!executorService.awaitTermination(10, TimeUnit.SECONDS)) {
executorService.shutdownNow(); // 如果超时，强制关闭
executorService.shutdownNow();
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt(); // 恢复中断状态
Thread.currentThread().interrupt();
}
}
}