From f62d1f65beab4a9e8714f47f495c7a525879aa7f Mon Sep 17 00:00:00 2001 From: Kris <2893855659@qq.com> Date: Fri, 24 Oct 2025 11:56:39 +0800 Subject: [PATCH] =?UTF-8?q?=E3=80=90feat=E3=80=91=20=E6=8F=90=E4=BA=A4bulk?= =?UTF-8?q?=20v2=E6=9C=89=E5=85=B3?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit (cherry picked from commit 713e545420cf3ded0315e77b39b13ac5b4ab0f67) --- .../datadump/config/SalesforceConnect.java | 7 + .../config/SalesforceTargetConnect.java | 7 + .../datadump/controller/JobController.java | 17 ++ .../celnet/datadump/job/DataDumpNewJob.java | 23 +++ .../datadump/service/CommonBatchService.java | 5 +- .../service/impl/CommonBatchServiceImpl.java | 169 +++++++++++++++++- .../com/celnet/datadump/util/BulkUtil.java | 60 +++++++ 7 files changed, 286 insertions(+), 2 deletions(-) diff --git a/src/main/java/com/celnet/datadump/config/SalesforceConnect.java b/src/main/java/com/celnet/datadump/config/SalesforceConnect.java index 69f12fd..e9de93b 100644 --- a/src/main/java/com/celnet/datadump/config/SalesforceConnect.java +++ b/src/main/java/com/celnet/datadump/config/SalesforceConnect.java @@ -54,12 +54,16 @@ public class SalesforceConnect { if ("SOURCE_ORG_PASSWORD".equals(map1.get("code"))) { map.put("password", String.valueOf(map1.get("value"))); } + if ("SOURCE_ORG_SEESION".equals(map1.get("code"))) { + map.put("seesionId", (String) map1.get("value")); + } } String username = map.get("username"); ConnectorConfig config = new ConnectorConfig(); config.setUsername(username); config.setPassword(map.get("password")); + config.setSessionId(map.get("seesionId")); String url = map.get("url"); config.setAuthEndpoint(url); config.setServiceEndpoint(url); @@ -106,6 +110,9 @@ public class SalesforceConnect { if ("SOURCE_ORG_PASSWORD".equals(map1.get("code"))) { map.put("password", String.valueOf(map1.get("value"))); } + if ("SOURCE_ORG_SEESION".equals(map1.get("code"))) { + map.put("seesionId", (String) map1.get("value")); + } } return BulkUtil.getBulkConnection(map.get("username"),map.get("password"),map.get("url")); } catch (Exception e) { diff --git a/src/main/java/com/celnet/datadump/config/SalesforceTargetConnect.java b/src/main/java/com/celnet/datadump/config/SalesforceTargetConnect.java index 2a788c1..6a21504 100644 --- a/src/main/java/com/celnet/datadump/config/SalesforceTargetConnect.java +++ b/src/main/java/com/celnet/datadump/config/SalesforceTargetConnect.java @@ -49,12 +49,16 @@ public class SalesforceTargetConnect { if ("TARGET_ORG_PASSWORD".equals(map1.get("code"))) { map.put("password", (String) map1.get("value")); } + if ("TARGET_ORG_SEESION".equals(map1.get("code"))) { + map.put("seesionId", (String) map1.get("value")); + } } String username = map.get("username"); ConnectorConfig config = new ConnectorConfig(); config.setUsername(username); config.setPassword(map.get("password")); + config.setSessionId(map.get("seesionId")); String url = map.get("url"); config.setAuthEndpoint(url); config.setServiceEndpoint(url); @@ -105,6 +109,9 @@ public class SalesforceTargetConnect { if ("TARGET_ORG_PASSWORD".equals(map1.get("code"))) { map.put("password", (String) map1.get("value")); } + if ("TARGET_ORG_SEESION".equals(map1.get("code"))) { + map.put("seesionId", (String) map1.get("value")); + } } return BulkUtil.getBulkConnection(map.get("username"),map.get("password"),map.get("url")); } catch (Exception e) { diff --git a/src/main/java/com/celnet/datadump/controller/JobController.java b/src/main/java/com/celnet/datadump/controller/JobController.java index 4085b69..7833322 100644 --- a/src/main/java/com/celnet/datadump/controller/JobController.java +++ b/src/main/java/com/celnet/datadump/controller/JobController.java @@ -337,6 +337,23 @@ public class JobController { return commonBatchService.dumpBigObject(param); } + @PostMapping("/dataDumpBigObjectV2Job") + @ApiOperation("存量任务（BigObject_v2）") + @LogServiceAnnotation(operateType = OperateTypeConstant.TYPE_INSERT, remark = "存量任务（BigObject_v2）") + public ReturnT dataDumpBigObjectV2Job(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, "参数解析失败！"); + } + + return commonBatchService.dumpBigObjectV2(param); + } + /** * bulk批量更新大数据量数据 * @param paramStr diff --git a/src/main/java/com/celnet/datadump/job/DataDumpNewJob.java b/src/main/java/com/celnet/datadump/job/DataDumpNewJob.java index a23834b..b382f4c 100644 --- a/src/main/java/com/celnet/datadump/job/DataDumpNewJob.java +++ b/src/main/java/com/celnet/datadump/job/DataDumpNewJob.java @@ -138,6 +138,29 @@ public class DataDumpNewJob { return commonBatchService.dumpBigObject(param); } + /** + * 存量任务（BigObject_V2） + * + * @param paramStr 参数json + * @return result + */ + @XxlJob("dataDumpBigObjectV2Job") + public ReturnT dataDumpBigObjectV2Job(String paramStr) throws Exception { + log.info("dataDumpBigObjectV2Job 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.dumpBigObjectV2(param); + } /** * 导入BigObject数据到目标ORG * diff --git a/src/main/java/com/celnet/datadump/service/CommonBatchService.java b/src/main/java/com/celnet/datadump/service/CommonBatchService.java index 16b37fc..0bbdd0b 100644 --- a/src/main/java/com/celnet/datadump/service/CommonBatchService.java +++ b/src/main/java/com/celnet/datadump/service/CommonBatchService.java @@ -8,7 +8,10 @@ public interface CommonBatchService { ReturnT dumpBatch(SalesforceParam param) throws Exception; ReturnT dumpBigObject(SalesforceParam param) throws Exception; - + + ReturnT dumpBigObjectV2(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/impl/CommonBatchServiceImpl.java b/src/main/java/com/celnet/datadump/service/impl/CommonBatchServiceImpl.java index 26a66cd..b58dd27 100644 --- a/src/main/java/com/celnet/datadump/service/impl/CommonBatchServiceImpl.java +++ b/src/main/java/com/celnet/datadump/service/impl/CommonBatchServiceImpl.java @@ -155,7 +155,23 @@ public class CommonBatchServiceImpl implements CommonBatchService { } return ReturnT.SUCCESS; } - + + @Override + public ReturnT dumpBigObjectV2(SalesforceParam param) throws Exception { + try { + // 手动任务 + ReturnT result = manualBigObjectDumpV2(param); + if (result != null) { + return result; + } + } catch (Throwable throwable) { + log.error("dump error", throwable); + throw throwable; + } + return ReturnT.SUCCESS; + } + + @Override public ReturnT importBigObject(SalesforceParam param) throws Exception { try { @@ -310,6 +326,157 @@ public class CommonBatchServiceImpl implements CommonBatchService { return ReturnT.SUCCESS; } + private ReturnT manualBigObjectDumpV2(SalesforceParam param) throws Exception { + + String api = param.getApi(); + BulkConnection bulkConnect = salesforceConnect.createBulkConnect(); + String jobId = null; + + try { + log.info("开始使用Bulk API v2导出BigObject数据, API: {}", api); + + List fieldList = dataFieldService.list(new QueryWrapper().eq("api", param.getApi())); + String fieldStr = fieldList.stream() + .map(DataField::getField) + .filter(field -> !"SystemModstamp".equals(field) && !"LastModifiedDate".equals(field)) + .collect(Collectors.joining(", ")); + + String sql = "select " + fieldStr + " from " + param.getApi(); + log.info("构建查询SQL: {}", sql); + + // 创建Bulk V2 API作业 + String restEndpoint = bulkConnect.getConfig().getRestEndpoint(); + String url = restEndpoint.substring(0, restEndpoint.indexOf("services/")); + log.debug("Salesforce实例URL: {}", url); + + jobId = BulkUtil.createBulkV2QueryJob(url, sql, bulkConnect); + log.info("创建Bulk V2查询作业成功, 作业ID: {}", jobId); + + // 等待作业完成 + BulkUtil.waitForBulkV2QueryJobCompletion(url, jobId, bulkConnect); + log.info("Bulk V2查询作业已完成, 作业ID: {}", jobId); + + if (Const.FILE_TYPE == FileType.SERVER) { + // 检测路径是否存在 不存在则创建 + File excel = new File(Const.SERVER_FILE_PATH + "/" + api); + if (!excel.exists()) { + log.info("创建API目录: {}", excel.getAbsolutePath()); + boolean mkdir = excel.mkdir(); + if (!mkdir) { + log.info("创建文件存储目录失败！"); + } + } + } + + // 使用确定性命名策略，避免使用随机时间戳 + String fileName = Const.SERVER_FILE_PATH + "/" + api + "/" + api + "_" + DateFormatUtils.format(new Date(), "yyyyMMddHHmmss"); + final long maxFileSize = 50 * 1024 * 1024; // 50MB + + int fileIndex = 0; + Path currentFilePath = Paths.get(fileName + "_" + fileIndex + ".csv"); + + log.info("开始写入数据到文件: {}", fileName); + + BufferedWriter bufferedWriter = null; + FileWriter fileWriter = null; + + try { + fileWriter = new FileWriter(currentFilePath.toFile(), true); + bufferedWriter = new BufferedWriter(fileWriter); + long currentFileSize = 0; + + String headerLine = null; + String locator = null; + boolean hasMoreData = true; + int totalRecords = 0; + + while (hasMoreData) { + // 获取查询结果 + BulkV2QueryResult result = BulkUtil.getBulkV2QueryJobResults(url, jobId, bulkConnect, locator); + String csvContent = result.getCsvContent(); + locator = result.getLocator(); + int recordCount = result.getRecordCount(); + + totalRecords += recordCount; + log.info("获取到{}条记录，总计{}条记录", recordCount, totalRecords); + + if (recordCount > 0) { + // 处理CSV内容 + try (BufferedReader bufferedReader = new BufferedReader(new StringReader(csvContent))) { + String line; + int lineCount = 0; + while ((line = bufferedReader.readLine()) != null) { + if (headerLine == null) { + // 保存第一行作为标题行，并替换CreatedDate和CreatedById为Original_Created_Date__c和Original_Created_By__c + // 但要确保不替换CreatedDate__c和CreatedById__c这样的自定义字段 + line = line.replaceAll("(? maxFileSize && currentFileSize > 0) { + // 关闭当前文件 + if (bufferedWriter != null) { + bufferedWriter.close(); + } + if (fileWriter != null) { + fileWriter.close(); + } + + log.info("当前文件已达到大小限制, 创建新文件. 文件路径: {}, 数据行数: {}", currentFilePath.toString(), lineCount); + + // 创建新文件 + fileIndex++; + currentFilePath = Paths.get(fileName + "_" + fileIndex + ".csv"); + fileWriter = new FileWriter(currentFilePath.toFile(), true); + bufferedWriter = new BufferedWriter(fileWriter); + currentFileSize = 0; + lineCount = 1; // 重置行计数，新文件将写入标题行 + + // 在新文件中写入标题行 + if (headerLine != null) { + bufferedWriter.write(headerLine + "\n"); + currentFileSize += (headerLine + "\n").getBytes(StandardCharsets.UTF_8).length; + } + } + + // 写入数据 + bufferedWriter.write(line + "\n"); + currentFileSize += lineBytes.length; + } + } + } + + // 检查是否还有更多数据 + hasMoreData = locator != null && !locator.isEmpty() && !"null".equals(locator) && StringUtils.isNotEmpty(locator); + + if (hasMoreData) { + log.info("还有更多数据需要获取，locator: {}", locator); + } + } + + log.info("所有数据写入完成, 总文件数: {}, 总记录数: {}", fileIndex + 1, totalRecords); + } finally { + if (bufferedWriter != null) { + bufferedWriter.close(); + } + if (fileWriter != null) { + fileWriter.close(); + } + } + + } catch (Exception e) { + log.error("使用Bulk API v2导出BigObject数据失败, API: {}", api, e); + throw e; + } + + log.info("BigObject数据导出任务完成, API: {}", api); + return ReturnT.SUCCESS; + } + private ReturnT manualBigObjectImport(SalesforceParam param) throws Exception { String api = param.getApi(); diff --git a/src/main/java/com/celnet/datadump/util/BulkUtil.java b/src/main/java/com/celnet/datadump/util/BulkUtil.java index 4f50f40..6919c77 100644 --- a/src/main/java/com/celnet/datadump/util/BulkUtil.java +++ b/src/main/java/com/celnet/datadump/util/BulkUtil.java @@ -205,6 +205,47 @@ public class BulkUtil { } } + /** + * 通过轮询作业状态等待 Bulk API 2.0 作业完成。 + * + * @param jobId 要检查的作业ID。 + * @param connection 用于获取会话信息的BulkConnection。 + * @throws Exception 轮询过程中发生错误时抛出异常。 + */ + public static void waitForBulkV2QueryJobCompletion(String url, String jobId, BulkConnection connection) throws Exception { + CloseableHttpClient httpClient = HttpClients.createDefault(); + String jobStatusUrl = url + "services/data/v56.0/jobs/query/" + jobId; + boolean jobCompleted = false; + + while (!jobCompleted) { + HttpGet httpGet = new HttpGet(jobStatusUrl); + httpGet.setHeader("Authorization", "Bearer " + connection.getConfig().getSessionId()); + + CloseableHttpResponse response = httpClient.execute(httpGet); + try { + int statusCode = response.getStatusLine().getStatusCode(); + if (statusCode == 200) { + String responseBody = EntityUtils.toString(response.getEntity()); + // 简单解析响应体中的状态字段，实际项目中建议使用JSON库如Jackson + if (responseBody.contains("\"state\":\"JobComplete\"")) { + System.out.println("作业成功完成。"); + jobCompleted = true; + } else if (responseBody.contains("\"state\":\"Failed\"") || responseBody.contains("\"state\":\"Aborted\"")) { + throw new RuntimeException("作业失败或已被中止。响应内容: " + responseBody); + } else { + System.out.println("作业仍在运行中。等待中..."); + Thread.sleep(10000); // 等待10秒后再次检查 + } + } else { + String errorBody = EntityUtils.toString(response.getEntity()); + throw new RuntimeException("获取作业状态失败，状态码: " + statusCode + "，错误: " + errorBody); + } + } finally { + response.close(); + } + } + httpClient.close(); + } /** * 通过轮询作业状态等待 Bulk API 2.0 作业完成。 * @@ -257,8 +298,26 @@ public class BulkUtil { * @throws Exception 如果在获取结果过程中发生错误 */ public static BulkV2QueryResult getBulkV2QueryJobResults(String url, String jobId, BulkConnection connection) throws Exception { + return getBulkV2QueryJobResults(url, jobId, connection, null); + } + + /** + * 获取 Bulk API 2.0 查询作业的结果 + * + * @param url Salesforce实例URL + * @param jobId 作业ID + * @param connection BulkConnection used to get session info. + * @param locator 分页定位器，用于获取下一批结果 + * @return 查询结果的字符串内容 + * @throws Exception 如果在获取结果过程中发生错误 + */ + public static BulkV2QueryResult getBulkV2QueryJobResults(String url, String jobId, BulkConnection connection, String locator) throws Exception { CloseableHttpClient httpClient = HttpClients.createDefault(); String resultsUrl = url + "/services/data/v56.0/jobs/query/" + jobId + "/results"; + // 只有当locator不为null且不为空时才添加到URL中 + if (locator != null && !locator.isEmpty() && !"null".equals(locator)) { + resultsUrl += "?locator=" + java.net.URLEncoder.encode(locator, "UTF-8"); + } HttpGet httpGet = new HttpGet(resultsUrl); httpGet.setHeader("Authorization", "Bearer " + connection.getConfig().getSessionId()); httpGet.setHeader("Content-Type", "application/json"); @@ -274,6 +333,7 @@ public class BulkUtil { String newLocator = (locatorHeader != null) ? locatorHeader.getValue() : null; int numberOfRecords = (numRecordsHeader != null) ? Integer.parseInt(numRecordsHeader.getValue()) : 0; String csvContent = EntityUtils.toString(response.getEntity()); + System.out.println("获取作业结果成功。" + response); return new BulkV2QueryResult(csvContent, newLocator, numberOfRecords); } else { String errorBody = EntityUtils.toString(response.getEntity());