【feat】 bigobject文件下载初版

This commit is contained in:
Kris 2025-09-26 11:41:07 +08:00
parent ff6f9c0ee1
commit 89dcbb48bd

View File

@ -10,6 +10,7 @@ import com.celnet.datadump.config.SalesforceConnect;
import com.celnet.datadump.config.SalesforceExecutor; import com.celnet.datadump.config.SalesforceExecutor;
import com.celnet.datadump.config.SalesforceTargetConnect; import com.celnet.datadump.config.SalesforceTargetConnect;
import com.celnet.datadump.entity.*; import com.celnet.datadump.entity.*;
import com.celnet.datadump.enums.FileType;
import com.celnet.datadump.global.Const; import com.celnet.datadump.global.Const;
import com.celnet.datadump.global.SystemConfigCode; import com.celnet.datadump.global.SystemConfigCode;
import com.celnet.datadump.mapper.CustomMapper; import com.celnet.datadump.mapper.CustomMapper;
@ -142,21 +143,17 @@ public class CommonBatchServiceImpl implements CommonBatchService {
@Override @Override
public ReturnT<String> dumpBigObject(SalesforceParam param) throws Exception { public ReturnT<String> dumpBigObject(SalesforceParam param) throws Exception {
List<Future<?>> futures = Lists.newArrayList();
try { try {
if (StringUtils.isNotBlank(param.getApi())) { // 手动任务
// 手动任务 ReturnT<String> result = manualBigObjectDump(param);
ReturnT<String> result = manualBigObjectDump(param, futures); if (result != null) {
if (result != null) { return result;
return result;
}
} }
return ReturnT.SUCCESS;
} catch (Throwable throwable) { } catch (Throwable throwable) {
salesforceExecutor.remove(futures.toArray(new Future<?>[]{}));
log.error("dump error", throwable); log.error("dump error", throwable);
throw throwable; throw throwable;
} }
return ReturnT.SUCCESS;
} }
@Override @Override
@ -178,7 +175,7 @@ public class CommonBatchServiceImpl implements CommonBatchService {
} }
} }
private ReturnT<String> manualBigObjectDump(SalesforceParam param, List<Future<?>> futures) throws Exception { private ReturnT<String> manualBigObjectDump(SalesforceParam param) throws Exception {
String api = param.getApi(); String api = param.getApi();
@ -215,43 +212,48 @@ public class CommonBatchServiceImpl implements CommonBatchService {
QueryResultList queryResultList = bulkConnect.getQueryResultList(job.getId(), batchInfo.getId()); QueryResultList queryResultList = bulkConnect.getQueryResultList(job.getId(), batchInfo.getId());
log.info("获取查询结果列表, 结果数量: {}", queryResultList.getResult().length); log.info("获取查询结果列表, 结果数量: {}", queryResultList.getResult().length);
// 检测路径是否存在 不存在则创建 if (Const.FILE_TYPE == FileType.SERVER) {
Path apiPath = Paths.get(param.getApi()); // 检测路径是否存在 不存在则创建
if (!Files.exists(apiPath)) { File excel = new File(Const.SERVER_FILE_PATH + "/" + api);
Files.createDirectories(apiPath); if (!excel.exists()) {
log.info("创建目录: {}", apiPath.toAbsolutePath()); log.info("创建API目录: {}", excel.getAbsolutePath());
boolean mkdir = excel.mkdir();
if (!mkdir) {
log.info("创建文件存储目录失败!");
}
}
} }
// 使用确定性命名策略避免使用随机时间戳 // 使用确定性命名策略避免使用随机时间戳
String fileName = param.getApi() + "/" + DateFormatUtils.format(new Date(), "yyyyMMddHHmmss"); String fileName = Const.SERVER_FILE_PATH + "/" + api+ "/" + DateFormatUtils.format(new Date(), "yyyyMMddHHmmss");
final long maxFileSize = 50 * 1024 * 1024; // 50MB final long maxFileSize = 50 * 1024 * 1024; // 50MB
int fileIndex = 0; int fileIndex = 0;
Path currentFilePath = Paths.get(fileName + "_" + fileIndex + ".csv"); Path currentFilePath = Paths.get(fileName + "_" + fileIndex + ".csv");
log.info("开始写入数据到文件 : {}", fileName); log.info("开始写入数据到文件 : {}", fileName);
BufferedWriter bufferedWriter = null; BufferedWriter bufferedWriter = null;
FileWriter fileWriter = null; FileWriter fileWriter = null;
try { try {
fileWriter = new FileWriter(currentFilePath.toFile(), true); fileWriter = new FileWriter(currentFilePath.toFile(), true);
bufferedWriter = new BufferedWriter(fileWriter); bufferedWriter = new BufferedWriter(fileWriter);
long currentFileSize = 0; long currentFileSize = 0;
int resultIndex = 0; int resultIndex = 0;
for (final String resultId : queryResultList.getResult()) { for (final String resultId : queryResultList.getResult()) {
log.info("处理结果ID: {}, 进度: {}/{}", resultId, ++resultIndex, queryResultList.getResult().length); log.info("处理结果ID: {}, 进度: {}/{}", resultId, ++resultIndex, queryResultList.getResult().length);
InputStream resultStream = bulkConnect.getQueryResultStream(job.getId(), batchInfo.getId(), resultId); InputStream resultStream = bulkConnect.getQueryResultStream(job.getId(), batchInfo.getId(), resultId);
try (BufferedReader bufferedReader = new BufferedReader(new InputStreamReader(resultStream, StandardCharsets.UTF_8))) { try (BufferedReader bufferedReader = new BufferedReader(new InputStreamReader(resultStream, StandardCharsets.UTF_8))) {
String line; String line;
int lineCount = 0; int lineCount = 0;
while ((line = bufferedReader.readLine()) != null) { while ((line = bufferedReader.readLine()) != null) {
lineCount++; lineCount++;
byte[] lineBytes = (line + "\n").getBytes(StandardCharsets.UTF_8); byte[] lineBytes = (line + "\n").getBytes(StandardCharsets.UTF_8);
// 如果添加这行会超出文件大小限制则创建新文件 // 如果添加这行会超出文件大小限制则创建新文件
if (currentFileSize + lineBytes.length > maxFileSize && currentFileSize > 0) { if (currentFileSize + lineBytes.length > maxFileSize && currentFileSize > 0) {
// 关闭当前文件 // 关闭当前文件
@ -261,9 +263,9 @@ public class CommonBatchServiceImpl implements CommonBatchService {
if (fileWriter != null) { if (fileWriter != null) {
fileWriter.close(); fileWriter.close();
} }
log.info("当前文件已达到大小限制, 创建新文件. 文件路径: {}, 数据行数: {}", currentFilePath.toString(), lineCount); log.info("当前文件已达到大小限制, 创建新文件. 文件路径: {}, 数据行数: {}", currentFilePath.toString(), lineCount);
// 创建新文件 // 创建新文件
fileIndex++; fileIndex++;
currentFilePath = Paths.get(fileName + "_" + fileIndex + ".csv"); currentFilePath = Paths.get(fileName + "_" + fileIndex + ".csv");
@ -272,7 +274,7 @@ public class CommonBatchServiceImpl implements CommonBatchService {
currentFileSize = 0; currentFileSize = 0;
lineCount = 0; lineCount = 0;
} }
// 写入数据 // 写入数据
bufferedWriter.write(line + "\n"); bufferedWriter.write(line + "\n");
currentFileSize += lineBytes.length; currentFileSize += lineBytes.length;