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

【feat】 动态调整bulk批次大小

Browse Source
This commit is contained in:
Kris 2025-09-25 10:39:11 +08:00
parent 03fa708ed7
commit 54533b0bb3
3 changed files with 39 additions and 10 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
src/main/java/com/celnet/datadump/global/SystemConfigCode.java
View File

@ -40,6 +40,11 @@ public class SystemConfigCode {
*/
public static final String INCREMENT_BATCH_TYPE = "INCREMENT_BATCH_TYPE";
/**
* 批次处理记录数
*/
public static final String BATCH_PROCESS_COUNT = "BATCH_PROCESS_COUNT";
public static final String BATCH_TYPE_WEEK = "WEEK";
public static final String BATCH_TYPE_MONTH = "MONTH";
public static final String BATCH_TYPE_YEAR = "YEAR";
@ -48,4 +53,4 @@ public class SystemConfigCode {
public static final String INFO_FLAG_FLASE = "FLASE";
}
}

39
src/main/java/com/celnet/datadump/service/impl/CommonBatchServiceImpl.java
View File

@ -11,6 +11,7 @@ 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.global.SystemConfigCode;
import com.celnet.datadump.mapper.CustomMapper;
import com.celnet.datadump.param.DataDumpParam;
import com.celnet.datadump.param.DataDumpSpecialParam;
@ -51,6 +52,7 @@ import org.apache.http.util.EntityUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import javax.annotation.PostConstruct;
import java.io.*;
import java.nio.charset.StandardCharsets;
import java.nio.file.Files;
@ -88,7 +90,30 @@ public class CommonBatchServiceImpl implements CommonBatchService {
private DataDumpSpecialService dataDumpSpecialService;
@Autowired
private DataBatchHistoryService dataBatchHistoryService;
// 批次处理记录数，默认为10000
private int batchProcessCount = 10000;
@Autowired
private SystemConfigService systemConfigService;
@PostConstruct
public void init() {
QueryWrapper<SystemConfig> queryWrapper = new QueryWrapper<>();
queryWrapper.eq("code", SystemConfigCode.BATCH_PROCESS_COUNT);
SystemConfig systemConfig = systemConfigService.getOne(queryWrapper);
if (systemConfig != null && systemConfig.getValue() != null) {
try {
batchProcessCount = Integer.parseInt(systemConfig.getValue());
log.info("批次处理记录数已设置为: {}", batchProcessCount);
} catch (NumberFormatException e) {
log.warn("无法解析批次处理记录数配置，使用默认值: {}", batchProcessCount);
}
} else {
log.info("未找到批次处理记录数配置，使用默认值: {}", batchProcessCount);
}
}
@Override
public ReturnT<String> dumpBatch(SalesforceParam param) throws Exception {
List<Future<?>> futures = Lists.newArrayList();
@ -528,7 +553,7 @@ public class CommonBatchServiceImpl implements CommonBatchService {
while (hasMore) {
int sfData = 0;
salesforceParam.setMaxId(maxId);
salesforceParam.setLimit(10000);
salesforceParam.setLimit(batchProcessCount);
map.put("param", salesforceParam);
JobInfo job = null;
try {
@ -592,9 +617,9 @@ public class CommonBatchServiceImpl implements CommonBatchService {
log.error("处理查询结果时异常: ", e);
throw new AsyncApiException("处理查询结果时异常: " + e.getMessage(), AsyncExceptionCode.Unknown);
}finally {
if (sfData != 10000){
if (sfData != batchProcessCount){
hasMore = false;
log.info("当前批次数据不足10000条, 结束循环");
log.info("当前批次数据不足{}条, 结束循环", batchProcessCount);
}
}
}
@ -644,7 +669,7 @@ public class CommonBatchServiceImpl implements CommonBatchService {
while (hasMore) {
int sfData = 0;
param.setMaxId(maxId);
param.setLimit(10000);
param.setLimit(batchProcessCount);
// 获取创建时间
param.setTimestamp(lastCreatedDate);
map.put("param", param);
@ -718,9 +743,9 @@ public class CommonBatchServiceImpl implements CommonBatchService {
log.error("处理查询结果时异常: ", e);
throw new AsyncApiException("处理查询结果时异常: " + e.getMessage(), AsyncExceptionCode.Unknown);
}finally {
if (sfData != 10000){
if (sfData != batchProcessCount){
hasMore = false;
log.info("当前批次数据不足10000条, 结束循环");
log.info("当前批次数据不足{}条, 结束循环", batchProcessCount);
}
}
}

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

@ -1239,7 +1239,7 @@ public class DataImportNewServiceImpl implements DataImportNewService {
//批量插入200一次
int page = count % 200 == 0 ? count / 200 : (count / 200) + 1;
for (int i = 0; i < page; i++) {
List<Map<String, Object>> linkList = customMapper.list("Id,LinkedEntityId,ContentDocumentId,LinkedEntity_Type,ShareType,Visibility", api, "ShareType = 'V' and new_id = '0' order by Id limit 200");
List<Map<String, Object>> linkList = customMapper.list("Id,LinkedEntityId,ContentDocumentId,LinkedEntity_Type,ShareType,Visibility", api, "ShareType = 'V' and new_id = '0' order by Id limit " + i * 200 + ",200");
SObject[] accounts = new SObject[linkList.size()];
String[] ids = new String[linkList.size()];
int index = 0;
@ -1308,7 +1308,6 @@ public class DataImportNewServiceImpl implements DataImportNewService {
return ReturnT.SUCCESS;
}
@Override
public ReturnT<String> dumpFileNew(SalesforceParam param) throws Exception {
String downloadUrl = null;