From 03fa708ed73757058ff8f87cc3a5bbbf43f44628 Mon Sep 17 00:00:00 2001
From: Kris <2893855659@qq.com>
Date: Thu, 25 Sep 2025 10:19:57 +0800
Subject: [PATCH] =?UTF-8?q?=E3=80=90feat=E3=80=91=20bigObject=E6=95=B0?=
=?UTF-8?q?=E6=8D=AE=E6=8E=A8=E9=80=81=E3=80=81=E5=88=A0=E9=99=A4=E7=9B=AE?=
=?UTF-8?q?=E6=A0=87org=E6=95=B0=E6=8D=AE=E3=80=81=E5=8F=8D=E5=86=99newId?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit
---
.run/data-dump（准时达） .run.xml | 14 +
.run/data-dump（格凌) .run.xml | 14 +
...）.run.xml => data-dump（雨花石测试）.run.xml} | 2 +-
DockerfileBase | 2 +-
.../datadump/controller/JobController.java | 15 +-
.../com/celnet/datadump/job/DataDumpJob.java | 21 +
.../celnet/datadump/job/DataDumpNewJob.java | 89 +++-
.../celnet/datadump/mapper/CustomMapper.java | 11 +-
.../datadump/service/CommonBatchService.java | 5 +-
.../datadump/service/CommonService.java | 2 +
.../service/DataImportNewService.java | 9 +-
.../service/impl/CommonBatchServiceImpl.java | 278 +++++++++++
.../service/impl/CommonServiceImpl.java | 71 ++-
.../impl/DataImportBatchServiceImpl.java | 25 +-
.../impl/DataImportNewServiceImpl.java | 449 +++++++++++-------
.../service/impl/DataImportServiceImpl.java | 8 +-
.../service/impl/DataVerifyServiceImpl.java | 11 +-
.../service/impl/FileServiceImpl.java | 66 ++-
.../service/impl/OrgConfigServiceImpl.java | 16 +-
src/main/resources/mapper/CustomMapper.xml | 10 +
.../resources/mapper/SalesforceMapper.xml | 31 ++
21 files changed, 941 insertions(+), 208 deletions(-)
create mode 100644 .run/data-dump（准时达） .run.xml
create mode 100644 .run/data-dump（格凌) .run.xml
rename .run/{data-dump（城博测试）.run.xml => data-dump（雨花石测试）.run.xml} (80%)
diff --git a/.run/data-dump（准时达） .run.xml b/.run/data-dump（准时达） .run.xml
new file mode 100644
index 0000000..6c4119d
--- /dev/null
+++ b/.run/data-dump（准时达） .run.xml
@@ -0,0 +1,14 @@
+
+
+
+
+
+
+
+
+
+
+
+
+
+
\ No newline at end of file
diff --git a/.run/data-dump（格凌) .run.xml b/.run/data-dump（格凌) .run.xml
new file mode 100644
index 0000000..51956bc
--- /dev/null
+++ b/.run/data-dump（格凌) .run.xml
@@ -0,0 +1,14 @@
+
+
+
+
+
+
+
+
+
+
+
+
+
+
\ No newline at end of file
diff --git a/.run/data-dump（城博测试）.run.xml b/.run/data-dump（雨花石测试）.run.xml
similarity index 80%
rename from .run/data-dump（城博测试）.run.xml
rename to .run/data-dump（雨花石测试）.run.xml
index 60cda99..f1e87cd 100644
--- a/.run/data-dump（城博测试）.run.xml
+++ b/.run/data-dump（雨花石测试）.run.xml
@@ -1,5 +1,5 @@
-
+
diff --git a/DockerfileBase b/DockerfileBase
index e274611..b093d39 100644
--- a/DockerfileBase
+++ b/DockerfileBase
@@ -1,5 +1,5 @@
FROM --platform=linux/amd64 centos:7
-MAINTAINER Red
+LABEL maintainer="Kris"
#nginx部署
ADD nginx-1.22.1.tar.gz /usr/local/
diff --git a/src/main/java/com/celnet/datadump/controller/JobController.java b/src/main/java/com/celnet/datadump/controller/JobController.java
index 75ae108..c3da061 100644
--- a/src/main/java/com/celnet/datadump/controller/JobController.java
+++ b/src/main/java/com/celnet/datadump/controller/JobController.java
@@ -363,17 +363,30 @@ public class JobController {
case "Contact":
dataImportNewService.getPersonContact(param);
case "User":
+ dataImportNewService.getNewIdByField(api,"Username");
case "BrandTemplate":
case "EmailTemplate":
case "Folder":
+ dataImportNewService.getNewIdByField(api,"DeveloperName");
case "Group":
+ dataImportNewService.getNewIdByField(api,"DeveloperName,Type");
case "RecordType":
+ dataImportNewService.getNewIdByField(api,"DeveloperName,SObjectType");
case "AccountContactRelation":
+ dataImportNewService.getNewIdByField(api,"AccountId,ContactId,IsPrimary");
case "TaskRelation":
+ dataImportNewService.getNewIdByField(api,"TaskId,RelationId,RelationType");
case "EventRelation":
+ dataImportNewService.getNewIdByField(api,"EventId,RelationId,RelationType");
case "CollaborationGroupMember":
+ dataImportNewService.getNewIdByField(api,"CollaborationGroupId,MemberId,CollaborationRole");
case "FeedAttachment":
- dataImportNewService.getNewIdByField(api,null);
+ dataImportNewService.getNewIdByField(api,"FeedItemId,AttachmentId");
+ case "Task":
+ dataImportNewService.getTaskNewId(param);
+ case "PermissionSet":
+ dataImportNewService.getNewIdByField(api,"Name");
+
default:
log.info("当前对象：{}，不存在反写NewId逻辑！！！" , param.getApi());
}
diff --git a/src/main/java/com/celnet/datadump/job/DataDumpJob.java b/src/main/java/com/celnet/datadump/job/DataDumpJob.java
index 2fd6e33..0ac1695 100644
--- a/src/main/java/com/celnet/datadump/job/DataDumpJob.java
+++ b/src/main/java/com/celnet/datadump/job/DataDumpJob.java
@@ -48,6 +48,27 @@ public class DataDumpJob {
return commonService.createApi(param);
}
+ /**
+ * 创建bigobject api
+ * @param paramStr
+ * @return
+ * @throws Exception
+ */
+ @XxlJob("createBigObjectApiJob")
+ public ReturnT createBigObjectApiJob(String paramStr) throws Exception {
+ log.info("createBigObjectApiJob execute start ..................");
+ SalesforceParam param = new SalesforceParam();
+ try {
+ if (StringUtils.isNotBlank(paramStr)) {
+ param = JSON.parseObject(paramStr, SalesforceParam.class);
+ }
+ } catch (Throwable throwable) {
+ return new ReturnT<>(500, "参数解析失败！");
+ }
+
+ return commonService.createBigObjectApi(param);
+ }
+
/**
* 存量任务
*
diff --git a/src/main/java/com/celnet/datadump/job/DataDumpNewJob.java b/src/main/java/com/celnet/datadump/job/DataDumpNewJob.java
index 7d7d5ef..e710306 100644
--- a/src/main/java/com/celnet/datadump/job/DataDumpNewJob.java
+++ b/src/main/java/com/celnet/datadump/job/DataDumpNewJob.java
@@ -114,6 +114,54 @@ public class DataDumpNewJob {
return commonBatchService.dumpBatch(param);
}
+ /**
+ * 存量任务（BigObject）
+ *
+ * @param paramStr 参数json
+ * @return result
+ */
+ @XxlJob("dataDumpBigObjectJob")
+ public ReturnT dataDumpBigObjectJob(String paramStr) throws Exception {
+ log.info("dataDumpBigObjectJob execute start ..................");
+ SalesforceParam param = new SalesforceParam();
+ try {
+ if (StringUtils.isNotBlank(paramStr)) {
+ param = JSON.parseObject(paramStr, SalesforceParam.class);
+ }
+ } catch (Throwable throwable) {
+ return new ReturnT<>(500, "参数解析失败！");
+ }
+ // 参数转换
+ param.setBeginCreateDate(param.getBeginDate());
+ param.setEndCreateDate(param.getEndDate());
+
+ return commonBatchService.dumpBigObject(param);
+ }
+
+ /**
+ * 导入BigObject数据到目标ORG
+ *
+ * @param paramStr 参数json
+ * @return result
+ */
+ @XxlJob("dataImportBigObjectJob")
+ public ReturnT dataImportBigObjectJob(String paramStr) throws Exception {
+ log.info("dataImportBigObjectJob execute start ..................");
+ SalesforceParam param = new SalesforceParam();
+ try {
+ if (StringUtils.isNotBlank(paramStr)) {
+ param = JSON.parseObject(paramStr, SalesforceParam.class);
+ }
+ } catch (Throwable throwable) {
+ return new ReturnT<>(500, "参数解析失败！");
+ }
+ // 参数转换
+ param.setBeginCreateDate(param.getBeginDate());
+ param.setEndCreateDate(param.getEndDate());
+
+ return commonBatchService.importBigObject(param);
+ }
+
/**
* 返写NewId
* @param paramStr 参数json
@@ -143,30 +191,46 @@ public class DataDumpNewJob {
switch (api){
case "Contact":
dataImportNewService.getPersonContact(param);
+ break;
case "User":
dataImportNewService.getNewIdByField(api,"Username");
+ break;
case "BrandTemplate":
case "EmailTemplate":
case "Folder":
dataImportNewService.getNewIdByField(api,"DeveloperName");
+ break;
case "Group":
dataImportNewService.getNewIdByField(api,"DeveloperName,Type");
+ break;
case "RecordType":
dataImportNewService.getNewIdByField(api,"DeveloperName,SObjectType");
+ break;
case "AccountContactRelation":
dataImportNewService.getNewIdByField(api,"AccountId,ContactId,IsPrimary");
+ break;
case "TaskRelation":
dataImportNewService.getNewIdByField(api,"TaskId,RelationId,RelationType");
+ break;
case "EventRelation":
dataImportNewService.getNewIdByField(api,"EventId,RelationId,RelationType");
+ break;
case "CollaborationGroupMember":
dataImportNewService.getNewIdByField(api,"CollaborationGroupId,MemberId,CollaborationRole");
+ break;
case "FeedAttachment":
dataImportNewService.getNewIdByField(api,"FeedItemId,AttachmentId");
+ break;
case "Task":
dataImportNewService.getTaskNewId(param);
+ break;
+ case "PermissionSet":
+ case "Organization":
+ dataImportNewService.getNewIdByField(api,"Name");
+ break;
default:
- log.info("当前对象：{}，不存在反写NewId逻辑！！！" , param.getApi());
+ log.info("当前对象：{}，不存在反写NewId逻辑！！！" , api);
+ break;
}
}
return ReturnT.SUCCESS;
@@ -404,6 +468,27 @@ public class DataDumpNewJob {
return dataImportNewService.checkDeletedData(param);
}
+ /**
+ * 删除目标ORG中的数据
+ * @param paramStr 参数json
+ * @return result
+ */
+ @XxlJob("deleteTargetOrgDataJob")
+ public ReturnT deleteTargetOrgDataJob(String paramStr) throws Exception {
+ log.info("deleteTargetOrgDataJob execute start ..................");
+ SalesforceParam param = new SalesforceParam();
+ try {
+ if (StringUtils.isNotBlank(paramStr)) {
+ param = JSON.parseObject(paramStr, SalesforceParam.class);
+ }
+ } catch (Throwable throwable) {
+ log.error("参数解析失败！", throwable);
+ return new ReturnT<>(500, "参数解析失败！");
+ }
+
+ // 调用新实现的删除方法
+ return dataImportNewService.deleteTargetOrgData(param);
+ }
/**
* ContentFolderMember 拉取
@@ -424,4 +509,6 @@ public class DataDumpNewJob {
return dataImportNewService.dumpContentFolderMemberJob(param);
}
+
+
}
diff --git a/src/main/java/com/celnet/datadump/mapper/CustomMapper.java b/src/main/java/com/celnet/datadump/mapper/CustomMapper.java
index da35166..045fb7b 100644
--- a/src/main/java/com/celnet/datadump/mapper/CustomMapper.java
+++ b/src/main/java/com/celnet/datadump/mapper/CustomMapper.java
@@ -4,6 +4,7 @@ import com.celnet.datadump.param.SalesforceParam;
import cn.hutool.json.JSONObject;
import org.apache.ibatis.annotations.Mapper;
import org.apache.ibatis.annotations.Param;
+import org.apache.ibatis.annotations.MapKey;
import java.util.Collection;
import java.util.List;
@@ -134,6 +135,14 @@ public interface CustomMapper {
*/
public void updateByNewId(@Param("maps") List> maps, @Param("tableName") String api, @Param("newId") String newId);
+ /**
+ * 根据map中的字段条件更新数据的new_id字段
+ * @param tableName 表名
+ * @param maps 包含条件字段的map
+ */
+ public void updateNewIdByFields(@Param("tableName") String tableName,
+ @Param("maps") List> maps,
+ @Param("newId") String newId);
/**
* 插入方法
*
@@ -199,4 +208,4 @@ public interface CustomMapper {
public void deleteOne(@Param("tableName") String tableName, @Param("id") String id);
-}
+}
\ No newline at end of file
diff --git a/src/main/java/com/celnet/datadump/service/CommonBatchService.java b/src/main/java/com/celnet/datadump/service/CommonBatchService.java
index c7e41d6..16b37fc 100644
--- a/src/main/java/com/celnet/datadump/service/CommonBatchService.java
+++ b/src/main/java/com/celnet/datadump/service/CommonBatchService.java
@@ -7,5 +7,8 @@ public interface CommonBatchService {
ReturnT dumpBatch(SalesforceParam param) throws Exception;
+ ReturnT dumpBigObject(SalesforceParam param) throws Exception;
+
+ ReturnT importBigObject(SalesforceParam param) throws Exception;
-}
+}
\ No newline at end of file
diff --git a/src/main/java/com/celnet/datadump/service/CommonService.java b/src/main/java/com/celnet/datadump/service/CommonService.java
index 4ae07b9..67530d9 100644
--- a/src/main/java/com/celnet/datadump/service/CommonService.java
+++ b/src/main/java/com/celnet/datadump/service/CommonService.java
@@ -37,6 +37,8 @@ public interface CommonService {
*/
ReturnT createApi(SalesforceParam param) throws Exception;
+ ReturnT createBigObjectApi(SalesforceParam param) throws Exception;
+
/**
* 获取所有对象
* @return
diff --git a/src/main/java/com/celnet/datadump/service/DataImportNewService.java b/src/main/java/com/celnet/datadump/service/DataImportNewService.java
index 52050e5..ee74e5b 100644
--- a/src/main/java/com/celnet/datadump/service/DataImportNewService.java
+++ b/src/main/java/com/celnet/datadump/service/DataImportNewService.java
@@ -22,6 +22,13 @@ public interface DataImportNewService {
ReturnT checkDeletedData(SalesforceParam param) throws Exception;
+ /**
+ * 删除目标ORG中的数据
+ * @param param 参数
+ * @return 执行结果
+ * @throws Exception 异常
+ */
+ ReturnT deleteTargetOrgData(SalesforceParam param) throws Exception;
ReturnT insertSingle(SalesforceParam param) throws Exception;
@@ -31,4 +38,4 @@ public interface DataImportNewService {
void getTaskNewId(SalesforceParam param) throws Exception;
-}
+}
\ No newline at end of file
diff --git a/src/main/java/com/celnet/datadump/service/impl/CommonBatchServiceImpl.java b/src/main/java/com/celnet/datadump/service/impl/CommonBatchServiceImpl.java
index 13ceb4e..e8bd11d 100644
--- a/src/main/java/com/celnet/datadump/service/impl/CommonBatchServiceImpl.java
+++ b/src/main/java/com/celnet/datadump/service/impl/CommonBatchServiceImpl.java
@@ -8,6 +8,7 @@ import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.core.conditions.update.UpdateWrapper;
import com.celnet.datadump.config.SalesforceConnect;
import com.celnet.datadump.config.SalesforceExecutor;
+import com.celnet.datadump.config.SalesforceTargetConnect;
import com.celnet.datadump.entity.*;
import com.celnet.datadump.global.Const;
import com.celnet.datadump.mapper.CustomMapper;
@@ -33,6 +34,7 @@ import com.sforce.soap.partner.sobject.SObject;
import com.sforce.ws.parser.XmlInputStream;
import com.xxl.job.core.biz.model.ReturnT;
import com.xxl.job.core.log.XxlJobLogger;
+import com.xxl.job.core.util.DateUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang3.ObjectUtils;
@@ -67,6 +69,8 @@ public class CommonBatchServiceImpl implements CommonBatchService {
@Autowired
private SalesforceExecutor salesforceExecutor;
@Autowired
+ private SalesforceTargetConnect salesforceTargetConnect;
+ @Autowired
private DataObjectService dataObjectService;
@Autowired
private DataBatchService dataBatchService;
@@ -104,6 +108,276 @@ public class CommonBatchServiceImpl implements CommonBatchService {
}
}
+ @Override
+ public ReturnT dumpBigObject(SalesforceParam param) throws Exception {
+ List> futures = Lists.newArrayList();
+ try {
+ if (StringUtils.isNotBlank(param.getApi())) {
+ // 手动任务
+ ReturnT result = manualBigObjectDump(param, futures);
+ if (result != null) {
+ return result;
+ }
+ }
+ return ReturnT.SUCCESS;
+ } catch (Throwable throwable) {
+ salesforceExecutor.remove(futures.toArray(new Future[]{}));
+ log.error("dump error", throwable);
+ throw throwable;
+ }
+ }
+
+ @Override
+ public ReturnT importBigObject(SalesforceParam param) throws Exception {
+ List> futures = Lists.newArrayList();
+ try {
+ if (StringUtils.isNotBlank(param.getApi())) {
+ // 手动任务
+ ReturnT result = manualBigObjectImport(param, futures);
+ if (result != null) {
+ return result;
+ }
+ }
+ return ReturnT.SUCCESS;
+ } catch (Throwable throwable) {
+ salesforceExecutor.remove(futures.toArray(new Future[]{}));
+ log.error("import error", throwable);
+ throw throwable;
+ }
+ }
+
+ private ReturnT manualBigObjectDump(SalesforceParam param, List> futures) throws Exception {
+
+ String api = param.getApi();
+
+ BulkConnection bulkConnect = salesforceConnect.createBulkConnect();
+
+ 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");
+ }
+
+ try {
+ List salesforceParams = null;
+ QueryWrapper 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 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) {
+ if (param.getIsSingleThread()){
+ manualBigObject(salesforceParam, bulkConnect);
+ }else {
+ Future future = salesforceExecutor.execute(() -> {
+ try {
+ manualBigObject(salesforceParam, bulkConnect);
+ } 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 (InterruptedException interruptedException){
+ return ReturnT.FAIL;
+ }catch (Exception e) {
+ throw e;
+ }
+ return ReturnT.SUCCESS;
+ }
+
+ private void manualBigObject(SalesforceParam param, BulkConnection bulkConnect) throws Exception {
+
+ String sql = "select * from " + param.getApi() + " where CreatedDate >= '" + param.getBeginCreateDate() + "' and CreatedDate <= '" + param.getEndCreateDate() + "'";
+
+ log.info("BigObject query SQL: {}", sql);
+ XxlJobLogger.log("BigObject query SQL: {}", sql);
+
+ // 获取Salesforce实例URL
+ String restEndpoint = bulkConnect.getConfig().getRestEndpoint();
+ String url = restEndpoint.substring(0, restEndpoint.indexOf("services/"));
+
+ // 创建Bulk API 2.0查询作业
+ String jobId = BulkUtil.createBulkV2Job(url, sql, bulkConnect);
+ log.info("Created Bulk API 2.0 job with ID: {}", jobId);
+
+ // 等待作业完成
+ BulkUtil.waitForBulkV2JobCompletion(url, jobId, bulkConnect);
+
+ // 获取查询结果
+ String csvResult = BulkUtil.getBulkV2QueryJobResults(url, jobId, bulkConnect);
+
+ // 保存到本地文件
+ String fileName = param.getApi() + "_" + System.currentTimeMillis() + ".csv";
+ String filePath = "bigobject/" + fileName;
+
+ // 确保目录存在
+ File directory = new File("bigobject");
+ if (!directory.exists()) {
+ directory.mkdirs();
+ }
+
+ // 写入文件
+ try (FileWriter fileWriter = new FileWriter(filePath);
+ BufferedWriter bufferedWriter = new BufferedWriter(fileWriter)) {
+ bufferedWriter.write(csvResult);
+ }
+
+ log.info("BigObject data saved to file: {}", filePath);
+ XxlJobLogger.log("BigObject data saved to file: {}", filePath);
+ }
+
+ private ReturnT manualBigObjectImport(SalesforceParam param, List> futures) throws Exception {
+ String api = param.getApi();
+
+ // 创建目标ORG连接
+ BulkConnection targetBulkConnect = salesforceTargetConnect.createBulkConnect();
+
+ 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");
+ }
+
+ try {
+ List salesforceParams = null;
+ QueryWrapper 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 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) {
+ if (param.getIsSingleThread()){
+ manualImportBigObject(salesforceParam, targetBulkConnect);
+ }else {
+ Future future = salesforceExecutor.execute(() -> {
+ try {
+ manualImportBigObject(salesforceParam, targetBulkConnect);
+ } 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 (InterruptedException interruptedException){
+ return ReturnT.FAIL;
+ }catch (Exception e) {
+ throw e;
+ }
+ return ReturnT.SUCCESS;
+ }
+
+ private void manualImportBigObject(SalesforceParam param, BulkConnection targetBulkConnect) throws Exception {
+ String api = param.getApi();
+
+ // 构造CSV文件路径 - 使用与导出时相同的文件名格式（但需要知道确切的时间戳）
+ // 这里应该查找最新的相关文件，而不是随机生成一个文件名
+ File directory = new File("bigobject");
+ if (!directory.exists() || !directory.isDirectory()) {
+ log.error("BigObject目录不存在: {}", directory.getAbsolutePath());
+ throw new FileNotFoundException("BigObject目录不存在: " + directory.getAbsolutePath());
+ }
+
+ // 查找匹配的CSV文件（最近的）
+ File[] files = directory.listFiles((dir, name) ->
+ name.startsWith(api + "_") && name.endsWith(".csv"));
+
+ if (files == null || files.length == 0) {
+ log.error("未找到API {} 的CSV文件", api);
+ throw new FileNotFoundException("未找到API " + api + " 的CSV文件");
+ }
+
+ // 选择最新的文件
+ File csvFile = Arrays.stream(files)
+ .max(Comparator.comparingLong(File::lastModified))
+ .orElse(null);
+
+ if (csvFile == null) {
+ log.error("未找到API {} 的CSV文件", api);
+ throw new FileNotFoundException("未找到API " + api + " 的CSV文件");
+ }
+
+ String filePath = csvFile.getAbsolutePath();
+ log.info("找到CSV文件: {}", filePath);
+
+ log.info("开始导入BigObject数据到目标ORG: {}", api);
+
+ // 创建Bulk API作业
+ JobInfo job = BulkUtil.createJob(targetBulkConnect, api, OperationEnum.insert);
+ log.info("创建Bulk作业成功, 作业ID: {}", job.getId());
+
+ // 从CSV文件创建批次
+ List batchInfos = BulkUtil.createBatchesFromCSVFile(targetBulkConnect, job, filePath);
+ log.info("创建批次成功, 批次数: {}", batchInfos.size());
+
+ // 等待作业完成
+ BulkUtil.awaitCompletion(targetBulkConnect, job, batchInfos);
+ log.info("数据导入完成");
+
+ // 检查作业状态
+ job = targetBulkConnect.getJobStatus(job.getId());
+ if (job.getState() == JobStateEnum.Failed) {
+ log.error("Bulk作业失败, 作业ID: {}", job.getId());
+ throw new RuntimeException("Bulk作业失败 ");
+ }
+
+ // 关闭作业
+ BulkUtil.closeJob(targetBulkConnect, job.getId());
+ log.info("关闭作业成功, 作业ID: {}", job.getId());
+
+ // 记录导入统计信息
+ int totalRecords = batchInfos.stream()
+ .mapToInt(BatchInfo::getNumberRecordsProcessed)
+ .sum();
+ log.info("成功导入 {} 条记录到 {}", totalRecords, api);
+ }
+
private ReturnT manualBatchDump(SalesforceParam param, List> futures) throws InterruptedException {
List apis;
apis = DataUtil.toIdList(param.getApi());
@@ -739,6 +1013,10 @@ public class CommonBatchServiceImpl implements CommonBatchService {
maps.add(paramMap3);
if (existsIds.contains(id)) {
+ Map paramMap = Maps.newHashMap();
+ paramMap.put("key", "is_update");
+ paramMap.put("value",0);
+ maps.add(paramMap);
customMapper.updateById(api, maps, id);
} else {
if (insert) {
diff --git a/src/main/java/com/celnet/datadump/service/impl/CommonServiceImpl.java b/src/main/java/com/celnet/datadump/service/impl/CommonServiceImpl.java
index df88157..c0dbd30 100644
--- a/src/main/java/com/celnet/datadump/service/impl/CommonServiceImpl.java
+++ b/src/main/java/com/celnet/datadump/service/impl/CommonServiceImpl.java
@@ -464,12 +464,12 @@ public class CommonServiceImpl implements CommonService {
// 等待当前所有线程执行完成
salesforceExecutor.waitForFutures(futures.toArray(new Future[]{}));
- update.setDataWork(0);
} catch (Throwable e) {
log.error("manualDump error", e);
throw new RuntimeException(e);
} finally {
if (isFull) {
+ update.setDataWork(0);
update.setName(api);
update.setDataLock(0);
dataObjectService.updateById(update);
@@ -921,6 +921,10 @@ public class CommonServiceImpl implements CommonService {
if (existsIds.contains(id)) {
+ Map paramMap = Maps.newHashMap();
+ paramMap.put("key", "is_update");
+ paramMap.put("value",0);
+ maps.add(paramMap);
customMapper.updateById(api, maps, id);
} else {
if (insert) {
@@ -1599,6 +1603,71 @@ public class CommonServiceImpl implements CommonService {
return ReturnT.SUCCESS;
}
+ @Override
+ public ReturnT createBigObjectApi(SalesforceParam param) throws Exception {
+
+ String api = param.getApi();
+
+ PartnerConnection connection = salesforceConnect.createConnect();
+
+ DescribeSObjectResult dsr = connection.describeSObject(api);
+
+ DataObject dataObject = dataObjectService.getById(api);
+ dataObject.setLabel(dsr.getLabel());
+ dataObjectService.updateById(dataObject);
+
+ Date startDate = dataObject.getLastUpdateDate() == null ? DataUtil.DEFAULT_BEGIN_DATE : dataObject.getLastUpdateDate();
+ Date endCreateDate = DateUtils.parseDate(DateFormatUtils.format(new Date(), "yyyy-MM-dd"), "yyyy-MM-dd");
+
+ // 按年生成批次
+ DataBatch one = new DataBatch();
+ one.setFirstDbNum(0);
+ one.setFirstSfNum(0);
+ one.setDbNum(0);
+ one.setSfNum(0);
+ one.setName(api);
+ one.setLabel(dsr.getLabel());
+
+ // 按年份划分开始时间和结束时间创建批次
+ Calendar startCalendar = Calendar.getInstance();
+ startCalendar.setTime(startDate);
+ int startYear = startCalendar.get(Calendar.YEAR);
+
+ Calendar endCalendar = Calendar.getInstance();
+ endCalendar.setTime(endCreateDate);
+ int endYear = endCalendar.get(Calendar.YEAR);
+
+ // 优化批次创建逻辑，确保正确的日期范围
+ for (int year = startYear; year <= endYear; year++) {
+ Date batchStartDate;
+ Date batchEndDate;
+
+ // 设置批次开始时间为当年1月1日 00:00:00
+ batchStartDate = DateUtils.parseDate(year + "-01-01 00:00:00", "yyyy-MM-dd HH:mm:ss");
+
+ if (year == endYear) {
+ // 结束年份使用实际结束日期
+ batchEndDate = endCreateDate;
+ } else {
+ int nextYear = year + 1;
+ // 其他年份结束时间为当年12月31日 00:00:00
+ batchEndDate = DateUtils.parseDate(nextYear + "-01-01 00:00:00", "yyyy-MM-dd HH:mm:ss");
+ }
+
+ // 确保批次开始时间不早于实际开始时间
+ if (batchStartDate.before(startDate)) {
+ batchStartDate = startDate;
+ }
+
+ DataBatch dataBatch = one.clone();
+ dataBatch.setSyncStartDate(batchStartDate);
+ dataBatch.setSyncEndDate(batchEndDate);
+ dataBatchService.save(dataBatch);
+ }
+
+ return ReturnT.SUCCESS;
+ }
+
@Override
public ReturnT getAllApi() throws Exception {
PartnerConnection partnerConnection = salesforceConnect.createConnect();
diff --git a/src/main/java/com/celnet/datadump/service/impl/DataImportBatchServiceImpl.java b/src/main/java/com/celnet/datadump/service/impl/DataImportBatchServiceImpl.java
index 6b8ea60..144c0eb 100644
--- a/src/main/java/com/celnet/datadump/service/impl/DataImportBatchServiceImpl.java
+++ b/src/main/java/com/celnet/datadump/service/impl/DataImportBatchServiceImpl.java
@@ -826,7 +826,7 @@ public class DataImportBatchServiceImpl implements DataImportBatchService {
for (int i = 0; i < page; i++) {
- List> mapList = customMapper.list("*", api, "new_id is not null and is_update = 0 and CreatedDate >= '" + beginDateStr + "' and CreatedDate < '" + endDateStr + "' order by Id asc limit " + i * 8000 + ",8000");
+ List> mapList = customMapper.list("*", api, "new_id is not null and is_update = 0 and CreatedDate >= '" + beginDateStr + "' and CreatedDate < '" + endDateStr + "' order by Id asc limit 8000");
log.error("总Update数据 count:{}；当前批次:{}；-开始时间:{}；-结束时间:{}；-api:{}；", count,i, beginDateStr, endDateStr, api);
@@ -860,7 +860,18 @@ public class DataImportBatchServiceImpl implements DataImportBatchService {
if (map.get(linkfield)!=null){
reference_to = map.get(linkfield).toString();
}else {
- log.info("对象类型:" + api + "的数据:"+ map.get("Id") +"的关联类型不存在！");
+ message = "对象类型:" + api + "的数据:"+ map.get("Id") +"的关联类型不存在！";
+ List> maps = new ArrayList<>();
+ Map linkMap = new HashMap<>();
+ linkMap.put("key", "is_update");
+ linkMap.put("value", 2);
+ maps.add(linkMap);
+ Map linkMap1 = new HashMap<>();
+ linkMap1.put("key", "error_message");
+ linkMap1.put("value", message);
+ maps.add(linkMap1);
+ customMapper.updateByNewId(maps,api, String.valueOf(map.get("new_id")));
+ log.info(message);
}
}
if (reference_to == null){
@@ -871,6 +882,16 @@ public class DataImportBatchServiceImpl implements DataImportBatchService {
account.put(field, m.get("new_id"));
}else {
message = "对象类型:" + api + "的数据:"+ map.get("Id") +"的引用对象:" + reference_to + "的数据:"+ map.get(field) +"不存在！";
+ List> maps = new ArrayList<>();
+ Map linkMap = new HashMap<>();
+ linkMap.put("key", "is_update");
+ linkMap.put("value", 2);
+ maps.add(linkMap);
+ Map linkMap1 = new HashMap<>();
+ linkMap1.put("key", "error_message");
+ linkMap1.put("value", message);
+ maps.add(linkMap1);
+ customMapper.updateByNewId(maps,api, String.valueOf(map.get("new_id")));
log.info(message);
break;
}
diff --git a/src/main/java/com/celnet/datadump/service/impl/DataImportNewServiceImpl.java b/src/main/java/com/celnet/datadump/service/impl/DataImportNewServiceImpl.java
index 6ea641a..3527b70 100644
--- a/src/main/java/com/celnet/datadump/service/impl/DataImportNewServiceImpl.java
+++ b/src/main/java/com/celnet/datadump/service/impl/DataImportNewServiceImpl.java
@@ -836,7 +836,18 @@ public class DataImportNewServiceImpl implements DataImportNewService {
if (map.get(linkfield)!=null){
reference_to = map.get(linkfield).toString();
}else {
- log.info("对象类型:" + api + "的数据:"+ map.get("Id") +"的关联类型不存在！");
+ message = "对象类型:" + api + "的数据:"+ map.get("Id") +"的关联类型不存在！";
+ List> maps = new ArrayList<>();
+ Map linkMap = new HashMap<>();
+ linkMap.put("key", "is_update");
+ linkMap.put("value", 2);
+ maps.add(linkMap);
+ Map linkMap1 = new HashMap<>();
+ linkMap1.put("key", "error_message");
+ linkMap1.put("value", message);
+ maps.add(linkMap1);
+ customMapper.updateByNewId(maps,api, String.valueOf(map.get("new_id")));
+ log.info(message);
}
}
if (reference_to == null){
@@ -847,6 +858,16 @@ public class DataImportNewServiceImpl implements DataImportNewService {
account.setField(field, m.get("new_id"));
}else {
message = "对象类型:" + api + "的数据:"+ map.get("Id") +"的引用对象:" + reference_to + "的数据:"+ map.get(field) +"不存在！";
+ List> maps = new ArrayList<>();
+ Map linkMap = new HashMap<>();
+ linkMap.put("key", "is_update");
+ linkMap.put("value", 2);
+ maps.add(linkMap);
+ Map linkMap1 = new HashMap<>();
+ linkMap1.put("key", "error_message");
+ linkMap1.put("value", message);
+ maps.add(linkMap1);
+ customMapper.updateByNewId(maps,api, String.valueOf(map.get("new_id")));
log.info(message);
break;
}
@@ -1214,7 +1235,7 @@ public class DataImportNewServiceImpl implements DataImportNewService {
try {
if (list != null && !list.isEmpty()) {
//表内数据总量
- Integer count = customMapper.countBySQL(api, "where ShareType = 'V' and new_id = '0'");
+ Integer count = customMapper.countBySQL(api, "where ShareType = 'V' ");
//批量插入200一次
int page = count % 200 == 0 ? count / 200 : (count / 200) + 1;
for (int i = 0; i < page; i++) {
@@ -2271,6 +2292,89 @@ public class DataImportNewServiceImpl implements DataImportNewService {
}
+ /**
+ * 删除目标ORG中的数据
+ * @param param 参数
+ * @return 执行结果
+ * @throws Exception 异常
+ */
+ @Override
+ public ReturnT deleteTargetOrgData(SalesforceParam param) throws Exception {
+ log.info("deleteTargetOrgData execute start ..................");
+
+ QueryWrapper wrapper = new QueryWrapper<>();
+ if (StringUtils.isNotBlank(param.getApi())) {
+ wrapper.in("name", DataUtil.toIdList(param.getApi()));
+ } else {
+ wrapper.eq("data_work", 1);
+ }
+ List objectList = dataObjectService.list(wrapper);
+ if (objectList.isEmpty()) {
+ log.info("没有对象需要删除数据！!!");
+ return ReturnT.FAIL;
+ }
+
+ PartnerConnection connect = salesforceTargetConnect.createConnect();
+
+ try {
+ for (DataObject dataObject : objectList) {
+
+ String api = dataObject.getName();
+
+ boolean moreData = true;
+
+ DescribeSObjectResult dsr = connect.describeSObject(api);
+
+ Field[] dsrFields = dsr.getFields();
+
+ String sql = "select Id from " + api + " Limit 200";
+
+ log.info("执行SQL：{}", sql);
+
+ while (moreData){
+ int size = 0;
+
+ QueryResult queryResult = connect.queryAll(sql);
+
+ if (ObjectUtils.isEmpty(queryResult) || ObjectUtils.isEmpty(queryResult.getRecords())) {
+ break;
+ }
+ SObject[] records = queryResult.getRecords();
+
+ com.alibaba.fastjson2.JSONArray objects = DataUtil.toJsonArray(records, dsrFields);
+ String[] ids = new String[objects.size()];
+ for (int i = 0; i < objects.size(); i++) {
+ JSONObject jsonObject = objects.getJSONObject(i);
+ String id = jsonObject.getString("Id");
+ ids[i] = id;
+ size ++;
+ }
+ DeleteResult[] deleteResults = connect.delete(ids);
+ for (DeleteResult deleteResult : deleteResults) {
+ if (deleteResult.isSuccess()) {
+ log.info("成功删除目标ORG中的数据，ID: {}，类型: {}", deleteResult.getId(), dataObject.getLabel());
+ } else {
+ log.error("ID:{}删除目标ORG数据失败：{}", deleteResult.getId(), deleteResult.getErrors());
+ }
+ }
+ if (size < 200){
+ moreData = false;
+ }
+ }
+ }
+ TimeUnit.MILLISECONDS.sleep(1);
+
+ }catch (InterruptedException interruptedException){
+ return ReturnT.FAIL;
+ }catch (Exception e){
+ log.info("目标ORG数据删除失败！", e);
+ }
+
+ log.info("目标ORG数据删除完成！");
+
+ return ReturnT.SUCCESS;
+ }
+
private ReturnT manualCheckDeletedData(SalesforceParam param) throws Exception {
String api = "DeleteEvent";
@@ -2505,15 +2609,8 @@ public class DataImportNewServiceImpl implements DataImportNewService {
@Override
public void getNewIdByField(String api, String queryFields) throws Exception {
List stringList = DataUtil.toIdList(queryFields);
- // 检测表 不存在就生成 但不生成批次
- commonService.checkApi(api, true);
- Map map = Maps.newHashMap();
- SalesforceParam salesforceParam = new SalesforceParam();
- salesforceParam.setApi(api);
- salesforceParam.setIdField("Id");
String maxId = null;
Field[] dsrFields;
-
List fields = Lists.newArrayList();
PartnerConnection connect = salesforceConnect.createConnect();
@@ -2526,178 +2623,190 @@ public class DataImportNewServiceImpl implements DataImportNewService {
}
fields.add(field.getName());
}
- if ("TaskRelation".equals(api) || "EventRelation".equals(api)){
+ // 添加特定对象的关联字段
+ addRelationFields(api, fields);
+
+ String joined = StringUtils.join(fields, ",");
+ com.alibaba.fastjson2.JSONArray objects;
+ int count = 0;
+
+ do {
+ String sql = null;
+ if (StringUtils.isNotEmpty(maxId)){
+ sql = "select " + joined + " from "+ api + " Where Id > '"+ maxId+"' LIMIT 200";
+ }else {
+ sql = "select " + joined + " from "+ api +" LIMIT 200";
+ }
+ QueryResult queryResult = connect.queryAll(sql);
+ if (ObjectUtils.isEmpty(queryResult) || ObjectUtils.isEmpty(queryResult.getRecords())) {
+ break;
+ }
+
+ SObject[] records = queryResult.getRecords();
+ objects = DataUtil.toJsonArray(records, dsrFields);
+ maxId = ((JSONObject) objects.get(objects.size() - 1)).getString(Const.ID);
+
+ // 批量处理记录
+ processRecords(api, objects, stringList);
+
+ int totalRecords = records.length;
+
+ count += totalRecords;
+ log.info("getNewId success count: {}", count);
+
+ if (totalRecords < 200) {
+ log.info("无更多数据，反写NewId任务结束！！！");
+ return;
+ }
+
+ TimeUnit.MILLISECONDS.sleep(1);
+ } while (true);
+ }
+
+ private void addRelationFields(String api, List fields) {
+ if ("TaskRelation".equals(api) || "EventRelation".equals(api)) {
fields.add("Relation.type");
}
- if ("CollaborationGroupMember".equals(api) || "MemberId".equals(api)){
+ if ("CollaborationGroupMember".equals(api)) {
fields.add("Member.type");
}
- if ("FeedAttachment".equals(api) || "FeedEntityId".equals(api)){
+ if ("FeedAttachment".equals(api)) {
fields.add("FeedEntity.type");
- }
- if ("FeedAttachment".equals(api) || "RecordId".equals(api)){
fields.add("Record.type");
}
- salesforceParam.setSelect(StringUtils.join(fields, ","));
- com.alibaba.fastjson2.JSONArray objects = null;
- int count = 0, failCount = 0;
- // 遍历
- while (true) {
- try {
- // 判断是否存在要排除的id
- salesforceParam.setMaxId(maxId);
- map.put("param", salesforceParam);
- String sql = SqlUtil.showSql("com.celnet.datadump.mapper.SalesforceMapper.listOrderById", map);
- log.info("query sql: {}", sql);
- XxlJobLogger.log("query sql: {}", sql);
- QueryResult queryResult = connect.queryAll(sql);
- if (ObjectUtils.isEmpty(queryResult) || ObjectUtils.isEmpty(queryResult.getRecords())) {
- break;
- }
- SObject[] records = queryResult.getRecords();
- objects = DataUtil.toJsonArray(records, dsrFields);
- maxId = ((JSONObject) objects.get(objects.size() - 1)).getString(Const.ID);
+ }
- // 更新new_id
- for (int i = 0; i < objects.size(); i++) {
- JSONObject jsonObject = objects.getJSONObject(i);
- try {
- Set keys = jsonObject.keySet();
- // update
- String id = jsonObject.getString(Const.ID);
- List> maps = Lists.newArrayList();
- for (String key : keys) {
- if ("AccountContactRelation".equals(api) && "ContactId".equals(key)) {
- Map m = customMapper.getByNewId("Id", "Contact", jsonObject.getString(key));
- if (m == null){
- break;
- }
- Map paramMap = Maps.newHashMap();
- paramMap.put("key", key);
- paramMap.put("value",m.get("Id"));
- maps.add(paramMap);
- continue;
- }
- if ("AccountContactRelation".equals(api) && "AccountId".equals(key)) {
- Map m = customMapper.getByNewId("Id", "Account", jsonObject.getString(key));
- if (m == null){
- break;
- }
- Map paramMap = Maps.newHashMap();
- paramMap.put("key", key);
- paramMap.put("value",m.get("Id"));
- maps.add(paramMap);
- continue;
- }
- if ("TaskRelation".equals(api) && "TaskId".equals(key)) {
- Map m = customMapper.getByNewId("Id", "Task", jsonObject.getString(key));
- if (m == null){
- break;
- }
- Map paramMap = Maps.newHashMap();
- paramMap.put("key", key);
- paramMap.put("value",m.get("Id"));
- maps.add(paramMap);
- continue;
- }
- if ("EventRelation".equals(api) && "EventId".equals(key)) {
- Map m = customMapper.getByNewId("Id", "Event", jsonObject.getString(key));
- if (m == null){
- break;
- }
- Map paramMap = Maps.newHashMap();
- paramMap.put("key", key);
- paramMap.put("value",m.get("Id"));
- maps.add(paramMap);
- continue;
- }
- if (("EventRelation".equals(api) || "TaskRelation".equals(api)) && "RelationId".equals(key)) {
- Map m = customMapper.getByNewId("Id", jsonObject.getString("Relation_Type"), jsonObject.getString(key));
- if (m == null){
- break;
- }
- Map paramMap = Maps.newHashMap();
- paramMap.put("key", key);
- paramMap.put("value",m.get("Id"));
- maps.add(paramMap);
- continue;
- }
- if ("CollaborationGroupMember".equals(api) && "CollaborationGroupId".equals(key)) {
- Map m = customMapper.getByNewId("Id", "CollaborationGroup", jsonObject.getString(key));
- if (m == null){
- break;
- }
- Map paramMap = Maps.newHashMap();
- paramMap.put("key", key);
- paramMap.put("value",m.get("Id"));
- maps.add(paramMap);
- continue;
- }
- if ("CollaborationGroupMember".equals(api) && "MemberId".equals(key)) {
- Map m = customMapper.getByNewId("Id", jsonObject.getString("Member_Type"), jsonObject.getString(key));
- if (m == null){
- break;
- }
- Map paramMap = Maps.newHashMap();
- paramMap.put("key", key);
- paramMap.put("value",m.get("Id"));
- maps.add(paramMap);
- continue;
- }
- if ("FeedAttachment".equals(api) && "FeedEntityId".equals(key)) {
- Map m = customMapper.getByNewId("Id", jsonObject.getString("FeedEntity_Type"), jsonObject.getString(key));
- if (m == null){
- break;
- }
- Map paramMap = Maps.newHashMap();
- paramMap.put("key", key);
- paramMap.put("value",m.get("Id"));
- maps.add(paramMap);
- continue;
- }
- if ("FeedAttachment".equals(api) && "RecordId".equals(key)) {
- Map m = customMapper.getByNewId("Id", jsonObject.getString("RecordId_Type"), jsonObject.getString(key));
- if (m == null){
- break;
- }
- Map paramMap = Maps.newHashMap();
- paramMap.put("key", key);
- paramMap.put("value",m.get("Id"));
- maps.add(paramMap);
- continue;
- }
- if (stringList.contains(key)) {
- Map paramMap = Maps.newHashMap();
- paramMap.put("key", key);
- paramMap.put("value", jsonObject.getString(key));
- maps.add(paramMap);
- }
- }
- customMapper.updateByNewId(maps, api, id);
- } catch (Throwable throwable) {
- if (throwable.toString().contains("interrupt")) {
- log.error("may interrupt error:", throwable);
- throw new InterruptedException();
- }
- log.info(JSON.toJSONString(throwable));
- }
- }
- count += records.length;
- log.info("getNewId success count: {}", count);
- failCount = 0;
- objects = null;
- } catch (InterruptedException e) {
- throw e;
- } catch (Throwable throwable) {
- failCount++;
- log.error("getNewId error api:{}, data:{}", api, com.alibaba.fastjson2.JSON.toJSONString(objects), throwable);
- if (failCount > Const.MAX_FAIL_COUNT) {
- log.info("getNewId error data:" + com.alibaba.fastjson2.JSON.toJSONString(objects));
- }
+ private String buildQuerySql(String api, String joined, String maxId) {
+ StringBuilder sqlBuilder = new StringBuilder("SELECT ");
+ if (StringUtils.isNotEmpty(joined)) {
+ sqlBuilder.append(joined).append(", ");
+ }
+ sqlBuilder.append("Id FROM ").append(api);
+ if (StringUtils.isNotEmpty(maxId)) {
+ sqlBuilder.append(" WHERE Id > '").append(maxId).append("'");
+ }
+ sqlBuilder.append(" LIMIT 200");
+ return sqlBuilder.toString();
+ }
+
+ private void processRecords(String api, com.alibaba.fastjson2.JSONArray objects, List stringList) throws Exception {
+ for (int i = 0; i < objects.size(); i++) {
+ JSONObject jsonObject = objects.getJSONObject(i);
+ String id = jsonObject.getString(Const.ID);
+ List> maps = buildUpdateMaps(api, jsonObject, stringList);
+
+ if (maps!=null && !maps.isEmpty()) {
+ customMapper.updateNewIdByFields(api, maps, id);
}
}
+ }
- // 3、拼接sql，更新本地数据库new_id
+ private List> buildUpdateMaps(String api, JSONObject jsonObject, List stringList) throws Exception {
+ List> maps = Lists.newArrayList();
+ Set keys = jsonObject.keySet();
+
+ for (String key : keys) {
+ Map paramMap = null;
+
+ switch (api) {
+ case "AccountContactRelation":
+ paramMap = handleAccountContactRelation(key, jsonObject);
+ if (paramMap == null){
+ return null;
+ }
+ break;
+ case "TaskRelation":
+ case "EventRelation":
+ paramMap = handleTaskEventRelation(api, key, jsonObject);
+ if (paramMap == null){
+ return null;
+ }
+ break;
+ case "CollaborationGroupMember":
+ paramMap = handleCollaborationGroupMember(key, jsonObject);
+ if (paramMap == null){
+ return null;
+ }
+ break;
+ case "FeedAttachment":
+ paramMap = handleFeedAttachment(key, jsonObject);
+ if (paramMap == null){
+ return null;
+ }
+ break;
+ default:
+ if (stringList.contains(key)) {
+ paramMap = Maps.newHashMap();
+ paramMap.put("key", key);
+ paramMap.put("value", jsonObject.getString(key));
+ }
+ break;
+ }
+
+ if (paramMap != null) {
+ maps.add(paramMap);
+ }
+ }
+
+ return maps;
+ }
+
+ private Map handleAccountContactRelation(String key, JSONObject jsonObject) throws Exception {
+ switch (key) {
+ case "ContactId":
+ return getNewIdMap(key, "Contact", jsonObject.getString(key));
+ case "AccountId":
+ return getNewIdMap(key, "Account", jsonObject.getString(key));
+ default:
+ return null;
+ }
+ }
+
+ private Map handleTaskEventRelation(String api, String key, JSONObject jsonObject) throws Exception {
+ switch (key) {
+ case "TaskId":
+ return getNewIdMap(key, "Task", jsonObject.getString(key));
+ case "EventId":
+ return getNewIdMap(key, "Event", jsonObject.getString(key));
+ case "RelationId":
+ return getNewIdMap(key, jsonObject.getString("Relation_Type"), jsonObject.getString(key));
+ default:
+ return null;
+ }
+ }
+
+ private Map handleCollaborationGroupMember(String key, JSONObject jsonObject) throws Exception {
+ switch (key) {
+ case "CollaborationGroupId":
+ return getNewIdMap(key, "CollaborationGroup", jsonObject.getString(key));
+ case "MemberId":
+ return getNewIdMap(key, jsonObject.getString("Member_Type"), jsonObject.getString(key));
+ default:
+ return null;
+ }
+ }
+
+ private Map handleFeedAttachment(String key, JSONObject jsonObject) throws Exception {
+ switch (key) {
+ case "FeedEntityId":
+ return getNewIdMap(key, jsonObject.getString("FeedEntity_Type"), jsonObject.getString(key));
+ case "RecordId":
+ return getNewIdMap(key, jsonObject.getString("RecordId_Type"), jsonObject.getString(key));
+ default:
+ return null;
+ }
+ }
+
+ private Map getNewIdMap(String key, String tableName, String id) throws Exception {
+ Map m = customMapper.getByNewId("Id", tableName, id);
+ if (m == null || m.get("Id") == null) {
+ log.info("关联对象未找到，跳过处理。字段: {}, 表名: {}, ID: {}", key, tableName, id);
+ return null; // 返回null表示跳过该字段
+ }
+ Map paramMap = Maps.newHashMap();
+ paramMap.put("key", key);
+ paramMap.put("value", m.get("Id"));
+ return paramMap;
}
@Override
diff --git a/src/main/java/com/celnet/datadump/service/impl/DataImportServiceImpl.java b/src/main/java/com/celnet/datadump/service/impl/DataImportServiceImpl.java
index a1de704..6cdbcbe 100644
--- a/src/main/java/com/celnet/datadump/service/impl/DataImportServiceImpl.java
+++ b/src/main/java/com/celnet/datadump/service/impl/DataImportServiceImpl.java
@@ -386,6 +386,7 @@ public class DataImportServiceImpl implements DataImportService {
//如果必填lookup字段没有值，跳过
dataObject.setDataIndex(Integer.parseInt(map.get("data_index").toString()) + 1);
dataObjectService.updateById(dataObject);
+ log.info("引用对象：" + reference + "不存在数据！请检查这个对象详情！！！");
return;
}else{
account.setField(dataField.getField(), referenceMap.get(0).get("new_id"));
@@ -455,12 +456,15 @@ public class DataImportServiceImpl implements DataImportService {
dataLogService.save(dataLog2);
TimeUnit.MILLISECONDS.sleep(1);
+ } catch (InterruptedException interruptedException){
+ return;
} catch (Exception e) {
- log.info(JSON.toJSONString(e));
+ log.info(JSON.toJSONString(e.getMessage()));
}
}
- if (z*10000+size >= count){
+ if (z * 10000 + size >= count) {
+ log.info("数据处理完成，共处理 {} 条记录，当前批次结束", count);
break;
}
}
diff --git a/src/main/java/com/celnet/datadump/service/impl/DataVerifyServiceImpl.java b/src/main/java/com/celnet/datadump/service/impl/DataVerifyServiceImpl.java
index b2b905f..d6a4635 100644
--- a/src/main/java/com/celnet/datadump/service/impl/DataVerifyServiceImpl.java
+++ b/src/main/java/com/celnet/datadump/service/impl/DataVerifyServiceImpl.java
@@ -629,7 +629,7 @@ public class DataVerifyServiceImpl implements DataVerifyService {
sourceRow.add(sourceField.isNillable());
sourceRow.add(sourceField.isUnique());
sourceRow.add(sourceField.isPolymorphicForeignKey() ? "是" : "否");
- sourceRow.add(sourceField.getReferenceTo().length> 2 ? "多态字段" : String.join(",", sourceField.getReferenceTo()));
+ sourceRow.add((sourceField.getReferenceTo()!=null && sourceField.getReferenceTo().length> 2) ? "多态字段" : String.join(",", sourceField.getReferenceTo()));
// 写入Excel
excelWriter.write(Collections.singletonList(sourceRow), fieldTypeSheet);
@@ -652,7 +652,7 @@ public class DataVerifyServiceImpl implements DataVerifyService {
targetRow.add(targetField.isNillable());
targetRow.add(targetField.isUnique());
targetRow.add(targetField.isPolymorphicForeignKey() ? "是" : "否");
- targetRow.add(targetField.getReferenceTo().length> 2 ? "多态字段" : String.join(",", sourceField.getReferenceTo()));
+ targetRow.add((targetField.getReferenceTo()!=null && targetField.getReferenceTo().length> 2 )? "多态字段" : String.join(",", sourceField.getReferenceTo()));
// 写入Excel
excelWriter.write(Collections.singletonList(targetRow), fieldTypeSheet);
@@ -672,14 +672,13 @@ public class DataVerifyServiceImpl implements DataVerifyService {
PartnerConnection sourceConn, PartnerConnection targetConn,Map> fieldMap) {
String startDate = DateFormatUtils.format(batch.getSyncStartDate(), "yyyy-MM-dd HH:mm:ss");
String endDate = DateFormatUtils.format(batch.getSyncEndDate(), "yyyy-MM-dd HH:mm:ss");
- log.info("开始处理数据批次: 批次ID={} 批次开始时间={} 批次结束时间={} 对象API={}", batch.getId(),startDate,endDate, objectApi);
+ log.info("开始处理数据批次: 批次ID={} 批次总量(大于十万只校验十万)={} 批次开始时间={} 批次结束时间={} 对象API={}", batch.getId(),batch.getFirstSfNum(),startDate,endDate, objectApi);
// 分页处理记录
int page = 0;
int totalRecordNum = 0;
int targetRecordNum = 0;
int sourceRecordNum = 0;
- int errorFieldNum = 0;
int sorceOnlyNum = 0;
int targetOnlyNum = 0;
boolean isError = false;
@@ -700,7 +699,6 @@ public class DataVerifyServiceImpl implements DataVerifyService {
List> idList = recordIds.subList(startIndex, endIndex);
totalRecordNum += idList.size();
- log.info("数据总量: {} 子批次: 批次数={}, 记录数={}, 累计记录数={}",batch.getFirstDbNum(), (page*50 + i), idList.size(), totalRecordNum);
List ids = Lists.newArrayList();
List newIds = Lists.newArrayList();
@@ -714,14 +712,11 @@ public class DataVerifyServiceImpl implements DataVerifyService {
// 批量查询源和目标数据
Map sourceRecords = queryRecords(sourceConn, objectApi, ids);
sourceRecordNum += sourceRecords.size();
- log.info("源系统查询完成，获取到{}条记录", sourceRecords.size());
Map targetRecords = queryRecords(targetConn, objectApi, newIds);
targetRecordNum += targetRecords.size();
- log.info("目标系统查询完成，获取到{}条记录", targetRecords.size());
// 比对记录
- log.info("开始比对{}条记录的字段值", idLinks.size());
for (String id : idLinks.keySet()) {
String newId = idLinks.get(id);
SObject sourceRecord = sourceRecords.get(id);
diff --git a/src/main/java/com/celnet/datadump/service/impl/FileServiceImpl.java b/src/main/java/com/celnet/datadump/service/impl/FileServiceImpl.java
index 2cd789b..1dec4c0 100644
--- a/src/main/java/com/celnet/datadump/service/impl/FileServiceImpl.java
+++ b/src/main/java/com/celnet/datadump/service/impl/FileServiceImpl.java
@@ -693,8 +693,17 @@ public class FileServiceImpl implements FileService {
qw.eq("name", parentType);
List objects = dataObjectService.list(qw);
if (objects.isEmpty()) {
- log.info( "文件ID：" + id + "关联对象不存在： {}", parentType);
- EmailUtil.send("DataDump ERROR", "文件ID：" + id + "关联对象"+ parentType+"不存在");
+ String format = "文件ID：" + id + "关联对象不存在：" + parentType;
+ log.info(format);
+ Map paramMap = Maps.newHashMap();
+ paramMap.put("key", "is_upload");
+ paramMap.put("value", 2);
+ maps.add(paramMap);
+ Map paramMap1 = Maps.newHashMap();
+ paramMap1.put("key", "error_message");
+ paramMap1.put("value",format);
+ maps.add(paramMap1);
+ customMapper.updateById(api, maps, id);
break;
}
@@ -702,7 +711,15 @@ public class FileServiceImpl implements FileService {
if(lMap == null) {
String format = "文件ID：" + id + ",对应关联对象类型：" + parentType + "，对应关联对象ID：" + parentId + "数据不存在！！！！";
log.info(format);
- EmailUtil.send("DataDump ERROR", format);
+ Map paramMap = Maps.newHashMap();
+ paramMap.put("key", "is_upload");
+ paramMap.put("value", 2);
+ maps.add(paramMap);
+ Map paramMap1 = Maps.newHashMap();
+ paramMap1.put("key", "error_message");
+ paramMap1.put("value",format);
+ maps.add(paramMap1);
+ customMapper.updateById(api, maps, id);
break;
}
@@ -710,7 +727,15 @@ public class FileServiceImpl implements FileService {
if(uMap == null) {
String format = "文件ID：" + id + ",对应用户ID：" + ownerId + "数据不存在！！！！";
log.info(format);
- EmailUtil.send("DataDump ERROR", format);
+ Map paramMap = Maps.newHashMap();
+ paramMap.put("key", "is_upload");
+ paramMap.put("value", 2);
+ maps.add(paramMap);
+ Map paramMap1 = Maps.newHashMap();
+ paramMap1.put("key", "error_message");
+ paramMap1.put("value",format);
+ maps.add(paramMap1);
+ customMapper.updateById(api, maps, id);
break;
}
@@ -718,7 +743,15 @@ public class FileServiceImpl implements FileService {
if(cMap == null) {
String format = "文件ID：" + id + ",对应用户ID：" + createdById + "数据不存在！！！！";
log.info(format);
- EmailUtil.send("DataDump ERROR", format);
+ Map paramMap = Maps.newHashMap();
+ paramMap.put("key", "is_upload");
+ paramMap.put("value", 2);
+ maps.add(paramMap);
+ Map paramMap1 = Maps.newHashMap();
+ paramMap1.put("key", "error_message");
+ paramMap1.put("value",format);
+ maps.add(paramMap1);
+ customMapper.updateById(api, maps, id);
break;
}
@@ -755,7 +788,18 @@ public class FileServiceImpl implements FileService {
maps.add(paramMap);
}
} else {
- log.info("文件ID：" + id + "上传失败，返回信息：" + EntityUtils.toString(response.getEntity(), "UTF-8"));
+ {
+ Map paramMap = Maps.newHashMap();
+ paramMap.put("key", "is_upload");
+ paramMap.put("value", 2);
+ maps.add(paramMap);
+ Map paramMap1 = Maps.newHashMap();
+ paramMap1.put("key", "error_message");
+ paramMap1.put("value", response == null?"上传失败" :EntityUtils.toString(response.getEntity(), "UTF-8"));
+ maps.add(paramMap1);
+ }
+ customMapper.updateById(api, maps, id);
+ log.info("---------文件ID：" + id + "上传失败--------" );
throw new RuntimeException();
}
}
@@ -770,12 +814,10 @@ public class FileServiceImpl implements FileService {
log.error("upload file error, id: {}", id, throwable);
failCount++;
if (Const.MAX_FAIL_COUNT < failCount) {
- {
- Map paramMap = Maps.newHashMap();
- paramMap.put("key", "is_upload");
- paramMap.put("value", 2);
- maps.add(paramMap);
- }
+ Map paramMap = Maps.newHashMap();
+ paramMap.put("key", "is_upload");
+ paramMap.put("value", 2);
+ maps.add(paramMap);
customMapper.updateById(api, maps, id);
break;
}
diff --git a/src/main/java/com/celnet/datadump/service/impl/OrgConfigServiceImpl.java b/src/main/java/com/celnet/datadump/service/impl/OrgConfigServiceImpl.java
index 23af364..d9e76a4 100644
--- a/src/main/java/com/celnet/datadump/service/impl/OrgConfigServiceImpl.java
+++ b/src/main/java/com/celnet/datadump/service/impl/OrgConfigServiceImpl.java
@@ -23,6 +23,8 @@ import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import static com.celnet.datadump.global.SystemConfigCode.EXECUTOR_SIZE;
+
/**
* org配置表 服务实现类
*/
@@ -73,7 +75,7 @@ public class OrgConfigServiceImpl extends ServiceImpl> list = customerMapper.list("code,value","system_config","code ='DEFAULT_START_DATE'");
+ List> list1 = customerMapper.list("code,value","system_config","code ='BATCH_TYPE'");
+
+ String format = String.format("源ORG与目标ORG连接配置成功！ \n源ORG地址:\n%s , \n目标ORG地址:\n%s \n---请检查文件上传下载地址---\n 文件上传地址：\n%s \n文件下载地址：\n%s \n---请检查批次开始时间（不对要重启服务）---\n%s \n---请检查批次类型（不对要重启服务）---\n%s",
+ sourceOrgUrl,targetOrgUrl,uploadUrl,dumpUrl,list.get(0).get("value"),list1.get(0).get("value"));
+ EmailUtil.send("校验ORG配置", format);
}
}
diff --git a/src/main/resources/mapper/CustomMapper.xml b/src/main/resources/mapper/CustomMapper.xml
index 626f1df..104f13b 100644
--- a/src/main/resources/mapper/CustomMapper.xml
+++ b/src/main/resources/mapper/CustomMapper.xml
@@ -138,6 +138,16 @@
new_id = #{newId}
+
+ UPDATE `${tableName}`
+ SET new_id = #{newId}
+
+
+ ${map.key} = #{map.value}
+
+
+
+
insert into
`${tableName}`
diff --git a/src/main/resources/mapper/SalesforceMapper.xml b/src/main/resources/mapper/SalesforceMapper.xml
index 687e4be..45ad41a 100644
--- a/src/main/resources/mapper/SalesforceMapper.xml
+++ b/src/main/resources/mapper/SalesforceMapper.xml
@@ -167,6 +167,37 @@
+
+