kris/data-dump
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

【feat】 优化更新关联类型任务

Browse Source 
(cherry picked from commit cdbb419290)
This commit is contained in:
Kris 2025-07-30 10:28:00 +08:00
parent 54c07fca38
commit 968c1dd8d6
1 changed files with 20 additions and 52 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

72
src/main/java/com/celnet/datadump/service/impl/DataImportNewServiceImpl.java
View File

@ -1204,7 +1204,7 @@ public class DataImportNewServiceImpl implements DataImportNewService {
case OSS: case OSS:
// 上传到oss // 上传到oss
OssUtil.upload(inputStream, filePath); OssUtil.upload(inputStream, filePath);
default: case SERVER:
dumpToServer(headers, id, filePath, url, response, inputStream); dumpToServer(headers, id, filePath, url, response, inputStream);
} }
Map<String, Object> paramMap = Maps.newHashMap(); Map<String, Object> paramMap = Maps.newHashMap();
@ -1751,8 +1751,6 @@ public class DataImportNewServiceImpl implements DataImportNewService {
} }
// 等待当前所有线程执行完成 // 等待当前所有线程执行完成
salesforceExecutor.waitForFutures(futures.toArray(new Future<?>[]{})); salesforceExecutor.waitForFutures(futures.toArray(new Future<?>[]{}));
} catch (InterruptedException interruptedException){
return null;
} catch (Exception e) { } catch (Exception e) {
throw e; throw e;
} }
@ -1777,64 +1775,35 @@ public class DataImportNewServiceImpl implements DataImportNewService {
beginDateStr = DateUtil.format(beginDate, "yyyy-MM-dd HH:mm:ss"); beginDateStr = DateUtil.format(beginDate, "yyyy-MM-dd HH:mm:ss");
endDateStr = DateUtil.format(endDate, "yyyy-MM-dd HH:mm:ss"); endDateStr = DateUtil.format(endDate, "yyyy-MM-dd HH:mm:ss");
} }
String sql = "CreatedDate >= '" + beginDateStr + "' and CreatedDate < '" + endDateStr + "'";
String allFiled = "Id";
if (linkConfigs.size() == 1){
for (LinkConfig linkConfig : linkConfigs) {
sql = sql + " and " + linkConfig.getField() + " is not null";
allFiled = allFiled + "," + linkConfig.getField();
}
}else {
String info = null;
for (LinkConfig linkConfig : linkConfigs) {
if (StringUtils.isBlank(info)){
info = linkConfig.getField() + " is not null" ;
}else {
info = info + " or " + linkConfig.getField() + " is not null";
}
allFiled = allFiled + "," + linkConfig.getField();
}
sql = sql +" and ("+ info +")";
}
// 表内数据总量 // 表内数据总量
Integer count = customMapper.countBySQL(param.getApi(), "where " + sql); Integer count = customMapper.countBySQL(param.getApi(), "where CreatedDate >= '" + beginDateStr + "' and CreatedDate < '" + endDateStr + "'");
log.info("表api：{} 存在" +count+ "条需更新数据！开始时间:{},结束时间：{}", param.getApi(), beginDateStr, endDateStr); log.info("表api：{} 存在" +count+ "条数据！", param.getApi());
if (count >0 ) { if (count >0 ) {
try { int page = count % 2000 == 0 ? count / 2000 : (count / 2000) + 1;
int page = count % 2000 == 0 ? count / 2000 : (count / 2000) + 1;
for (int i = 0; i < page; i++) { for (int i = 0; i < page; i++) {
log.info("表api：{}，批次：{}，单批次数：2000，开始时间：{}，结束时间：{}，执行数据更新!", param.getApi(), i,beginDateStr, endDateStr); log.info("表api：{}，数据量：{}，执行数据更新!", param.getApi(), 2000* (i+1));
List<Map<String, Object>> mapList = customMapper.list(allFiled, param.getApi(), sql + " order by Id limit " + i * 2000 + ",2000"); List<Map<String, Object>> mapList = customMapper.list("*", param.getApi(), "1=1 order by Id limit " + i * 2000 + ",2000");
for (int j = 1; j <= mapList.size(); j++) { List<Map<String, Object>> updateMapList = new ArrayList<>();
List<Map<String, Object>> updateMapList = new ArrayList<>(); for (int j = 1; j <= mapList.size(); j++) {
Map<String, Object> map = mapList.get(j - 1); Map<String, Object> map = mapList.get(j - 1);
for (LinkConfig config : linkConfigs) { for (LinkConfig config : linkConfigs) {
if (map.get(config.getField()) != null){ if (map.get(config.getField()) != null){
String type = resultMap.get(map.get(config.getField()).toString().substring(0, 3)); String type = resultMap.get(map.get(config.getField()).toString().substring(0, 3));
Map<String, Object> paramMap = Maps.newHashMap(); Map<String, Object> paramMap = Maps.newHashMap();
paramMap.put("key", config.getLinkField()); paramMap.put("key", config.getLinkField());
paramMap.put("value", type); paramMap.put("value", type);
updateMapList.add(paramMap); updateMapList.add(paramMap);
}
}
if (!updateMapList.isEmpty()) {
customMapper.updateById(param.getApi(), updateMapList, String.valueOf(mapList.get(j - 1).get("Id") != null?mapList.get(j - 1).get("Id") : mapList.get(j - 1).get("id")));
}else {
log.error("对象：{}的Id为：{}数据不存在可更新的关联类型！！！！",param.getApi(),String.valueOf(mapList.get(j - 1).get("Id") != null?mapList.get(j - 1).get("Id") : mapList.get(j - 1).get("id")));
} }
} }
customMapper.updateById(param.getApi(), updateMapList, String.valueOf(mapList.get(j - 1).get("Id") != null?mapList.get(j - 1).get("Id") : mapList.get(j - 1).get("id")));
} }
}catch (Exception e){
log.info(e.getMessage());
} }
} }
@ -1915,9 +1884,8 @@ public class DataImportNewServiceImpl implements DataImportNewService {
List<Map<String, Object>> mapList = customMapper.list("*", api, sql2 + i * 2000 + ",2000" ); List<Map<String, Object>> mapList = customMapper.list("*", api, sql2 + i * 2000 + ",2000" );
List<Map<String, Object>> updateMapList = new ArrayList<>();
for (int j = 1; j <= mapList.size(); j++) { for (int j = 1; j <= mapList.size(); j++) {
List<Map<String, Object>> updateMapList = new ArrayList<>();
Map<String, Object> map = mapList.get(j - 1); Map<String, Object> map = mapList.get(j - 1);
for (LinkConfig config : linkConfigs) { for (LinkConfig config : linkConfigs) {
if (map.get(config.getField()) != null){ if (map.get(config.getField()) != null){