kris/datai
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

【feat】 初步实现数据同步拉取

Browse Source
This commit is contained in:
Kris 2026-01-07 13:52:00 +08:00
parent cb09602386
commit 4f0a941ab6
8 changed files with 399 additions and 98 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
datai-admin/src/main/resources/application-plugins.yml
View File

@ -3,13 +3,13 @@ spring:
# redis 配置
redis:
# 地址
host: 134.175.52.143
host: 127.0.0.1
# 端口，默认为6379
port: 6379
# 数据库索引
database: 0
# 密码
password: czsj@2024
password:
# 连接超时时间
timeout: 10s
lettuce:

31
datai-scenes/datai-scene-salesforce/datai-salesforce-common/src/main/java/com/datai/salesforce/common/utils/SoqlBuilder.java
View File

@ -611,6 +611,37 @@ public class SoqlBuilder {
return soql.toString();
}
/**
* 构建COUNT查询语句
* 生成类似 "SELECT count(Id) FROM Object WHERE ..." 的查询
*
* 注意：
* - COUNT查询不使用SELECT子句中指定的字段，而是使用count(Id)
* - COUNT查询不支持GROUP BY、HAVING、ORDER BY、LIMIT、OFFSET
* - 只使用FROM和WHERE条件
*
* @return COUNT查询语句
* @throws IllegalStateException 如果FROM子句为空
*/
public String buildCountQuery() {
if (fromObject == null || fromObject.trim().isEmpty()) {
throw new IllegalStateException("FROM子句不能为空，必须指定查询对象");
}
StringBuilder soql = new StringBuilder();
// 构建SELECT count(Id)子句
soql.append("SELECT count(Id)");
// 构建FROM子句
soql.append(" FROM ").append(fromObject);
// 构建WHERE子句
buildWhereClause(soql);
return soql.toString();
}
/**
* 验证构建器配置
*

43
datai-scenes/datai-scene-salesforce/datai-salesforce-integration/src/main/java/com/datai/integration/controller/DataiIntegrationObjectController.java
View File

@ -258,4 +258,47 @@ public class DataiIntegrationObjectController extends BaseController
return error("同步对象数据时发生异常: " + e.getMessage());
}
}
/**
* 同步多个对象数据到本地数据库
*
* 该方法用于触发多个对象的批量数据同步操作，将多个Salesforce对象的数据同步到本地数据库
* 同步操作包括全量同步和增量同步两种模式，具体模式由每个对象的配置决定
* 该方法会依次同步每个对象，即使某个对象同步失败，也会继续同步其他对象
*
* @param ids 对象ID数组，用于标识需要同步的多个Salesforce对象
* @return AjaxResult 同步结果，包含成功/失败状态、每个对象的同步结果、总耗时等信息
* - 成功时返回：success=true, message="多对象数据同步完成", 以及详细的同步信息
* - 失败时返回：success=false, message=错误信息
*/
@Operation(summary = "同步多个对象数据到本地数据库")
@PreAuthorize("@ss.hasPermi('integration:object:syncData')")
@Log(title = "对象同步控制", businessType = BusinessType.UPDATE)
@PostMapping("/syncMultipleData")
public AjaxResult syncMultipleObjectData(@RequestBody Integer[] ids)
{
if (ids == null || ids.length == 0) {
log.error("对象ID数组为空，无法同步数据");
return error("对象ID不能为空");
}
try {
log.info("开始同步多个对象数据，对象ID数量: {}", ids.length);
Map<String, Object> result = dataiIntegrationObjectService.syncMultipleObjectData(ids);
if ((Boolean) result.get("success")) {
log.info("多对象数据同步完成，成功数量: {}, 失败数量: {}",
result.get("successCount"), result.get("failureCount"));
return success(result);
} else {
log.error("多对象数据同步失败，错误信息: {}", result.get("message"));
return error((String) result.get("message"));
}
} catch (Exception e) {
log.error("同步多个对象数据时发生异常", e);
return error("同步多个对象数据时发生异常: " + e.getMessage());
}
}
}

