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

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

Browse Source
This commit is contained in:
Kris 2025-07-29 14:05:38 +08:00
parent ab78eb2039
commit e658a1b61f
4 changed files with 251 additions and 7 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
src/main/java/com/celnet/datadump/job/DataDumpNewJob.java
View File

@ -191,7 +191,7 @@ public class DataDumpNewJob {
return new ReturnT<>(500, "参数解析失败！");
}
return commonService.updateLinkType(param);
return dataImportNewService.updateLinkTypeJob(param);
}
/**

8
src/main/java/com/celnet/datadump/service/DataImportNewService.java
View File

@ -39,5 +39,11 @@ public interface DataImportNewService {
ReturnT<String> uploadFileNew(SalesforceParam param, DataObject dataObject) throws Exception;
/**
* 更新关联类型
* @param param 参数
* @return ReturnT
* @throws Exception exception
*/
ReturnT<String> updateLinkTypeJob(SalesforceParam param) throws Exception;
}

2
src/main/java/com/celnet/datadump/service/impl/DataImportBatchServiceImpl.java
View File

@ -517,7 +517,7 @@ public class DataImportBatchServiceImpl implements DataImportBatchService {
}
/**
* 【单彪】组装Update执行参数
* 【单表】组装Update执行参数
*/
public ReturnT<String> updateSfDataBatch(SalesforceParam param, List<Future<?>> futures) throws Exception {
List<String> apis;

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

@ -1,5 +1,6 @@
package com.celnet.datadump.service.impl;
import cn.hutool.core.lang.UUID;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson2.JSONObject;
@ -16,12 +17,13 @@ import com.celnet.datadump.mapper.CustomMapper;
import com.celnet.datadump.param.DataDumpParam;
import com.celnet.datadump.param.SalesforceParam;
import com.celnet.datadump.service.*;
import com.celnet.datadump.util.DataUtil;
import com.celnet.datadump.util.EmailUtil;
import com.celnet.datadump.util.HttpUtil;
import com.celnet.datadump.util.OssUtil;
import com.celnet.datadump.util.*;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.sforce.async.BatchInfo;
import com.sforce.async.BulkConnection;
import com.sforce.async.JobInfo;
import com.sforce.async.OperationEnum;
import com.sforce.soap.partner.*;
import com.sforce.soap.partner.sobject.SObject;
import com.xxl.job.core.biz.model.ReturnT;
@ -1665,4 +1667,240 @@ public class DataImportNewServiceImpl implements DataImportNewService {
}
}
/**
* updateLinkType入口
*/
@Override
public ReturnT<String> updateLinkTypeJob(SalesforceParam param) throws Exception {
List<Future<?>> futures = Lists.newArrayList();
try {
if (param.getType() == 1){
return updateLinkType(param, futures);
}else {
return updateLinkTypeBatch(param,futures);
}
} catch (InterruptedException e) {
throw e;
} catch (Throwable throwable) {
salesforceExecutor.remove(futures.toArray(new Future<?>[]{}));
log.error("updateLinkTypeJob error", throwable);
throw throwable;
}
}
/**
* 【batch逻辑】组装updateLinkType执行参数
*/
public ReturnT<String> updateLinkTypeBatch(SalesforceParam param, List<Future<?>> futures) throws Exception {
List<String> apis;
if (StringUtils.isBlank(param.getApi())) {
apis = dataObjectService.list().stream().map(DataObject::getName).collect(Collectors.toList());
} else {
apis = DataUtil.toIdList(param.getApi());
}
String beginDateStr = null;
String endDateStr = null;
if (param.getBeginCreateDate() != null && param.getEndCreateDate() != null){
Date beginDate = param.getBeginCreateDate();
Date endDate = param.getEndCreateDate();
beginDateStr = DateUtil.format(beginDate, "yyyy-MM-dd HH:mm:ss");
endDateStr = DateUtil.format(endDate, "yyyy-MM-dd HH:mm:ss");
}
QueryWrapper<LinkConfig> wrapper = new QueryWrapper<>();
wrapper.in("api", apis);
Map<String, String> resultMap = dataObjectService.list().stream()
.collect(Collectors.toMap(
DataObject::getKeyPrefix,
DataObject::getName,
(existing, replacement) -> replacement));
for (String api : apis) {
try {
List<SalesforceParam> salesforceParams = null;
QueryWrapper<DataBatch> dbQw = new QueryWrapper<>();
dbQw.eq("name", api);
if (StringUtils.isNotEmpty(beginDateStr) && StringUtils.isNotEmpty(endDateStr)) {
dbQw.eq("sync_start_date", beginDateStr); // 等于开始时间
dbQw.eq("sync_end_date", endDateStr); // 等于结束时间
}
List<DataBatch> list = dataBatchService.list(dbQw);
AtomicInteger batch = new AtomicInteger(1);
if (CollectionUtils.isNotEmpty(list)) {
salesforceParams = list.stream().map(t -> {
SalesforceParam salesforceParam = param.clone();
salesforceParam.setApi(t.getName());
salesforceParam.setBeginCreateDate(t.getSyncStartDate());
salesforceParam.setEndCreateDate(t.getSyncEndDate());
salesforceParam.setBatch(batch.getAndIncrement());
return salesforceParam;
}).collect(Collectors.toList());
}
// 手动任务优先执行
for (SalesforceParam salesforceParam : salesforceParams) {
Future<?> future = salesforceExecutor.execute(() -> {
try {
updateLinkBatch(salesforceParam, resultMap);
} catch (Throwable throwable) {
log.error("salesforceExecutor error", throwable);
throw new RuntimeException(throwable);
}
}, salesforceParam.getBatch(), 1);
futures.add(future);
}
// 等待当前所有线程执行完成
salesforceExecutor.waitForFutures(futures.toArray(new Future<?>[]{}));
} catch (Exception e) {
throw e;
}
}
return null;
}
/**
* 执行数据updateLinkType
*/
private void updateLinkBatch(SalesforceParam param, Map<String, String> resultMap) throws Exception {
QueryWrapper<LinkConfig> dbQw = new QueryWrapper<>();
dbQw.eq("api", param.getApi());
List<LinkConfig> linkConfigs = linkConfigService.list(dbQw);
String beginDateStr = null;
String endDateStr = null;
Date beginDate = param.getBeginCreateDate();
Date endDate = param.getEndCreateDate();
if (param.getBeginCreateDate() != null && param.getEndCreateDate() != null){
beginDateStr = DateUtil.format(beginDate, "yyyy-MM-dd HH:mm:ss");
endDateStr = DateUtil.format(endDate, "yyyy-MM-dd HH:mm:ss");
}
// 表内数据总量
Integer count = customMapper.countBySQL(param.getApi(), "where CreatedDate >= '" + beginDateStr + "' and CreatedDate < '" + endDateStr + "'");
log.info("表api：{} 存在" +count+ "条数据！", param.getApi());
if (count >0 ) {
int page = count % 2000 == 0 ? count / 2000 : (count / 2000) + 1;
for (int i = 0; i < page; i++) {
log.info("表api：{}，数据量：{}，执行数据更新!", param.getApi(), 2000* (i+1));
List<Map<String, Object>> mapList = customMapper.list("*", param.getApi(), "1=1 order by Id limit " + i * 2000 + ",2000");
List<Map<String, Object>> updateMapList = new ArrayList<>();
for (int j = 1; j <= mapList.size(); j++) {
Map<String, Object> map = mapList.get(j - 1);
for (LinkConfig config : linkConfigs) {
if (map.get(config.getField()) != null){
String type = resultMap.get(map.get(config.getField()).toString().substring(0, 3));
Map<String, Object> paramMap = Maps.newHashMap();
paramMap.put("key", config.getLinkField());
paramMap.put("value", type);
updateMapList.add(paramMap);
}
}
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")));
}
}
}
}
/**
* 组装updateLinkType执行参数
*/
public ReturnT<String> updateLinkType(SalesforceParam param, List<Future<?>> futures) throws Exception {
List<String> apis;
if (StringUtils.isBlank(param.getApi())) {
apis = dataObjectService.list().stream().map(DataObject::getName).collect(Collectors.toList());
} else {
apis = DataUtil.toIdList(param.getApi());
}
QueryWrapper<LinkConfig> wrapper = new QueryWrapper<>();
wrapper.in("api", apis).eq("is_link",1);
Map<String, String> resultMap = dataObjectService.list().stream()
.collect(Collectors.toMap(
DataObject::getKeyPrefix,
DataObject::getName,
(existing, replacement) -> replacement));
for (String api : apis) {
Future<?> future = salesforceExecutor.execute(() -> {
try {
updateLink(api,param, resultMap);
} catch (Throwable throwable) {
log.error("salesforceExecutor error", throwable);
throw new RuntimeException(throwable);
}
}, 0, 1);
futures.add(future);
}
// 等待当前所有线程执行完成
salesforceExecutor.waitForFutures(futures.toArray(new Future<?>[]{}));
return ReturnT.SUCCESS;
}
/**
* 执行数据updateLinkType
*/
private void updateLink(String api,SalesforceParam param, Map<String, String> resultMap) throws Exception {
QueryWrapper<LinkConfig> dbQw = new QueryWrapper<>();
dbQw.eq("api", param.getApi());
List<LinkConfig> linkConfigs = linkConfigService.list(dbQw);
String beginDateStr = null;
Date beginDate = param.getBeginModifyDate();
if (param.getBeginModifyDate() != null ){
beginDateStr = DateUtil.format(beginDate, "yyyy-MM-dd HH:mm:ss");
}
String sql1 = null;
String sql2 = null;
if (beginDateStr != null){
sql1 = "where SystemModstamp >= '" + beginDateStr;
sql2 = "SystemModstamp >= "+ beginDateStr +" order by Id limit " ;
}else {
sql1 = null;
sql2 = "1=1 order by Id limit " ;
}
// 表内数据总量
Integer count = customMapper.countBySQL(api, sql1);
log.info("表api：{} 存在" +count+ "条数据！", param.getApi());
if (count >0 ) {
int page = count % 2000 == 0 ? count / 2000 : (count / 2000) + 1;
for (int i = 0; i < page; i++) {
log.info("表api：{}，数据量：{}，执行数据更新!", api, 2000* (i+1));
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++) {
Map<String, Object> map = mapList.get(j - 1);
for (LinkConfig config : linkConfigs) {
if (map.get(config.getField()) != null){
String type = resultMap.get(map.get(config.getField()).toString().substring(0, 3));
Map<String, Object> paramMap = Maps.newHashMap();
paramMap.put("key", config.getLinkField());
paramMap.put("value", type);
updateMapList.add(paramMap);
}
customMapper.updateById(api, updateMapList, String.valueOf(mapList.get(j - 1).get("Id") != null?mapList.get(j - 1).get("Id") : mapList.get(j - 1).get("id")));
}
}
}
}
}
}