【feat】 优化增量任务

(cherry picked from commit c12fbf263a)

src/main/java/com/celnet/datadump/service/impl/CommonBatchServiceImpl.java

@@ -132,6 +132,8 @@ public class CommonBatchServiceImpl implements CommonBatchService {
                 if (result != null) {
                     return result;
                 }
+            }else {
+                autoBatchDump(param, futures);
             }
             return ReturnT.SUCCESS;
         } catch (Throwable throwable) {
@@ -949,6 +951,87 @@ public class CommonBatchServiceImpl implements CommonBatchService {
         return ReturnT.SUCCESS;
     }
 
+    private void autoBatchDump(SalesforceParam param, List<Future<?>> futures) throws InterruptedException {
+        QueryWrapper<DataObject> qw = new QueryWrapper<>();
+        qw.eq("data_work", 1)
+                .eq("data_lock", 0)
+                .orderByAsc("data_index")
+                .last(" limit 10");
+        while (true) {
+            List<DataObject> dataObjects = dataObjectService.list(qw);
+            if (CollectionUtils.isEmpty(dataObjects)) {
+                break;
+            }
+
+            // 根据参数获取sql
+            for (DataObject update : dataObjects) {
+                String api = update.getName();
+                TimeUnit.MILLISECONDS.sleep(1);
+                try {
+                    commonService.checkApi(api, true);
+                    if (!dataFieldService.hasCreatedDate(api)) {
+                        DataDumpSpecialParam dataDumpSpecialParam = new DataDumpSpecialParam();
+                        dataDumpSpecialParam.setApi(api);
+                        Future<?> future = getData(dataDumpSpecialParam, salesforceConnect.createBulkConnect());
+                        // 等待当前所有线程执行完成
+                        salesforceExecutor.waitForFutures(future);
+                        continue;
+                    }
+                    param.setApi(api);
+                    List<SalesforceParam> salesforceParams = null;
+                    update.setName(api);
+                    update.setDataLock(1);
+                    dataObjectService.updateById(update);
+                    QueryWrapper<DataBatch> dbQw = new QueryWrapper<>();
+                    dbQw.eq("name", api)
+                            .isNull("first_sync_date");
+                    if (param.getBeginDate() != null && param.getEndDate() != null) {
+                        dbQw.eq("sync_start_date", param.getBeginDate());
+                        dbQw.eq("sync_end_date", param.getEndDate());
+                    }
+                    List<DataBatch> 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());
+                    } else {
+                        salesforceParams = DataUtil.splitTask(param);
+                    }
+                    // 手动任务优先执行
+                    for (SalesforceParam salesforceParam : salesforceParams) {
+                        Future<?> future = salesforceExecutor.execute(() -> {
+                            try {
+                                dumpBatchData(salesforceParam);
+                            } catch (Throwable throwable) {
+                                log.error("salesforceExecutor error", throwable);
+                                throw new RuntimeException(throwable);
+                            }
+                        }, salesforceParam.getBatch(), 1);
+                        futures.add(future);
+                    }
+
+                    // 等待当前所有线程执行完成
+                    salesforceExecutor.waitForFutures(futures.toArray(new Future<?>[]{}));
+
+                    update.setDataWork(0);
+                } catch (Throwable e) {
+                    log.error("manualDump error", e);
+                    throw new RuntimeException(e);
+                } finally {
+                    update.setName(api);
+                    update.setDataLock(0);
+                    dataObjectService.updateById(update);
+                }
+            }
+        }
+    }
+
     private BulkConnection dumpBatchData(SalesforceParam param) throws Throwable {
         String api = param.getApi();
         BulkConnection bulkConnect = null;
@@ -1119,6 +1202,7 @@ public class CommonBatchServiceImpl implements CommonBatchService {
         Map<String, Object> map = Maps.newHashMap();
         String dateName = param.getType() == 1 ? Const.CREATED_DATE : param.getUpdateField();
         int count = 0;
+        int batch = 0;
         Date lastCreatedDate = null;
         String maxId = null;
 
@@ -1155,7 +1239,9 @@ public class CommonBatchServiceImpl implements CommonBatchService {
                 }
                 log.info("query sql: {}", sql);
                 XxlJobLogger.log("query sql: {}", sql);
-                
+
+                DataLog dataLog = new DataLog(api, null, new Date(), null, "数据拉取，拉取第" + batch + "批数据", "QuerySF");
+
                 job = BulkUtil.createJob(bulkConnect, api, OperationEnum.queryAll);
                 log.info("创建Bulk作业成功, 作业ID: {}", job.getId());
 
@@ -1165,6 +1251,8 @@ public class CommonBatchServiceImpl implements CommonBatchService {
                 int completionOne = BulkUtil.awaitCompletionOne(bulkConnect, job, batchInfo);
                 log.info("批次处理完成, 处理记录数: {}", completionOne);
 
+                dataLogService.save(dataLog);
+
                 if (completionOne == 0){
                     log.info("无更多数据, 结束处理");
                     BulkUtil.closeJob(bulkConnect, job.getId());
@@ -1173,6 +1261,8 @@ public class CommonBatchServiceImpl implements CommonBatchService {
 
                 List<Map<String, Object>> batchRecords = new ArrayList<>();
 
+                DataLog dataLog1 = new DataLog(api, null, new Date(), null, "数据拉取，Upsert第" + batch + "批数据", "UpsertDB");
+
                 QueryResultList queryResultList = bulkConnect.getQueryResultList(job.getId(), batchInfo.getId());
                 log.info("获取查询结果列表, 结果数量: {}", queryResultList.getResult().length);
 
@@ -1194,6 +1284,8 @@ public class CommonBatchServiceImpl implements CommonBatchService {
                     }
                 }
 
+                dataLogService.save(dataLog1);
+
                 if (!batchRecords.isEmpty()){
                     log.info("开始保存或更新数据, 记录数: {}", batchRecords.size());
                     sfData = saveOrUpdate(api, batchRecords, true);
@@ -1206,6 +1298,7 @@ public class CommonBatchServiceImpl implements CommonBatchService {
                 }
 
                 BulkUtil.closeJob(bulkConnect, job.getId());
+                batch ++;
                 log.info("关闭作业成功, 作业ID: {}", job.getId());
             }catch (InterruptedException interruptedException){
                 return 0;

src/main/java/com/celnet/datadump/service/impl/DataDumpSpecialServiceImpl.java

@@ -262,8 +262,9 @@ public class DataDumpSpecialServiceImpl implements DataDumpSpecialService {
                         // 判断是否存在要排除的id
                         salesforceParam.setMaxId(maxId);
                         salesforceParam.setBeginModifyDate(dataObject.getLastUpdateDate());
+                        salesforceParam.setUpdateField(dataFieldService.returnUpdateDateField(dataObject.getName()));
                         map.put("param", salesforceParam);
-                        String sql = SqlUtil.showSql("com.celnet.datadump.mapper.SalesforceMapper.listOrderByIdNew", map);
+                        String sql = SqlUtil.showSql("com.celnet.datadump.mapper.SalesforceMapper.listOrderByIdShare", map);
                         log.info("query sql: {}", sql);
                         XxlJobLogger.log("query sql: {}", sql);
                         QueryResult queryResult = connect.queryAll(sql);

src/main/java/com/celnet/datadump/service/impl/DataImportBatchServiceImpl.java

@@ -21,9 +21,6 @@ import com.celnet.datadump.util.DataUtil;
 import com.celnet.datadump.util.EmailUtil;
 import com.google.common.collect.Lists;
 import com.sforce.async.*;
-import com.sforce.soap.partner.PartnerConnection;
-import com.sforce.soap.partner.SaveResult;
-import com.sforce.soap.partner.sobject.SObject;
 import com.xxl.job.core.biz.model.ReturnT;
 import com.xxl.job.core.log.XxlJobLogger;
 import com.xxl.job.core.util.DateUtil;
@@ -81,6 +78,9 @@ public class DataImportBatchServiceImpl implements DataImportBatchService {
     @Autowired
     private LinkConfigService linkConfigService;
 
+    @Autowired
+    private DataLogService dataLogService;
+
     /**
      * Insert入口
      */
@@ -286,6 +286,11 @@ public class DataImportBatchServiceImpl implements DataImportBatchService {
         String api = param.getApi();
         QueryWrapper<DataField> dbQw = new QueryWrapper<>();
         dbQw.eq("api", api);
+        dbQw.and(wrapper -> wrapper.eq("is_createable", 1)
+                .eq("is_nillable", 0)
+                .eq("is_defaulted_on_create", 0)
+                        .or().eq("field", "CreatedDate")
+                        .or().eq("field", "CreatedById"));
         List<DataField> list = dataFieldService.list(dbQw);
 
         String beginDateStr = null;
@@ -326,6 +331,8 @@ public class DataImportBatchServiceImpl implements DataImportBatchService {
 
         for (int i = 0; i < page; i++) {
 
+            DataLog dataLog = new DataLog(api, "开始时间："+beginDateStr+";结束时间："+endDateStr, new Date(), null, "数据更新（BULK），查询并组装第" + i + "批数据", "QueryDB");
+
             List<JSONObject> data = customMapper.listJsonObject("*", api, "new_id is null  and CreatedDate >= '" + beginDateStr + "' and CreatedDate < '" + endDateStr + "' limit 8000");
             int size = data.size();
 
@@ -344,10 +351,12 @@ public class DataImportBatchServiceImpl implements DataImportBatchService {
                 String message = null;
                 JSONObject account = new JSONObject();
                 for (DataField dataField : list) {
-                    if ("OwnerId".equals(dataField.getField()) || "Owner_Type".equals(dataField.getField())
+                    if ("OwnerId".equals(dataField.getField()) || "Owner_Type".equals(dataField.getField())){
-                            || "Id".equals(dataField.getField())){
                         continue;
                     }
+                    if ("Id".equals(dataField.getField())){
+                        account.put("Id", data.get(j - 1).get(dataField.getField()));
+                    }
                     if ("CreatedDate".equals(dataField.getField()) && dataField.getIsCreateable()){
                         // 转换为UTC时间并格式化
                         LocalDateTime localDateTime = LocalDateTime.parse(String.valueOf(data.get(j - 1).get("CreatedDate")),  inputFormatter);
@@ -367,6 +376,7 @@ public class DataImportBatchServiceImpl implements DataImportBatchService {
                         }
                         continue;
                     }
+
                     if (dataField.getIsCreateable() !=null && dataField.getIsCreateable() && !dataField.getIsNillable() && !dataField.getIsDefaultedOnCreate()) {
                         if ("reference".equals(dataField.getSfType()) && data.get(j - 1).get(dataField.getField()) != null){
                             //引用类型
@@ -417,23 +427,35 @@ public class DataImportBatchServiceImpl implements DataImportBatchService {
                     account.put("old_sfdc_id__c", data.get(j-1).get("Id"));
                 }
 
-                ids[j-1] = data.get(j-1).get("Id").toString();
                 insertList.add(account);
                 if (i*8000+j >= count){
                     break;
                 }
             }
+            dataLog.setEndTime(new Date());
+            dataLogService.save(dataLog);
+
             JobInfo salesforceInsertJob = null;
 
             try {
+                DataLog dataLog1 = new DataLog(api, "开始时间："+beginDateStr+";结束时间："+endDateStr, new Date(), null, "数据更新（BULK），更新SF第" + i + "批数据", "UpdateSF");
+
                 salesforceInsertJob = BulkUtil.createJob(bulkConnection, api, OperationEnum.insert);
 
-                BatchInfo batchInfo = CsvConverterUtil.writeToCsvNewOne(insertList, UUID.randomUUID().toString(), list, true, bulkConnection, salesforceInsertJob);
+                BatchInfo batchInfo = CsvConverterUtil.writeToCsvOne(insertList, UUID.randomUUID().toString(), list, true, bulkConnection, salesforceInsertJob,ids);
 
                 BulkUtil.awaitOneCompletion(bulkConnection, salesforceInsertJob, batchInfo);
 
+                dataLog1.setEndTime(new Date());
+                dataLogService.save(dataLog1);
+
+                DataLog dataLog2 = new DataLog(api, "开始时间："+beginDateStr+";结束时间："+endDateStr, new Date(), null, "数据更新（BULK），更新DB第" + i + "批数据", "UpdateDB");
+
                 sfNum = sfNum + checkInsertResultsOne(bulkConnection, salesforceInsertJob, batchInfo, api, ids);
 
+                dataLog2.setEndTime(new Date());
+                dataLogService.save(dataLog2);
+
                 TimeUnit.MILLISECONDS.sleep(1);
 
             } catch (InterruptedException interruptedException){
@@ -511,11 +533,29 @@ public class DataImportBatchServiceImpl implements DataImportBatchService {
      * 读写Insert结果
      */
     public int checkInsertResultsOne(BulkConnection connection, JobInfo job,
-                                  BatchInfo b,String api,String[] ids)
+                              BatchInfo b,String api,String[] ids)
-            throws AsyncApiException, IOException {
+        throws Exception {
         int index = 0;
-        CSVReader rdr =
+        CSVReader rdr = null;
-                new CSVReader(connection.getBatchResultStream(job.getId(), b.getId()));
+        int retryCount = 0;
+        final int maxRetries = 3;
+        
+        while (retryCount <= maxRetries) {
+            try {
+                rdr = new CSVReader(connection.getBatchResultStream(job.getId(), b.getId()));
+                break; // 成功执行，跳出重试循环
+            }catch (Exception e) {
+                retryCount++;
+                log.info("checkInsertResultsOne failed, retrying {}/{} for api: {}", retryCount, maxRetries, api);
+                try {
+                    Thread.sleep(1000 * retryCount); // 简单的退避策略
+                } catch (InterruptedException ie) {
+                    Thread.currentThread().interrupt();
+                    throw new IOException("Interrupted while retrying", ie);
+                }
+            }
+        }
+
         List<String> resultHeader = rdr.nextRecord();
         int resultCols = resultHeader.size();
 
@@ -543,7 +583,7 @@ public class DataImportBatchServiceImpl implements DataImportBatchService {
                 linkMap1.put("value", JSON.toJSONString(error));
                 maps.add(linkMap1);
                 customMapper.updateById(api, maps, ids[index]);
-                log.error("Id:{},saveResults: {}",ids[index], error);
+                log.info("Id:{},saveResults: {}",ids[index], error);
             }
             index ++;
         }
@@ -786,6 +826,8 @@ public class DataImportBatchServiceImpl implements DataImportBatchService {
         String api = param.getApi();
         QueryWrapper<DataField> dbQw = new QueryWrapper<>();
         dbQw.eq("api", api);
+        dbQw.and(wrapper -> wrapper.eq("is_updateable", 1)
+                .or().eq("field","Id"));
         List<DataField> list = dataFieldService.list(dbQw);
 
         String beginDateStr = null;
@@ -826,6 +868,8 @@ public class DataImportBatchServiceImpl implements DataImportBatchService {
 
         for (int i = 0; i < page; i++) {
 
+            DataLog dataLog = new DataLog(api, "开始时间："+beginDateStr+";结束时间："+endDateStr, new Date(), null, "数据更新（BULK），查询并组装第" + i + "批数据", "QueryDB");
+
             List<Map<String, Object>> 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);
@@ -900,32 +944,48 @@ public class DataImportBatchServiceImpl implements DataImportBatchService {
                         if (map.get(field) != null && StringUtils.isNotBlank(dataField.getSfType())) {
                             account.put(field, DataUtil.localBulkDataToSfData(dataField.getSfType(), String.valueOf(map.get(field))));
                         }else {
-                            account.put(field, value);
+                            if (map.get(field) == null){
+                                account.put(field, "");
+                            }else {
+                                account.put(field, value);
+                            }
                         }
                     }
                 }
                 
                 if (dataObject.getIsEditable()){
-                    account.put("old_owner_id__c", map.get("OwnerId"));
+                    account.put("old_owner_id__c", map.get("OwnerId") == null?"":map.get("OwnerId"));
                     account.put("old_sfdc_id__c", map.get("Id"));
                 }
                 updateList.add(account);
             }
+            dataLog.setEndTime(new Date());
+            dataLogService.save(dataLog);
 
             JobInfo salesforceUpdateJob = null;
 
             try {
+                DataLog dataLog1 = new DataLog(api, "开始时间："+beginDateStr+";结束时间："+endDateStr, new Date(), null, "数据更新（BULK），更新SF第" + i + "批数据", "UpdateSF");
 
                 salesforceUpdateJob = BulkUtil.createJob(bulkConnection, api, OperationEnum.update);
 
-                BatchInfo batchInfo = CsvConverterUtil.writeToCsvNewOne(updateList, UUID.randomUUID().toString(), list, true, bulkConnection, salesforceUpdateJob);
+                BatchInfo batchInfo = CsvConverterUtil.writeToCsvNewOne(updateList, UUID.randomUUID().toString(), list, dataObject.getIsEditable(), bulkConnection, salesforceUpdateJob);
 
                 BulkUtil.awaitOneCompletion(bulkConnection, salesforceUpdateJob, batchInfo);
 
+                dataLog1.setEndTime(new Date());
+
+                dataLogService.save(dataLog1);
+
+                DataLog dataLog2 = new DataLog(api, "开始时间："+beginDateStr+";结束时间："+endDateStr, new Date(), null, "数据更新（BULK），更新DB第" + i + "批数据", "UpdateDB");
+
                 int updatedCount = checkUpdateResultsOne(bulkConnection, salesforceUpdateJob, batchInfo,api);
 
                 sfNum = sfNum + updatedCount;
 
+                dataLog2.setEndTime(new Date());
+                dataLogService.save(dataLog2);
+
                 TimeUnit.MILLISECONDS.sleep(1);
 
             } catch (InterruptedException interruptedException){
@@ -1100,8 +1160,8 @@ public class DataImportBatchServiceImpl implements DataImportBatchService {
                 update.setDataLock(1);
                 dataObjectService.updateById(update);
 
-                if (api.endsWith("Share") || "GroupMember".equals(api)){
+                if (!dataFieldService.hasCreatedDate(api)){
-                    insertSingleShareData(api,bulkConnection);
+                    insertSingleShareData(api,bulkConnection,update);
                     continue;
                 }
 
@@ -1178,8 +1238,8 @@ public class DataImportBatchServiceImpl implements DataImportBatchService {
                     update.setDataLock(1);
                     dataObjectService.updateById(update);
 
-                    if (api.endsWith("Share") || "GroupMember".equals(api)){
+                    if (dataFieldService.hasCreatedDate(api)){
-                        insertSingleShareData(api,bulkConnection);
+                        insertSingleShareData(api,bulkConnection,dataObject);
                         continue;
                     }
 
@@ -1236,6 +1296,8 @@ public class DataImportBatchServiceImpl implements DataImportBatchService {
         String api = param.getApi();
         QueryWrapper<DataField> dbQw = new QueryWrapper<>();
         dbQw.eq("api", api);
+        dbQw.and(wrapper -> wrapper.eq("is_createable", 1)
+                .or().eq("field","Id"));
         List<DataField> list = dataFieldService.list(dbQw);
         TimeUnit.MILLISECONDS.sleep(1);
 
@@ -1248,8 +1310,11 @@ public class DataImportBatchServiceImpl implements DataImportBatchService {
             endDateStr = DateUtil.format(endDate, "yyyy-MM-dd HH:mm:ss");
         }
 
+        String sql = "where new_id is null and CreatedDate >= '" + beginDateStr + "' and CreatedDate < '" + endDateStr + "'";
+        String sql2 = "new_id is null and CreatedDate >= '" + beginDateStr + "' and CreatedDate < '" + endDateStr + "' order by Id asc limit 8000";
+
         //表内数据总量
-        Integer count = customMapper.countBySQL(api, "where new_id is null and CreatedDate >= '" + beginDateStr + "' and CreatedDate < '" + endDateStr + "'");
+        Integer count = customMapper.countBySQL(api, sql);
         if (count == 0) {
             return;
         }
@@ -1277,7 +1342,9 @@ public class DataImportBatchServiceImpl implements DataImportBatchService {
 
         for (int i = 0; i < page; i++) {
 
-            List<JSONObject> data = customMapper.listJsonObject("*", api, "new_id is null  and CreatedDate >= '" + beginDateStr + "' and CreatedDate < '" + endDateStr + "' limit 8000");
+            DataLog dataLog = new DataLog(api, "开始时间："+beginDateStr+";结束时间："+endDateStr, new Date(), null, "一次性插入（BULK），查询并组装" + i + "批数据", "QueryBD");
+
+            List<JSONObject> data = customMapper.listJsonObject("*", api, sql2);
             int size = data.size();
 
             log.error("总Insert数据 count:{}；当前批次：{}，-开始时间:{}；-结束时间:{}；-api:{}；", count,i, beginDateStr, endDateStr, api);
@@ -1296,9 +1363,12 @@ public class DataImportBatchServiceImpl implements DataImportBatchService {
                 String message = null;
                 for (DataField dataField : list) {
 
-                    if ("Owner_Type".equals(dataField.getField()) || "Id".equals(dataField.getField())){
+                    if ("Owner_Type".equals(dataField.getField())){
                         continue;
                     }
+                    if ("Id".equals(dataField.getField())){
+                        account.put("Id", data.get(j - 1).get(dataField.getField()));
+                    }
                     if ("CreatedDate".equals(dataField.getField()) && dataField.getIsCreateable()){
                         // 转换为UTC时间并格式化
                         LocalDateTime localDateTime = LocalDateTime.parse(String.valueOf(data.get(j - 1).get("CreatedDate")),  inputFormatter);
@@ -1347,7 +1417,11 @@ public class DataImportBatchServiceImpl implements DataImportBatchService {
                             if (data.get(j - 1).get(dataField.getField()) != null && StringUtils.isNotBlank(dataField.getSfType())) {
                                 account.put(dataField.getField(), DataUtil.localBulkDataToSfData(dataField.getSfType(), data.get(j - 1).get(dataField.getField()).toString()));
                             }else {
-                                account.put(dataField.getField(), data.get(j - 1).get(dataField.getField()) );
+                                if (data.get(j - 1).get(dataField.getField()) == null){
+                                    account.put(dataField.getField(), "");
+                                }else {
+                                    account.put(dataField.getField(), data.get(j - 1).get(dataField.getField()) );
+                                }
                             }
                         }
                     }
@@ -1362,10 +1436,9 @@ public class DataImportBatchServiceImpl implements DataImportBatchService {
                     continue;
                 }
                 if (dataObject.getIsEditable()) {
-                    account.put("old_owner_id__c", data.get(j - 1).get("OwnerId"));
+                    account.put("old_owner_id__c", data.get(j - 1).get("OwnerId") == null? "":data.get(j - 1).get("OwnerId"));
                     account.put("old_sfdc_id__c", data.get(j - 1).get("Id"));
                 }
-                ids[j-1] = data.get(j-1).get("Id").toString();
                 insertList.add(account);
                 if (i*8000+j >= count){
                     break;
@@ -1374,15 +1447,30 @@ public class DataImportBatchServiceImpl implements DataImportBatchService {
 
             JobInfo salesforceInsertJob = null;
 
+            dataLog.setEndTime(new Date());
+            dataLogService.save(dataLog);
+
             try {
+                DataLog dataLog1 = new DataLog(api, "开始时间："+beginDateStr+";结束时间："+endDateStr, new Date(), null, "一次性插入（BULK），插入SF第" + i + "批数据", "InsertSF");
+
                 salesforceInsertJob = BulkUtil.createJob(bulkConnection, api, OperationEnum.insert);
 
-                BatchInfo batchInfo = CsvConverterUtil.writeToCsvNewOne(insertList, UUID.randomUUID().toString(), list, true, bulkConnection, salesforceInsertJob);
+                BatchInfo batchInfo = CsvConverterUtil.writeToCsvOne(insertList, UUID.randomUUID().toString(), list, dataObject.getIsEditable(), bulkConnection, salesforceInsertJob,ids);
 
                 BulkUtil.awaitOneCompletion(bulkConnection, salesforceInsertJob, batchInfo);
 
+                dataLog1.setEndTime(new Date());
+
+                dataLogService.save(dataLog1);
+
+                DataLog dataLog2 = new DataLog(api, "开始时间："+beginDateStr+";结束时间："+endDateStr, new Date(), null, "一次性插入（BULK），更新DB第" + i + "批数据", "UpdateDB");
+
                 sfNum = sfNum + checkInsertResultsOne(bulkConnection, salesforceInsertJob, batchInfo, api, ids);
 
+                dataLog2.setEndTime(new Date());
+
+                dataLogService.save(dataLog2);
+
             } catch (Exception e) {
                 log.error("createdNewIdBatch error api:{}", api, e);
                 throw e;
@@ -1410,7 +1498,7 @@ public class DataImportBatchServiceImpl implements DataImportBatchService {
     /**
      * 执行一次性Insert Share数据
      */
-    private void insertSingleShareData(String api, BulkConnection bulkConnection) throws Exception {
+    private void insertSingleShareData(String api, BulkConnection bulkConnection, DataObject dataObject) throws Exception {
 
         QueryWrapper<DataField> dbQw = new QueryWrapper<>();
         dbQw.eq("api", api);
@@ -1420,8 +1508,19 @@ public class DataImportBatchServiceImpl implements DataImportBatchService {
         String beginDateStr = null;
         String endDateStr = null;
 
+        String sql = "";
+        String sql2 = "";
+
+        if (api.endsWith("Share")){
+            sql = "where RowCause NOT IN ('Owner','Rule','Territory','Team','ImplicitChild','ImplicitParent','TerritoryRule','ImplicitCallCenter','PortalRole','Portal','ImplicitPerson','ImplicitGrant') and new_id is null ";
+            sql2 = "RowCause NOT IN ('Owner','Rule','Territory','Team','ImplicitChild','ImplicitParent','TerritoryRule','ImplicitCallCenter','PortalRole','Portal','ImplicitPerson','ImplicitGrant') and new_id is null limit 8000";
+        }else {
+            sql = "where new_id is null ";
+            sql2 = "new_id is null limit 8000";
+        }
+
         //表内数据总量
-        Integer count = customMapper.countBySQL(api, "where new_id is null and RowCause NOT IN ('Owner','Rule','Territory','Team','ImplicitChild','ImplicitParent','TerritoryRule','ImplicitCallCenter','PortalRole','Portal','ImplicitPerson','ImplicitGrant')");
+        Integer count = customMapper.countBySQL(api, sql);
 
         if (count == 0) {
             return;
@@ -1448,7 +1547,9 @@ public class DataImportBatchServiceImpl implements DataImportBatchService {
 
         for (int i = 0; i < page; i++) {
 
-            List<JSONObject> data = customMapper.listJsonObject("*", api, "new_id is null and RowCause NOT IN ('Owner','Rule','Territory','Team','ImplicitChild','ImplicitParent','TerritoryRule','ImplicitCallCenter','PortalRole','Portal','ImplicitPerson','ImplicitGrant') order by Id asc limit " + i * 10000 + ",10000");
+            DataLog dataLog = new DataLog(api, "开始时间："+beginDateStr+";结束时间："+endDateStr, new Date(), null, "一次性插入（BULK），查询并组装" + i + "批数据", "QueryBD");
+
+            List<JSONObject> data = customMapper.listJsonObject("*", api, sql2);
             int size = data.size();
 
             log.error("总Insert数据 count:{}；当前批次：{}，-开始时间:{}；-结束时间:{}；-api:{}；", count,i, beginDateStr, endDateStr, api);
@@ -1463,9 +1564,12 @@ public class DataImportBatchServiceImpl implements DataImportBatchService {
                 String message = null;
 
                 for (DataField dataField : list) {
-                    if ("Owner_Type".equals(dataField.getField()) || "Id".equals(dataField.getField())){
+                    if ("Owner_Type".equals(dataField.getField())){
                         continue;
                     }
+                    if ("Id".equals(dataField.getField())){
+                        account.put("Id", data.get(j - 1).get(dataField.getField()));
+                    }
                     if (dataField.getIsCreateable() !=null && dataField.getIsCreateable()) {
                         if ("reference".equals(dataField.getSfType()) && data.get(j - 1).get(dataField.getField()) != null){
                             //引用类型
@@ -1500,7 +1604,11 @@ public class DataImportBatchServiceImpl implements DataImportBatchService {
                             if (data.get(j - 1).get(dataField.getField()) != null && StringUtils.isNotBlank(dataField.getSfType())) {
                                 account.put(dataField.getField(), DataUtil.localBulkDataToSfData(dataField.getSfType(), data.get(j - 1).get(dataField.getField()).toString()));
                             }else {
-                                account.put(dataField.getField(), data.get(j - 1).get(dataField.getField()) );
+                                if (data.get(j - 1).get(dataField.getField()) == null){
+                                    account.put(dataField.getField(), "");
+                                }else {
+                                    account.put(dataField.getField(), data.get(j - 1).get(dataField.getField()) );
+                                }
                             }
                         }
                     }
@@ -1516,30 +1624,43 @@ public class DataImportBatchServiceImpl implements DataImportBatchService {
                     continue;
                 }
 
-                ids[j-1] = data.get(j-1).get("Id").toString();
                 insertList.add(account);
                 if (i*8000+j == count){
                     break;
                 }
             }
 
+            dataLog.setEndTime(new Date());
+
+            dataLogService.save(dataLog);
+
             JobInfo salesforceInsertJob = null;
 
             try {
+                DataLog dataLog1 = new DataLog(api, "开始时间："+beginDateStr+";结束时间："+endDateStr, new Date(), null, "一次性插入（BULK），插入SF第" + i + "批数据", "InsertSF");
 
                 salesforceInsertJob = BulkUtil.createJob(bulkConnection, api, OperationEnum.insert);
 
-                List<BatchInfo> batchInfos = CsvConverterUtil.writeToCsvNew(insertList, UUID.randomUUID().toString(),list,true,bulkConnection,salesforceInsertJob);
+                BatchInfo batchInfo = CsvConverterUtil.writeToCsvOne(insertList, UUID.randomUUID().toString(), list,dataObject.getIsEditable(), bulkConnection, salesforceInsertJob,ids);
 
-                BulkUtil.awaitCompletion(bulkConnection, salesforceInsertJob, batchInfos);
+                BulkUtil.awaitOneCompletion(bulkConnection, salesforceInsertJob, batchInfo);
 
-                checkInsertResults(bulkConnection, salesforceInsertJob, batchInfos, api, ids);
+                dataLog1.setEndTime(new Date());
+
+                dataLogService.save(dataLog1);
+
+                DataLog dataLog2 = new DataLog(api, "开始时间："+beginDateStr+";结束时间："+endDateStr, new Date(), null, "一次性插入（BULK），更新DB第" + i + "批数据", "UpdateDB");
+
+                checkInsertResultsOne(bulkConnection, salesforceInsertJob, batchInfo, api, ids);
+
+                dataLog2.setEndTime(new Date());
+
+                dataLogService.save(dataLog2);
 
             } catch (Exception e) {
-                log.error("insertSingleShareData error api:{}", api, e);
+                log.error("createdNewIdBatch error api:{}", api, e);
                 throw e;
             }finally {
-
                 BulkUtil.closeJob(bulkConnection, salesforceInsertJob.getId());
             }
         }

src/main/java/com/celnet/datadump/service/impl/DataImportNewServiceImpl.java

@@ -3503,7 +3503,7 @@ public class DataImportNewServiceImpl implements DataImportNewService {
 
                             String value = String.valueOf(data.get(j - 1).get(field));
                             //根据旧sfid查找引用对象新sfid
-                            if (dataField.getIsCreateable() != null && !dataField.getIsCreateable()) {
+                            if (dataField.getIsCreateable() == null || !dataField.getIsCreateable()) {
                                 continue;
                             } else if ("reference".equals(dataField.getSfType()) && data.get(j - 1).get(dataField.getField()) != null) {
 

src/main/java/com/celnet/datadump/util/BulkUtil.java

@@ -165,6 +165,7 @@ public class BulkUtil {
         job.setObject(sobjectType);
         job.setOperation(operation);
         job.setContentType(ContentType.CSV);
+        job.setConcurrencyMode(ConcurrencyMode.Serial);
         job = connection.createJob(job);
         return job;
     }

src/main/java/com/celnet/datadump/util/CsvConverterUtil.java

@@ -192,13 +192,11 @@ public class CsvConverterUtil {
 
         if (isEditable){
             header = Stream.concat(
-                    fields.stream().filter(dataField ->(dataField.getIsCreateable() != null && dataField.getIsCreateable()))
+                    fields.stream().map(DataField::getField),
-                            .map(DataField::getField),
+                    Stream.of("old_owner_id__c", "old_sfdc_id__c")
-                    Stream.of("Id","old_owner_id__c", "old_sfdc_id__c")
             ).toArray(String[]::new);
         }else {
-            header = fields.stream().filter(dataField ->(dataField.getIsCreateable() != null && dataField.getIsCreateable()))
+            header = fields.stream().map(DataField::getField).toArray(String[]::new);
-                    .map(DataField::getField).toArray(String[]::new);
         }
 
         BatchInfo batch = null;
@@ -240,6 +238,81 @@ public class CsvConverterUtil {
         return batch;
     }
 
+    public static BatchInfo writeToCsvOne(List<JSONObject> jsonList,
+                                             String fileName,
+                                             List<DataField> fields,
+                                             Boolean isEditable,
+                                             BulkConnection connection,
+                                             JobInfo jobInfo,String[] ids) {
+
+        // 1. 创建目标目录（不存在则创建）
+        File targetDir = FileUtil.mkdir("data-dump/dataFile");
+
+        // 2. 构建完整文件路径
+        String fullPath = targetDir.getAbsolutePath()  + File.separator  + fileName + ".csv";
+
+        CsvWriter csvWriter = CsvUtil.getWriter(fullPath, CharsetUtil.CHARSET_UTF_8, false);
+
+        String[] header = null;
+
+        File file = null;
+
+        if (isEditable){
+            header = Stream.concat(
+                    fields.stream().filter(dataField ->(dataField.getIsCreateable() != null && dataField.getIsCreateable()))
+                            .map(DataField::getField),
+                    Stream.of("old_owner_id__c", "old_sfdc_id__c")
+            ).toArray(String[]::new);
+        }else {
+            header = fields.stream().filter(dataField ->(dataField.getIsCreateable() != null && dataField.getIsCreateable()))
+                    .map(DataField::getField).toArray(String[]::new);
+        }
+
+        BatchInfo batch = null;
+
+        try {
+
+            // 2. 写入表头（必须使用 String[]）
+            csvWriter.writeHeaderLine(header);
+
+            int index = 0;
+
+            // 遍历数据列表
+            for (JSONObject jsonObject : jsonList) {
+                // 按表头顺序获取值
+                String[] row = new String[header.length];
+                for (int i = 0; i < header.length;  i++) {
+                    // 将每个值转换为字符串（如果为null则转为空字符串）
+                    Object value = jsonObject.get(header[i]);
+                    row[i] = String.valueOf(value);
+                }
+                ids[index] = jsonObject.get("Id").toString();
+
+                csvWriter.writeLine(row);
+
+                index ++;
+            }
+
+            // 关闭writer（在try-with-resources中可省略，但这里我们显式关闭）
+            csvWriter.close();
+
+            file = new File(fullPath);
+
+            InputStream inputStream = Files.newInputStream(file.toPath());
+
+            batch = connection.createBatchFromStream(jobInfo, inputStream);
+
+        } catch (IOException e) {
+            log.error("CSV文件操作失败: {}", fullPath, e);
+            file.deleteOnExit();
+        } catch (AsyncApiException e) {
+            log.error("Bulk API上传失败: {}", e.getMessage(), e);
+            file.deleteOnExit();
+        }
+
+        return batch;
+    }
+
     public static String exportToCsv(List<Map<String, Object>> data, String fileName) throws IOException {
         // 1. 创建目标目录（不存在则创建）
         File targetDir = FileUtil.mkdir("data-dump/dataFile");

src/main/java/com/celnet/datadump/util/DataUtil.java

@@ -511,6 +511,7 @@ public class DataUtil {
 
     public static Object localDataToSfData(String fieldType, String data) throws ParseException {
         SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss'Z'");
+        sdf.setTimeZone(TimeZone.getTimeZone("UTC"));
         SimpleDateFormat sd = new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss");
         Date date;
         //date转Calendar类型

src/main/resources/mapper/SalesforceMapper.xml

@@ -238,4 +238,43 @@
         </if>
     </select>
 
+
+    <select id="listOrderByIdShare">
+        select ${param.select} from ${param.api}
+        <where>
+            <if test="param != null">
+                <if test="param.maxId != null">
+                    AND id > #{param.maxId}
+                </if>
+                <if test="param.isDeleted != null">
+                    AND IsDeleted = #{param.isDeleted}
+                </if>
+                <if test="param.sql != null">
+                    AND ${param.sql}
+                </if>
+                <!-- 判断 param.api 是否以 "Share" 结尾 -->
+                <if test="param.api != null and param.api.endsWith('Share')">
+                    AND RowCause NOT IN (
+                    'Owner', 'Rule', 'Territory', 'Team',
+                    'ImplicitChild', 'ImplicitParent', 'TerritoryRule',
+                    'ImplicitCallCenter', 'PortalRole', 'Portal',
+                    'ImplicitPerson', 'ImplicitGrant'
+                    )
+                </if>
+                <if test="param.beginModifyDate">
+                    AND #{param.updateField} >= #{param.beginModifyDate}
+                </if>
+
+            </if>
+        </where>
+        <if test="param != null">
+            <if test="param.idField != null">
+                order by ${param.idField} asc
+            </if>
+        </if>
+        <if test="param != null and param.limit != null">
+            limit #{param.limit}
+        </if>
+    </select>
+
 </mapper>