19
datai-scenes/datai-scene-salesforce/datai-salesforce-integration/src/main/java/com/datai/integration/model/domain/DataiIntegrationBatch.java
View File

@ -11,6 +11,25 @@ import com.datai.common.core.domain.BaseEntity;
/**
* 数据批次对象 datai_integration_batch
*
* 字段更新规则：
* 1. 创建批次时设置的字段（后续不允许修改）：
* - id: 批次ID
* - deptId: 部门ID
* - api: 对象API
* - label: 对象名称
* - syncType: 同步类型（FULL/INCREMENTAL）
* - batchField: 批次字段
* - syncStartDate: 开始同步时间（批次的时间范围起点）
* - syncEndDate: 结束同步时间（批次的时间范围终点）
*
* 2. 同步时更新的字段：
* - sfNum: SF数据量（每次同步时更新）
* - dbNum: 本地数据量（每次同步时更新）
* - syncStatus: 同步状态（true=成功，false=失败）
* - firstSyncTime: 首次同步时间（仅在首次同步时设置）
* - lastSyncTime: 最后同步时间（每次同步时更新）
* - updateTime: 更新时间（每次更新时自动设置）
*
* @author datai
* @date 2026-01-01
*/

7
datai-scenes/datai-scene-salesforce/datai-salesforce-integration/src/main/java/com/datai/integration/model/param/DataiSyncParam.java
View File

@ -44,6 +44,11 @@ public class DataiSyncParam {
* 开始修改时间
*/
private Date beginModifyDate;
/**
* 数据拉取截断时间
*/
private Date lastDate;
/**
* 结束修改时间
*/
@ -69,7 +74,7 @@ public class DataiSyncParam {
*/
private String select;
/**
* 是否查询删除记录
* 是否存在IsDeleted删除字段
*/
private Boolean isDeleted;

8
datai-scenes/datai-scene-salesforce/datai-salesforce-integration/src/main/java/com/datai/integration/service/IDataiIntegrationObjectService.java
View File

@ -116,4 +116,12 @@ public interface IDataiIntegrationObjectService
* @return 同步结果
*/
public Map<String, Object> syncSingleObjectData(Integer id);
/**
* 同步多个对象数据到本地数据库
*
* @param ids 对象ID集合
* @return 同步结果
*/
public Map<String, Object> syncMultipleObjectData(Integer[] ids);
}

260
datai-scenes/datai-scene-salesforce/datai-salesforce-integration/src/main/java/com/datai/integration/service/impl/DataiIntegrationBatchServiceImpl.java
View File

@ -175,7 +175,7 @@ public class DataiIntegrationBatchServiceImpl implements IDataiIntegrationBatchS
log.info("批次 {} 共有 {} 条失败的同步记录，准备重试", id, failedHistories.size());
boolean retryResult = syncObjectDataByBatch(batch.getApi(), id.toString());
boolean retryResult = syncObjectDataByBatch(batch.getApi(), id);
if (retryResult) {
log.info("批次 {} 重试成功", id);
@ -299,12 +299,13 @@ public class DataiIntegrationBatchServiceImpl implements IDataiIntegrationBatchS
}
String objectApi = batch.getApi();
String batchId = id.toString();
log.info("准备同步对象 {} 的批次 {} 数据", objectApi, batchId);
log.info("准备同步对象 {} 的批次 {} 数据", objectApi, id);
boolean syncResult = syncObjectDataByBatch(objectApi, batchId);
boolean syncResult = syncObjectDataByBatch(objectApi, id);
batch = selectDataiIntegrationBatchById(id);
if (syncResult) {
log.info("批次 {} 数据同步成功", id);
result.put("success", true);
@ -312,12 +313,14 @@ public class DataiIntegrationBatchServiceImpl implements IDataiIntegrationBatchS
result.put("batchId", id);
result.put("api", objectApi);
result.put("label", batch.getLabel());
result.put("syncNum", batch.getDbNum());
} else {
log.error("批次 {} 数据同步失败", id);
result.put("success", false);
result.put("message", "批次数据同步失败");
result.put("batchId", id);
result.put("api", objectApi);
result.put("syncNum", batch.getDbNum() != null ? batch.getDbNum() : 0);
}
} catch (Exception e) {
@ -336,10 +339,10 @@ public class DataiIntegrationBatchServiceImpl implements IDataiIntegrationBatchS
* @param batchId 批次ID
* @return 同步结果
*/
private boolean syncObjectDataByBatch(String objectApi, String batchId) {
private boolean syncObjectDataByBatch(String objectApi, Integer batchId) {
log.info("准备同步Salesforce对象的指定批次数据，对象API: {}, 批次ID: {}", objectApi, batchId);
if (objectApi == null || objectApi.trim().isEmpty() || batchId == null || batchId.trim().isEmpty()) {
if (objectApi == null || objectApi.trim().isEmpty() || batchId == null ) {
log.error("对象API或批次ID为空，无法同步指定批次数据");
return false;
}
@ -350,7 +353,7 @@ public class DataiIntegrationBatchServiceImpl implements IDataiIntegrationBatchS
LocalDateTime syncStartTime = LocalDateTime.now();
try {
batch = selectDataiIntegrationBatchById(Integer.parseInt(batchId));
batch = selectDataiIntegrationBatchById(batchId);
if (batch == null) {
log.error("批次不存在，批次ID: {}", batchId);
return false;
@ -370,8 +373,18 @@ public class DataiIntegrationBatchServiceImpl implements IDataiIntegrationBatchS
param.setIsDeleted(true);
}
int totalCount = executeQueryAndProcessData(connection, param, fieldList);
log.info("对象 {} 批次 {} 数据同步完成，共处理 {} 条记录", objectApi, batchId, totalCount);
if (batch.getSyncStartDate() != null) {
param.setBeginDate(java.sql.Timestamp.valueOf(batch.getSyncStartDate()));
}
if (batch.getSyncEndDate() != null) {
param.setEndDate(java.sql.Timestamp.valueOf(batch.getSyncEndDate()));
}
int sfTotalCount = querySalesforceDataCount(connection, param);
log.info("对象 {} 批次 {} Salesforce中共有 {} 条记录需要同步", objectApi, batchId, sfTotalCount);
int dbProcessedCount = executeQueryAndProcessData(connection, param, fieldList);
log.info("对象 {} 批次 {} 数据同步完成，共处理 {} 条记录", objectApi, batchId, dbProcessedCount);
long endTime = System.currentTimeMillis();
long duration = endTime - startTime;
@ -379,9 +392,9 @@ public class DataiIntegrationBatchServiceImpl implements IDataiIntegrationBatchS
batchHistory.setApi(objectApi);
batchHistory.setLabel(batch.getLabel());
batchHistory.setBatchId(Integer.parseInt(batchId));
batchHistory.setBatchId(batchId);
batchHistory.setBatchField(batch.getBatchField());
batchHistory.setSyncNum(totalCount);
batchHistory.setSyncNum(dbProcessedCount);
batchHistory.setSyncType(batch.getSyncType());
batchHistory.setSyncStatus(true);
batchHistory.setStartTime(syncStartTime);
@ -390,9 +403,21 @@ public class DataiIntegrationBatchServiceImpl implements IDataiIntegrationBatchS
batchHistory.setSyncStartTime(syncStartTime);
batchHistory.setSyncEndTime(syncEndTime);
batchHistoryService.insertDataiIntegrationBatchHistory(batchHistory);
log.info("批次 {} 历史记录已保存，同步数据量: {}, 耗时: {}ms", batchId, totalCount, duration);
insertSyncLog(objectApi, batch.getLabel(), batch.getSyncType(), Integer.parseInt(batchId), true, null, duration);
log.info("批次 {} 历史记录已保存，同步数据量: {}, 耗时: {}ms", batchId, dbProcessedCount, duration);
batch.setSfNum(sfTotalCount);
batch.setDbNum(dbProcessedCount);
batch.setSyncStatus(true);
if (batch.getFirstSyncTime() == null) {
batch.setFirstSyncTime(syncStartTime);
}
batch.setLastSyncTime(syncEndTime);
batch.setUpdateTime(DateUtils.getNowDate());
updateDataiIntegrationBatch(batch);
log.info("批次 {} 信息已更新，SF数据量: {}, 本地数据量: {}, 同步状态: {}, 首次同步时间: {}, 最后同步时间: {}",
batchId, sfTotalCount, dbProcessedCount, true, batch.getFirstSyncTime(), syncEndTime);
insertSyncLog(objectApi, batch.getLabel(), batch.getSyncType(), batchId, true, null, duration);
return true;
} catch (Exception e) {
@ -404,7 +429,7 @@ public class DataiIntegrationBatchServiceImpl implements IDataiIntegrationBatchS
batchHistory.setApi(objectApi);
batchHistory.setLabel(batch != null ? batch.getLabel() : "");
batchHistory.setBatchId(Integer.parseInt(batchId));
batchHistory.setBatchId(batchId);
batchHistory.setBatchField(batch != null ? batch.getBatchField() : "");
batchHistory.setSyncNum(0);
batchHistory.setSyncType(batch != null ? batch.getSyncType() : "");
@ -417,7 +442,15 @@ public class DataiIntegrationBatchServiceImpl implements IDataiIntegrationBatchS
batchHistoryService.insertDataiIntegrationBatchHistory(batchHistory);
log.info("批次 {} 失败历史记录已保存，耗时: {}ms", batchId, duration);
insertSyncLog(objectApi, batch != null ? batch.getLabel() : "", batch != null ? batch.getSyncType() : "", Integer.parseInt(batchId), false, e.getMessage(), duration);
if (batch != null) {
batch.setSyncStatus(false);
batch.setLastSyncTime(syncEndTime);
batch.setUpdateTime(DateUtils.getNowDate());
updateDataiIntegrationBatch(batch);
log.info("批次 {} 信息已更新，同步状态: {}, 结束时间: {}", batchId, false, syncEndTime);
}
insertSyncLog(objectApi, batch != null ? batch.getLabel() : "", batch != null ? batch.getSyncType() : "", batchId, false, e.getMessage(), duration);
return false;
}
@ -514,79 +547,48 @@ public class DataiIntegrationBatchServiceImpl implements IDataiIntegrationBatchS
}
/**
* 执行查询并处理结果
* 查询Salesforce数据量
*
* @param connection SOAP连接
* @param param 查询参数
* @param fieldList 字段列表
* @return 处理的数据条数
* @param connection Salesforce连接
* @param param 查询参数
* @return 符合条件的总记录数
*/
private int executeQueryAndProcessData(PartnerConnection connection, DataiSyncParam param, List<String> fieldList) {
private int querySalesforceDataCount(PartnerConnection connection, DataiSyncParam param) {
int totalCount = 0;
JSONArray objects = null;
String maxId = null;
Date lastDate = null;
try {
DescribeSObjectResult describeSObject = connection.describeSObject(param.getApi());
Field[] dsrFields = describeSObject.getFields();
while (true) {
// 获取创建时间
param.setBeginDate(lastDate);
// 判断是否存在要排除的id
param.setMaxId(maxId);
String query = buildDynamicQuery(param);
log.info("执行查询SQL: {}", query);
QueryResult result = connection.queryAll(query);
if (result.getRecords() != null && result.getRecords().length > 0) {
objects = ConvertUtil.toJsonArray(result.getRecords(), dsrFields);
Date maxDate = objects.getJSONObject(objects.size() - 1).getDate(param.getBatchField());
// 在当前批次中找出所有具有相同时间戳的记录，并获取其中ID最大的那条记录的ID
// 这个ID将作为下一批次查询的起始位置，避免重复处理相同时间戳的数据
Optional<String> maxIdOptional = objects.stream()
.map(t -> (JSONObject) t)
.filter(t -> maxDate.equals(t.getDate(param.getBatchField())))
.map(t -> t.getString("Id"))
.max(String::compareTo);
maxId = maxIdOptional.orElse(null);
lastDate = maxDate;
totalCount += processQueryResult(param.getApi(), result.getRecords(), objects);
log.info("已处理 {} 条记录", totalCount);
String countQuery = buildCountQuery(param);
log.info("查询数据量SQL: {}", countQuery);
QueryResult result = connection.queryAll(countQuery);
if (result != null && result.getRecords() != null && result.getRecords().length > 0) {
SObject record = result.getRecords()[0];
Object countValue = record.getField("expr0");
if (countValue != null) {
totalCount = Integer.parseInt(countValue.toString());
}
if (result.isDone()) {
break;
}
result = connection.queryMore(result.getQueryLocator());
TimeUnit.MILLISECONDS.sleep(100);
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
log.error("查询数据时线程被中断，对象API: {}", param.getApi(), e);
log.info("对象 {} 符合条件的总记录数: {}", param.getApi(), totalCount);
} catch (Exception e) {
log.error("查询处理数据时发生异常，对象API: {}", param.getApi(), e);
log.error("查询数据量时发生异常，对象API: {}", param.getApi(), e);
}
return totalCount;
}
/**
* 构建动态查询语句
* 构建COUNT查询语句
*
* @param param 查询参数
* @return SOQL查询语句
* @return SOQL COUNT查询语句
*/
private String buildDynamicQuery(DataiSyncParam param) {
private String buildCountQuery(DataiSyncParam param) {
SoqlBuilder builder = new SoqlBuilder();
builder.from(param.getApi());
String[] selectFields = param.getSelect().split(",");
builder.select(selectFields)
.from(param.getApi());
if (param.getIsDeleted() != null && !param.getIsDeleted()) {
if (param.getIsDeleted() != null && param.getIsDeleted()) {
builder.where("IsDeleted = false");
}
@ -617,6 +619,117 @@ public class DataiIntegrationBatchServiceImpl implements IDataiIntegrationBatchS
builder.whereLe(param.getBatchField(), param.getEndModifyDate());
}
return builder.buildCountQuery();
}
/**
* 执行查询并处理结果
*
* @param connection SOAP连接
* @param param 查询参数
* @param fieldList 字段列表
* @return 处理的数据条数
*/
private int executeQueryAndProcessData(PartnerConnection connection, DataiSyncParam param, List<String> fieldList) {
int totalCount = 0;
JSONArray objects = null;
String maxId = null;
Date lastDate = null;
try {
DescribeSObjectResult describeSObject = connection.describeSObject(param.getApi());
Field[] dsrFields = describeSObject.getFields();
while (true) {
DataiSyncParam queryParam = new DataiSyncParam();
queryParam.setApi(param.getApi());
queryParam.setSelect(param.getSelect());
queryParam.setBatchField(param.getBatchField());
queryParam.setIdField(param.getIdField());
queryParam.setIsDeleted(param.getIsDeleted());
queryParam.setBeginDate(param.getBeginDate());
queryParam.setEndDate(param.getEndDate());
queryParam.setLastDate(lastDate);
queryParam.setMaxId(maxId);
String query = buildDynamicQuery(queryParam);
log.info("执行查询SQL: {}", query);
QueryResult result = connection.queryAll(query);
if (result.getRecords() != null && result.getRecords().length > 0) {
objects = ConvertUtil.toJsonArray(result.getRecords(), dsrFields);
Date maxDate = objects.getJSONObject(objects.size() - 1).getDate(param.getBatchField());
Optional<String> maxIdOptional = objects.stream()
.map(t -> (JSONObject) t)
.filter(t -> maxDate.equals(t.getDate(param.getBatchField())))
.map(t -> t.getString("Id"))
.max(String::compareTo);
maxId = maxIdOptional.orElse(null);
lastDate = maxDate;
totalCount += processQueryResult(param.getApi(), result.getRecords(), objects);
log.info("已处理 {} 条记录", totalCount);
}
if (result.isDone() || result.getRecords() == null || result.getRecords().length == 0) {
break;
}
}
} catch (Exception e) {
log.error("查询处理数据时发生异常，对象API: {}", param.getApi(), e);
}
return totalCount;
}
/**
* 构建动态查询语句
*
* @param param 查询参数
* @return SOQL查询语句
*/
private String buildDynamicQuery(DataiSyncParam param) {
SoqlBuilder builder = new SoqlBuilder();
String[] selectFields = param.getSelect().split(",");
builder.select(selectFields)
.from(param.getApi());
if (param.getIsDeleted() != null && param.getIsDeleted()) {
builder.where("IsDeleted = false");
}
if (param.getBeginDate() != null && param.getEndDate() != null) {
builder.whereGe(param.getBatchField(), param.getBeginDate())
.whereLe(param.getBatchField(), param.getEndDate());
} else if (param.getBeginDate() != null) {
builder.whereGe(param.getBatchField(), param.getBeginDate());
} else if (param.getEndDate() != null) {
builder.whereLe(param.getBatchField(), param.getEndDate());
}
if (param.getBeginCreateDate() != null && param.getEndCreateDate() != null) {
builder.whereGe(param.getBatchField(), param.getBeginCreateDate())
.whereLe(param.getBatchField(), param.getEndCreateDate());
} else if (param.getBeginCreateDate() != null) {
builder.whereGe(param.getBatchField(), param.getBeginCreateDate());
} else if (param.getEndCreateDate() != null) {
builder.whereLe(param.getBatchField(), param.getEndCreateDate());
}
if (param.getLastDate() != null) {
builder.whereLe(param.getBatchField(), param.getLastDate());
}
if (param.getBeginModifyDate() != null && param.getEndModifyDate() != null) {
builder.whereGe(param.getBatchField(), param.getBeginModifyDate())
.whereLe(param.getBatchField(), param.getEndModifyDate());
} else if (param.getBeginModifyDate() != null) {
builder.whereGe(param.getBatchField(), param.getBeginModifyDate());
} else if (param.getEndModifyDate() != null) {
builder.whereLe(param.getBatchField(), param.getEndModifyDate());
}
if (StringUtils.isNotEmpty(param.getMaxId())) {
builder.whereGt(param.getIdField(), param.getMaxId());
}
@ -624,6 +737,7 @@ public class DataiIntegrationBatchServiceImpl implements IDataiIntegrationBatchS
if (param.getLimit() != null && param.getLimit() > 0) {
builder.limit(param.getLimit());
}
return builder.build();
}
@ -771,4 +885,6 @@ public class DataiIntegrationBatchServiceImpl implements IDataiIntegrationBatchS
return partitionName;
}
}

125
datai-scenes/datai-scene-salesforce/datai-salesforce-integration/src/main/java/com/datai/integration/service/impl/DataiIntegrationObjectServiceImpl.java
View File

@ -17,7 +17,6 @@ import com.datai.integration.service.IDataiIntegrationBatchService;
import com.datai.integration.service.IDataiIntegrationFieldService;
import com.datai.integration.service.IDataiIntegrationObjectService;
import com.datai.salesforce.common.utils.SoqlBuilder;
import com.datai.integration.util.ConvertUtil;
import com.datai.setting.future.SalesforceExecutor;
import com.sforce.soap.partner.DescribeSObjectResult;
import com.sforce.soap.partner.Field;
@ -710,6 +709,86 @@ public class DataiIntegrationObjectServiceImpl implements IDataiIntegrationObjec
return result;
}
@Override
public Map<String, Object> syncMultipleObjectData(Integer[] ids) {
Map<String, Object> result = new HashMap<>();
long startTime = System.currentTimeMillis();
try {
if (ids == null || ids.length == 0) {
log.error("对象ID数组不能为空");
result.put("success", false);
result.put("message", "对象ID数组不能为空");
return result;
}
log.info("开始同步多个对象数据，对象数量: {}", ids.length);
List<Map<String, Object>> syncResults = new ArrayList<>();
int successCount = 0;
int failureCount = 0;
int totalSyncRecords = 0;
List<Map<String, Object>> failedObjects = new ArrayList<>();
for (int i = 0; i < ids.length; i++) {
Integer id = ids[i];
log.info("开始同步第 {}/{} 个对象，对象ID: {}", i + 1, ids.length, id);
try {
Map<String, Object> singleResult = syncSingleObjectData(id);
syncResults.add(singleResult);
if ((Boolean) singleResult.get("success")) {
successCount++;
Integer totalCount = (Integer) singleResult.get("totalCount");
if (totalCount != null) {
totalSyncRecords += totalCount;
}
log.info("对象 {} 同步成功，同步记录数: {}", id, totalCount);
} else {
failureCount++;
Map<String, Object> failedObject = new HashMap<>();
failedObject.put("objectId", id);
failedObject.put("message", singleResult.get("message"));
failedObjects.add(failedObject);
log.error("对象 {} 同步失败: {}", id, singleResult.get("message"));
}
} catch (Exception e) {
failureCount++;
Map<String, Object> failedObject = new HashMap<>();
failedObject.put("objectId", id);
failedObject.put("message", "同步异常: " + e.getMessage());
failedObjects.add(failedObject);
log.error("同步对象 {} 时发生异常", id, e);
}
}
long endTime = System.currentTimeMillis();
long duration = endTime - startTime;
result.put("success", true);
result.put("message", "多对象数据同步完成");
result.put("totalObjects", ids.length);
result.put("successCount", successCount);
result.put("failureCount", failureCount);
result.put("totalSyncRecords", totalSyncRecords);
result.put("duration", duration);
result.put("syncResults", syncResults);
result.put("failedObjects", failedObjects);
log.info("多对象数据同步完成，总对象数: {}, 成功: {}, 失败: {}, 总同步记录数: {}, 耗时: {}ms",
ids.length, successCount, failureCount, totalSyncRecords, duration);
} catch (Exception e) {
log.error("同步多个对象数据时发生异常", e);
result.put("success", false);
result.put("message", "同步多个对象数据时发生异常: " + e.getMessage());
}
return result;
}
/**
* 全量数据拉取
* 查询所有批次并使用多线程并行拉取
@ -739,7 +818,6 @@ public class DataiIntegrationObjectServiceImpl implements IDataiIntegrationObjec
List<java.util.concurrent.Future<?>> futures = new ArrayList<>();
List<Map<String, Object>> batchResults = new CopyOnWriteArrayList<>();
int totalSyncCount = 0;
int successBatchCount = 0;
int failedBatchCount = 0;
@ -787,9 +865,6 @@ public class DataiIntegrationObjectServiceImpl implements IDataiIntegrationObjec
for (Map<String, Object> batchResult : batchResults) {
if ((Boolean) batchResult.get("success")) {
successBatchCount++;
if (batchResult.get("syncNum") != null) {
totalSyncCount += (Integer) batchResult.get("syncNum");
}
} else {
failedBatchCount++;
}
@ -800,7 +875,17 @@ public class DataiIntegrationObjectServiceImpl implements IDataiIntegrationObjec
LocalDateTime now = LocalDateTime.now();
object.setLastSyncDate(now);
object.setTotalRows(totalSyncCount);
DataiIntegrationBatch queryAllBatches = new DataiIntegrationBatch();
queryAllBatches.setApi(objectApi);
List<DataiIntegrationBatch> allBatches = dataiIntegrationBatchService.selectDataiIntegrationBatchList(queryAllBatches);
int totalDbNum = allBatches.stream()
.filter(batch -> batch.getDbNum() != null)
.mapToInt(DataiIntegrationBatch::getDbNum)
.sum();
object.setTotalRows(totalDbNum);
object.setSyncStatus(failedBatchCount == 0);
object.setUpdateTime(DateUtils.getNowDate());
updateDataiIntegrationObject(object);
@ -809,7 +894,7 @@ public class DataiIntegrationObjectServiceImpl implements IDataiIntegrationObjec
result.put("message", "全量数据拉取完成");
result.put("objectId", object.getId());
result.put("objectApi", objectApi);
result.put("totalCount", totalSyncCount);
result.put("totalCount", totalDbNum);
result.put("duration", duration);
result.put("syncType", "full");
result.put("lastBatchDate", object.getLastBatchDate());
@ -818,7 +903,7 @@ public class DataiIntegrationObjectServiceImpl implements IDataiIntegrationObjec
result.put("failedBatchCount", failedBatchCount);
log.info("对象 {} 全量数据拉取完成，共处理 {} 个批次，成功 {} 个，失败 {} 个，总记录数: {}，耗时: {}ms",
objectApi, batches.size(), successBatchCount, failedBatchCount, totalSyncCount, duration);
objectApi, batches.size(), successBatchCount, failedBatchCount, totalDbNum, duration);
} catch (Exception e) {
log.error("全量数据拉取失败，对象API: {}", objectApi, e);
@ -857,8 +942,6 @@ public class DataiIntegrationObjectServiceImpl implements IDataiIntegrationObjec
result.put("message", "无法获取Salesforce连接");
return result;
}
List<String> fieldList = getSalesforceObjectFields(connection, objectApi);
String batchField = object.getBatchField();
if (batchField == null || batchField.trim().isEmpty()) {
@ -876,7 +959,7 @@ public class DataiIntegrationObjectServiceImpl implements IDataiIntegrationObjec
DataiIntegrationBatch incrementalBatch = new DataiIntegrationBatch();
incrementalBatch.setApi(objectApi);
incrementalBatch.setLabel(object.getLabel());
incrementalBatch.setBatchField(batchField);
incrementalBatch.setBatchField(dataiIntegrationFieldService.getUpdateField(objectApi));
incrementalBatch.setSyncType("INCREMENTAL");
incrementalBatch.setSyncStatus(false);
incrementalBatch.setSyncStartDate(lastBatchDate);
@ -884,26 +967,22 @@ public class DataiIntegrationObjectServiceImpl implements IDataiIntegrationObjec
incrementalBatch.setCreateTime(DateUtils.getNowDate());
incrementalBatch.setUpdateTime(DateUtils.getNowDate());
int batchId = dataiIntegrationBatchService.insertDataiIntegrationBatch(incrementalBatch);
dataiIntegrationBatchService.insertDataiIntegrationBatch(incrementalBatch);
Integer batchId = incrementalBatch.getId();
log.info("为对象 {} 创建增量批次，批次ID: {}", objectApi, batchId);
object.setLastBatchDate(incrementalBatch.getSyncEndDate());
updateDataiIntegrationObject(object);
Map<String, Object> batchResult = dataiIntegrationBatchService.syncBatchData(batchId);
Integer totalDbNum = customMapper.countBySQL(objectApi, null);
long endTime = System.currentTimeMillis();
long duration = endTime - startTime;
LocalDateTime now = LocalDateTime.now();
if ((Boolean) batchResult.get("success")) {
object.setLastSyncDate(now);
Integer syncNum = (Integer) batchResult.get("syncNum");
if (object.getTotalRows() != null) {
object.setTotalRows(object.getTotalRows() + syncNum);
} else {
object.setTotalRows(syncNum);
}
object.setTotalRows(totalDbNum);
object.setSyncStatus(true);
object.setUpdateTime(DateUtils.getNowDate());
updateDataiIntegrationObject(object);
@ -912,13 +991,13 @@ public class DataiIntegrationObjectServiceImpl implements IDataiIntegrationObjec
result.put("message", "增量数据拉取成功");
result.put("objectId", object.getId());
result.put("objectApi", objectApi);
result.put("totalCount", syncNum);
result.put("totalCount", batchResult.get(""));
result.put("duration", duration);
result.put("syncType", "incremental");
result.put("lastBatchDate", object.getLastBatchDate());
result.put("lastSyncDate", object.getLastSyncDate());
log.info("对象 {} 增量数据拉取成功，同步数据量: {}，耗时: {}ms", objectApi, syncNum, duration);
log.info("对象 {} 增量数据拉取成功，总数据量: {}，耗时: {}ms", objectApi, totalDbNum, duration);
} else {
object.setSyncStatus(false);
object.setUpdateTime(DateUtils.getNowDate());