【feat】 20250805大版本（bulk相关）调整

src/main/java/com/celnet/datadump/job/DataDumpNewJob.java

@@ -131,6 +131,26 @@ public class DataDumpNewJob {
         return dataImportNewService.immigrationUpdateNew(param);
     }
 
+    /**
+     * 一次性插入数据
+     * @param paramStr 参数json
+     * @return result
+     */
+    @XxlJob("insertSingleJob")
+    public ReturnT<String> insertSingleJob(String paramStr) throws Exception {
+        log.info("insertSingleJob 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 dataImportNewService.insertSingle(param);
+    }
+
     /**
      * 拉取文件关联表
      * @return result
@@ -195,7 +215,7 @@ public class DataDumpNewJob {
     }
 
     /**
-     * 一次性插入数据
+     * 一次性插入数据（大数据量）
      * @param paramStr
      * @author kris
      * @return

src/main/java/com/celnet/datadump/service/DataImportNewService.java

@@ -49,4 +49,8 @@ public interface DataImportNewService {
 
     ReturnT<String> checkDeletedData(SalesforceParam param) throws Exception;
 
+    /**
+     * 一次性插入数据
+     */
+    ReturnT<String> insertSingle(SalesforceParam param) throws Exception;
 }

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

@@ -24,6 +24,7 @@ import com.sforce.soap.partner.*;
 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;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.collections.CollectionUtils;
 import org.apache.commons.lang3.ArrayUtils;
@@ -417,6 +418,10 @@ public class CommonServiceImpl implements CommonService {
                     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)) {
@@ -512,7 +517,9 @@ public class CommonServiceImpl implements CommonService {
             boolean isSuccess = false;
             while (failCount <= Const.MAX_FAIL_COUNT) {
                 try {
+                    if (!"FeedItem".equals(api) && !"ContentFolderMember ".equals(api)){
                         dataBatchHistory.setSfNum(countSfNum(connect, param));
+                    }
                     isSuccess = true;
                     break;
                 } catch (Throwable throwable) {

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

@@ -35,8 +35,10 @@ import org.apache.commons.lang3.StringUtils;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Service;
 
-import java.io.File;
+import java.io.*;
-import java.io.IOException;
+import java.net.HttpURLConnection;
+import java.net.URL;
+import java.net.URLEncoder;
 import java.nio.file.Files;
 import java.nio.file.Paths;
 import java.text.ParseException;
@@ -284,6 +286,9 @@ public class DataImportBatchServiceImpl implements DataImportBatchService {
         QueryWrapper<DataField> dbQw = new QueryWrapper<>();
         dbQw.eq("api", api);
         List<DataField> list = dataFieldService.list(dbQw);
+        QueryWrapper<DataObject> obQw = new QueryWrapper<>();
+        obQw.eq("name", api);
+        DataObject dataObject = dataObjectService.getOne(obQw);
 
         String beginDateStr = null;
         String endDateStr = null;
@@ -372,7 +377,6 @@ public class DataImportBatchServiceImpl implements DataImportBatchService {
                             if (reference == null){
                                 continue;
                             }
-//                            log.info("----------" + dataField.getField() + "的引用类型 ------------" + reference);
                             List<Map<String, Object>> referenceMap = customMapper.list("new_id", reference, "new_id is not null limit 1");
                             if (referenceMap.isEmpty()){
                                 QueryWrapper<DataObject> maxIndex = new QueryWrapper<>();
@@ -401,6 +405,8 @@ public class DataImportBatchServiceImpl implements DataImportBatchService {
                     }
 
                 }
+                account.put("old_owner_id__c", 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);
@@ -408,27 +414,59 @@ public class DataImportBatchServiceImpl implements DataImportBatchService {
                     break;
                 }
             }
+            JobInfo salesforceInsertJob = null;
+
+            JobInfo salesforceQueryJob = null;
+
+            String fullPath = null;
 
             try {
 
                 //写入csv文件
-                String fullPath = CsvConverterUtil.writeToCsv(insertList, UUID.randomUUID().toString(),list,false);
+                fullPath = CsvConverterUtil.writeToCsv(insertList, UUID.randomUUID().toString(),list,dataObject.getIsEditable());
 
-                JobInfo salesforceInsertJob = BulkUtil.createJob(bulkConnection, api, OperationEnum.insert);
+                salesforceInsertJob = BulkUtil.createJob(bulkConnection, api, OperationEnum.insert);
 
                 List<BatchInfo> batchInfos = BulkUtil.createBatchesFromCSVFile(bulkConnection, salesforceInsertJob, fullPath);
 
                 BulkUtil.awaitCompletion(bulkConnection, salesforceInsertJob, batchInfos);
 
-                sfNum = sfNum + checkInsertResults(bulkConnection, salesforceInsertJob, batchInfos, api, ids);
+                List<String> resultIds = checkInsertResults(bulkConnection, salesforceInsertJob, batchInfos, api, ids);
+
+                String idStr = "(";
+                for (String id : resultIds) {
+                    idStr += "'" + id + "',"; // 拼接每个ID
+                }
+                if (idStr.endsWith(",")) { // 如果最后一个字符是逗号，说明循环正常结束
+                    idStr = idStr.substring(0, idStr.length() - 1); // 去掉最后一个多余的逗号
+                }
+                idStr += ")"; // 添加右括号
+
+                String query = "SELECT Id,old_sfdc_id__c FROM " + api + " where Id in " + idStr;
+
+                salesforceQueryJob = BulkUtil.createJob(bulkConnection, api, OperationEnum.queryAll);
+
+                ByteArrayInputStream bout = new ByteArrayInputStream(query.getBytes());
+
+                BatchInfo info = bulkConnection.createBatchFromStream(salesforceQueryJob, bout);
+
+                List<BatchInfo> batchQueryInfos = new ArrayList<>();
+
+                batchQueryInfos.add(info);
+
+                BulkUtil.awaitCompletion(bulkConnection, salesforceQueryJob, batchQueryInfos);
+
+                sfNum = sfNum + checkQueryResultsNew(bulkConnection, salesforceQueryJob, info, api);
+
+            } catch (Exception e) {
+                log.error("createdNewIdBatch error api:{}", api, e);
+                throw e;
+            }finally {
+                new File(fullPath).delete();
 
                 BulkUtil.closeJob(bulkConnection, salesforceInsertJob.getId());
 
-                new File(fullPath).delete();
+                BulkUtil.closeJob(bulkConnection, salesforceQueryJob.getId());
-
-            } catch (Exception e) {
-                log.error("manualCreatedNewId error api:{}", api, e);
-                throw e;
             }
         }
 
@@ -451,10 +489,10 @@ public class DataImportBatchServiceImpl implements DataImportBatchService {
     /**
      * 读写Insert结果
      */
-    public int checkInsertResults(BulkConnection connection, JobInfo job,
+    public List<String> checkInsertResults(BulkConnection connection, JobInfo job,
                                    List<BatchInfo> batchInfoList,String api,String[] ids)
             throws AsyncApiException, IOException {
-        int index = 0;
+        ArrayList<String> strings = new ArrayList<>();
         // batchInfoList was populated when batches were created and submitted
         for (BatchInfo b : batchInfoList) {
             CSVReader rdr =
@@ -464,7 +502,7 @@ public class DataImportBatchServiceImpl implements DataImportBatchService {
 
             List<String> row;
             while ((row = rdr.nextRecord()) != null) {
-                Map<String, String> resultInfo = new HashMap<String, String>();
+                Map<String, String> resultInfo = new HashMap<>();
                 for (int i = 0; i < resultCols; i++) {
                     resultInfo.put(resultHeader.get(i), row.get(i));
                 }
@@ -473,20 +511,47 @@ public class DataImportBatchServiceImpl implements DataImportBatchService {
                 String id = resultInfo.get("Id");
                 String error = resultInfo.get("Error");
                 if (insertStatus && created) {
-                    List<Map<String, Object>> maps = new ArrayList<>();
+                    strings.add(id);
-                    Map<String, Object> m = new HashMap<>();
-                    m.put("key", "new_id");
-                    m.put("value", id);
-                    maps.add(m);
-                    customMapper.updateById(api, maps, ids[index]);
-                    index ++;
                     log.info("Created Success row with id " + id);
                 } else if (!insertStatus) {
                     log.error("Created Fail with error: " + error);
                 }
             }
         }
-        return index;
+        return strings;
+    }
+
+    public Integer checkQueryResultsNew(BulkConnection connection, JobInfo job, BatchInfo batchInfo,String api) throws AsyncApiException, IOException {
+        int num = 0;
+        // 获取查询结果的 ResultId (对于Query类型，每个Batch可能包含多个ResultId)
+        QueryResultList resultList = connection.getQueryResultList(job.getId(), batchInfo.getId());
+
+        // 遍历每个 ResultId
+        for (String resultId : resultList.getResult()) {
+            InputStream resultStream = connection.getQueryResultStream(job.getId(), batchInfo.getId(), resultId);
+
+            // 使用 CSVReader 读取结果
+            CSVReader rdr = new CSVReader(resultStream);
+            List<String> header = rdr.nextRecord();
+            List<String> row;
+
+            while ((row = rdr.nextRecord()) != null) {
+                Map<String, String> resultInfo = new HashMap<>();
+                for (int i = 0; i < header.size(); i++) {
+                    resultInfo.put(header.get(i), row.get(i));
+                    String id = resultInfo.get("Id");
+                    String oldSfdcId = resultInfo.get("old_sfdc_id__c");
+                    List<Map<String, Object>> maps = new ArrayList<>();
+                    Map<String, Object> m = new HashMap<>();
+                    m.put("key", "new_id");
+                    m.put("value", id);
+                    maps.add(m);
+                    num ++;
+                    customMapper.updateById(api, maps, oldSfdcId);
+                }
+            }
+        }
+        return num;
     }
 
     /**
@@ -636,7 +701,6 @@ public class DataImportBatchServiceImpl implements DataImportBatchService {
             }
 
             for (DataObject object : dataObjects) {
-
                 try {
                     TimeUnit.MILLISECONDS.sleep(1);
                     List<SalesforceParam> salesforceParams = null;
@@ -703,7 +767,6 @@ public class DataImportBatchServiceImpl implements DataImportBatchService {
         }
         // 表内数据总量
         Integer count = customMapper.countBySQL(api, "where new_id is not null and CreatedDate >= '" + beginDateStr + "' and CreatedDate < '" + endDateStr + "'");
-        log.error("总Update数据 count:{}；-开始时间:{}；-结束时间:{}；-api:{}；", count, beginDateStr, endDateStr, api);
         if(count == 0){
             return;
         }
@@ -722,11 +785,13 @@ public class DataImportBatchServiceImpl implements DataImportBatchService {
                     ));
         }
 
-
         // 总更新数
         int sfNum = 0;
         // 批量更新10000一次
         int page = count%10000 == 0 ? count/10000 : (count/10000) + 1;
+
+        log.error("总Update数据 count:{}；总批次:{}；-开始时间:{}；-结束时间:{}；-api:{}；", count,page, beginDateStr, endDateStr, api);
+
         for (int i = 0; i < page; i++) {
             List<Map<String, Object>> mapList = customMapper.list("*", api, "new_id is not null and CreatedDate >= '" + beginDateStr + "' and CreatedDate < '" + endDateStr + "' order by Id asc limit " + i * 10000 + ",10000");
 
@@ -760,7 +825,6 @@ public class DataImportBatchServiceImpl implements DataImportBatchService {
                             if (reference_to == null){
                                 continue;
                             }
-//                            log.info("----------" + dataField.getField() + "的引用类型 ------------" + reference_to);
                             //判断reference_to内是否包含User字符串
                             if (reference_to.contains(",User") || reference_to.contains("User,")) {
                                 reference_to = "User";
@@ -918,16 +982,10 @@ public class DataImportBatchServiceImpl implements DataImportBatchService {
             TimeUnit.MILLISECONDS.sleep(1);
             try {
                 List<SalesforceParam> salesforceParams = null;
-
                 update.setName(api);
                 update.setDataLock(1);
                 dataObjectService.updateById(update);
 
-                if (api.endsWith("Share")){
-                    insertSingleShareData(api,bulkConnection);
-                    continue;
-                }
-
                 QueryWrapper<DataBatch> dbQw = new QueryWrapper<>();
                 dbQw.eq("name", api);
                 List<DataBatch> list = dataBatchService.list(dbQw);
@@ -1001,11 +1059,6 @@ public class DataImportBatchServiceImpl implements DataImportBatchService {
                     update.setDataLock(1);
                     dataObjectService.updateById(update);
 
-                    if (api.endsWith("Share")){
-                        insertSingleShareData(api,bulkConnection);
-                        continue;
-                    }
-
                     List<SalesforceParam> salesforceParams = null;
                     QueryWrapper<DataBatch> dbQw = new QueryWrapper<>();
                     dbQw.eq("name", api);
@@ -1061,9 +1114,6 @@ public class DataImportBatchServiceImpl implements DataImportBatchService {
         dbQw.eq("api", api);
         List<DataField> list = dataFieldService.list(dbQw);
         TimeUnit.MILLISECONDS.sleep(1);
-        QueryWrapper<DataObject> obQw = new QueryWrapper<>();
-        obQw.eq("name", api);
-        DataObject dataObject = dataObjectService.getOne(obQw);
 
         String beginDateStr = null;
         String endDateStr = null;
@@ -1076,14 +1126,17 @@ public class DataImportBatchServiceImpl implements DataImportBatchService {
 
         //表内数据总量
         Integer count = customMapper.countBySQL(api, "where new_id is null and CreatedDate >= '" + beginDateStr + "' and CreatedDate < '" + endDateStr + "'");
-        log.error("总Insert数据 count:{}；-开始时间:{}；-结束时间:{}；-api:{}；", count, beginDateStr, endDateStr, api);
         if (count == 0) {
             return;
         }
         //批量插入10000一次
         int page = count%10000 == 0 ? count/10000 : (count/10000) + 1;
+
+        log.error("总Insert数据 count:{}；批次：{}，-开始时间:{}；-结束时间:{}；-api:{}；", count,page, beginDateStr, endDateStr, api);
+
         //总插入数
         int sfNum = 0;
+
         for (int i = 0; i < page; i++) {
 
             List<JSONObject> data = customMapper.listJsonObject("*", api, "new_id is null  and CreatedDate >= '" + beginDateStr + "' and CreatedDate < '" + endDateStr + "' limit 10000");
@@ -1106,10 +1159,6 @@ public class DataImportBatchServiceImpl implements DataImportBatchService {
                         ));
             }
 
-            //判断引用对象是否存在new_id
-            DataObject update = new DataObject();
-            update.setName(api);
-
             //更新对象的new_id
             String[] ids = new String[size];
 
@@ -1162,9 +1211,7 @@ public class DataImportBatchServiceImpl implements DataImportBatchService {
                                 account.put(dataField.getField(), m.get("new_id"));
                             }else {
                                 String message = "对象类型:" + api + "的数据:"+ data.get(j - 1).get("Id") +"的引用对象:" + reference_to + "的数据:"+ data.get(j - 1).get(dataField.getField()) +"不存在！";
-                                EmailUtil.send("DataDump ERROR", message);
                                 log.info(message);
-                                return;
                             }
                         }else {
                             if (data.get(j - 1).get(dataField.getField()) != null && StringUtils.isNotBlank(dataField.getSfType())) {
@@ -1177,38 +1224,69 @@ public class DataImportBatchServiceImpl implements DataImportBatchService {
 
                 }
 
-                if (dataObject.getIsEditable()){
                 account.put("old_owner_id__c", 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*10000+j == count){
+                if (i*1000+j == count){
                     break;
                 }
             }
 
+            JobInfo salesforceInsertJob = null;
+
+            JobInfo salesforceQueryJob = null;
+
+            String fullPath = null;
+
             try {
 
                 //写入csv文件
-                String fullPath = CsvConverterUtil.writeToCsv(insertList, UUID.randomUUID().toString(),list,dataObject.getIsEditable());
+                fullPath = CsvConverterUtil.writeToCsv(insertList, UUID.randomUUID().toString(),list,true);
 
-                JobInfo salesforceInsertJob = BulkUtil.createJob(bulkConnection, api, OperationEnum.insert);
+                salesforceInsertJob = BulkUtil.createJob(bulkConnection, api, OperationEnum.insert);
 
                 List<BatchInfo> batchInfos = BulkUtil.createBatchesFromCSVFile(bulkConnection, salesforceInsertJob, fullPath);
 
                 BulkUtil.awaitCompletion(bulkConnection, salesforceInsertJob, batchInfos);
 
-                sfNum = sfNum + checkInsertResults(bulkConnection, salesforceInsertJob, batchInfos, api, ids);
+                List<String> resultIds = checkInsertResults(bulkConnection, salesforceInsertJob, batchInfos, api, ids);
 
-                BulkUtil.closeJob(bulkConnection, salesforceInsertJob.getId());
+                String idStr = "(";
+                for (String id : resultIds) {
+                    idStr += "'" + id + "',"; // 拼接每个ID
+                }
+                if (idStr.endsWith(",")) { // 如果最后一个字符是逗号，说明循环正常结束
+                    idStr = idStr.substring(0, idStr.length() - 1); // 去掉最后一个多余的逗号
+                }
+                idStr += ")"; // 添加右括号
 
-                new File(fullPath).delete();
+                String query = "SELECT Id,old_sfdc_id__c FROM " + api + " where Id in " + idStr;
+
+                salesforceQueryJob = BulkUtil.createJob(bulkConnection, api, OperationEnum.queryAll);
+
+                ByteArrayInputStream bout = new ByteArrayInputStream(query.getBytes());
+
+                BatchInfo info = bulkConnection.createBatchFromStream(salesforceQueryJob, bout);
+
+                List<BatchInfo> batchQueryInfos = new ArrayList<>();
+
+                batchQueryInfos.add(info);
+
+                BulkUtil.awaitCompletion(bulkConnection, salesforceQueryJob, batchQueryInfos);
+
+                sfNum = sfNum + checkQueryResultsNew(bulkConnection, salesforceQueryJob, info, api);
 
             } catch (Exception e) {
                 log.error("insertSingleData error api:{}", api, e);
                 throw e;
+            }finally {
+                new File(fullPath).delete();
+
+                BulkUtil.closeJob(bulkConnection, salesforceInsertJob.getId());
+
+                BulkUtil.closeJob(bulkConnection, salesforceQueryJob.getId());
             }
         }
 
@@ -1228,135 +1306,5 @@ public class DataImportBatchServiceImpl implements DataImportBatchService {
 
     }
 
-    /**
-     * 执行一次性Insert Share数据
-     */
-    private void insertSingleShareData(String api, BulkConnection bulkConnection) throws Exception {
-
-        QueryWrapper<DataField> dbQw = new QueryWrapper<>();
-        dbQw.eq("api", api);
-        List<DataField> list = dataFieldService.list(dbQw);
-        TimeUnit.MILLISECONDS.sleep(1);
-
-        String beginDateStr = null;
-        String endDateStr = null;
-
-        //表内数据总量
-        Integer count = customMapper.countBySQL(api, "where new_id is null and RowCause = 'Manual'");
-        log.error("总Insert数据 count:{}；-开始时间:{}；-结束时间:{}；-api:{}；", count, beginDateStr, endDateStr, api);
-        if (count == 0) {
-            return;
-        }
-        //批量插入10000一次
-        int page = count%10000 == 0 ? count/10000 : (count/10000) + 1;
-        //总插入数
-        int sfNum = 0;
-        for (int i = 0; i < page; i++) {
-
-            List<JSONObject> data = customMapper.listJsonObject("*", api, "new_id is null and RowCause = 'Manual' limit 10000");
-            int size = data.size();
-
-            log.info("执行api:{}, 执行page:{}, 执行size:{}", api, i+1, size);
-            List<JSONObject> insertList = new ArrayList<>();
-
-            //查询当前对象多态字段映射
-            QueryWrapper<LinkConfig> queryWrapper = new QueryWrapper<>();
-            queryWrapper.eq("api",api).eq("is_create",1).eq("is_link",true);
-            List<LinkConfig> configs = linkConfigService.list(queryWrapper);
-            Map<String, String> fieldMap = new HashMap<>();
-
-            if (!configs.isEmpty())  {
-                fieldMap = configs.stream()
-                        .collect(Collectors.toMap(
-                                LinkConfig::getField,      // Key提取器
-                                LinkConfig::getLinkField,  // Value提取器
-                                (oldVal, newVal) -> newVal // 解决重复Key冲突（保留新值）
-                        ));
-            }
-
-            //判断引用对象是否存在new_id
-            DataObject update = new DataObject();
-            update.setName(api);
-
-            //更新对象的new_id
-            String[] ids = new String[size];
-
-            for (int j = 1; j <= size; j++) {
-                JSONObject account = new JSONObject();
-                for (DataField dataField : list) {
-
-                    if ("Owner_Type".equals(dataField.getField()) || "Id".equals(dataField.getField())){
-                        continue;
-                    }
-
-                    if (dataField.getIsCreateable() !=null && dataField.getIsCreateable()) {
-                        if ("reference".equals(dataField.getSfType()) && data.get(j - 1).get(dataField.getField()) != null){
-                            //引用类型
-                            String reference_to = dataField.getReferenceTo();
-                            //引用类型字段
-                            String linkfield = fieldMap.get(dataField.getField());
-
-                            if (StringUtils.isNotBlank(linkfield)){
-                                reference_to = data.get(j-1).get(linkfield)!=null?data.get(j-1).get(linkfield).toString():null;
-                            }
-
-                            if (reference_to == null){
-                                continue;
-                            }
-
-                            if (reference_to.contains(",User") || reference_to.contains("User,")) {
-                                reference_to = "User";
-                            }
-                            Map<String, Object> m = customMapper.getById("new_id", reference_to, data.get(j - 1).get(dataField.getField()).toString());
-                            if (m != null && !m.isEmpty()) {
-                                account.put(dataField.getField(), m.get("new_id"));
-                            }else {
-                                String message = "对象类型:" + api + "的数据:"+ data.get(j - 1).get("Id") +"的引用对象:" + reference_to + "的数据:"+ data.get(j - 1).get(dataField.getField()) +"不存在！";
-                                EmailUtil.send("DataDump ERROR", message);
-                                log.info(message);
-                                return;
-                            }
-                        }else {
-                            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()) );
-                            }
-                        }
-                    }
-
-                }
-
-                ids[j-1] = data.get(j-1).get("Id").toString();
-                insertList.add(account);
-                if (i*10000+j == count){
-                    break;
-                }
-            }
-
-            try {
-
-                //写入csv文件
-                String fullPath = CsvConverterUtil.writeToCsv(insertList, UUID.randomUUID().toString(),list,false);
-
-                JobInfo salesforceInsertJob = BulkUtil.createJob(bulkConnection, api, OperationEnum.insert);
-
-                List<BatchInfo> batchInfos = BulkUtil.createBatchesFromCSVFile(bulkConnection, salesforceInsertJob, fullPath);
-
-                BulkUtil.awaitCompletion(bulkConnection, salesforceInsertJob, batchInfos);
-
-                sfNum = sfNum + checkInsertResults(bulkConnection, salesforceInsertJob, batchInfos, api, ids);
-
-                BulkUtil.closeJob(bulkConnection, salesforceInsertJob.getId());
-
-                new File(fullPath).delete();
-
-            } catch (Exception e) {
-                log.error("insertSingleShareData error api:{}", api, e);
-                throw e;
-            }
-        }
-
-    }
 
 }

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

@@ -50,6 +50,7 @@ import java.io.*;
 import java.text.ParseException;
 import java.text.SimpleDateFormat;
 import java.util.*;
+import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.Future;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
@@ -210,13 +211,13 @@ public class DataImportNewServiceImpl implements DataImportNewService {
         //表内数据总量
         Integer count = customMapper.countBySQL(api, "where new_id is null and IsPersonAccount = 1 and CreatedDate >= '" + beginDateStr + "' and CreatedDate < '" + endDateStr + "'");
 
-        log.info("总Insert数据 count:{}；-开始时间:{}；-结束时间:{}；-api:{}；", count, beginDateStr, endDateStr, api);
         if (count == 0) {
             return;
         }
 
         //批量插入200一次
         int page = count%200 == 0 ? count/200 : (count/200) + 1;
+        log.info("总Insert数据 count:{}；总批次：{}；-开始时间:{}；-结束时间:{}；-api:{}；", count,page, beginDateStr, endDateStr, api);
 
         for (int i = 0; i < page; i++) {
             List<Map<String, Object>> data = customMapper.list("*", api, "new_id is null and IsPersonAccount = 1 and CreatedDate >= '" + beginDateStr + "' and CreatedDate < '" + endDateStr + "' limit 200");
@@ -459,31 +460,16 @@ public class DataImportNewServiceImpl implements DataImportNewService {
         for (String api : apis) {
             DataObject update = dataObjectService.getById(api);
             try {
-                QueryWrapper<DataObject> qw = new QueryWrapper<>();
-                qw.eq("name", api);
-                DataObject dataObject = dataObjectService.getOne(qw);
-
-                List<SalesforceParam> salesforceParams = null;
                 QueryWrapper<DataBatch> dbQw = new QueryWrapper<>();
-                dbQw.eq("name", api);
+                dbQw.eq("name", api).orderByDesc("sync_end_date").last("LIMIT 1");
-                List<DataBatch> list = dataBatchService.list(dbQw);
+                DataBatch t = dataBatchService.getOne(dbQw);
-                AtomicInteger batch = new AtomicInteger(1);
-                if (CollectionUtils.isNotEmpty(list)) {
-                    salesforceParams = list.stream().map(t -> {
-                        if(t.getSyncEndDate().toInstant().isAfter(t.getLastModifiedDate().toInstant())){
                 SalesforceParam salesforceParam = param.clone();
                 salesforceParam.setApi(t.getName());
                 salesforceParam.setBeginCreateDate(t.getSyncStartDate());
                 salesforceParam.setEndCreateDate(t.getSyncEndDate());
-                            salesforceParam.setBatch(batch.getAndIncrement());
+                salesforceParam.setBatch(new AtomicInteger(1).getAndIncrement());
-                            return salesforceParam;
-                        }
-                        return null;
-                    }).collect(Collectors.toList());
-                }
 
                 // 手动任务优先执行
-                for (SalesforceParam salesforceParam : salesforceParams) {
                 Future<?> future = salesforceExecutor.execute(() -> {
                     try {
                         UpdateSfDataNew(salesforceParam, partnerConnection,update);
@@ -493,21 +479,23 @@ public class DataImportNewServiceImpl implements DataImportNewService {
                     }
                 }, salesforceParam.getBatch(), 1);
                 futures.add(future);
-                }
+
-                // 等待当前所有线程执行完成
+                update.setNeedUpdate(false);
-                salesforceExecutor.waitForFutures(futures.toArray(new Future<?>[]{}));
 
             } catch (Exception e) {
                 throw e;
             } finally {
                 if (isFull) {
-                    update.setNeedUpdate(false);
                     update.setName(api);
                     update.setDataLock(0);
                     dataObjectService.updateById(update);
                 }
             }
         }
+        // 等待当前所有线程执行完成
+        salesforceExecutor.waitForFutures(futures.toArray(new Future<?>[]{}));
+        futures.clear();
+
         return ReturnT.SUCCESS;
     }
 
@@ -654,27 +642,17 @@ public class DataImportNewServiceImpl implements DataImportNewService {
                 TimeUnit.MILLISECONDS.sleep(1);
                 try {
                     String api = dataObject.getName();
-                    List<SalesforceParam> salesforceParams = null;
                     QueryWrapper<DataBatch> dbQw = new QueryWrapper<>();
-                    dbQw.eq("name", api);
+                    dbQw.eq("name", api).orderByDesc("sync_end_date").last("LIMIT 1");
-                    List<DataBatch> list = dataBatchService.list(dbQw);
+                    DataBatch t = dataBatchService.getOne(dbQw);
+
                     AtomicInteger batch = new AtomicInteger(1);
-                    if (CollectionUtils.isNotEmpty(list)) {
-                        salesforceParams = list.stream().map(t -> {
-                            if(t.getSyncEndDate().toInstant().isAfter(t.getLastModifiedDate().toInstant())){
                     SalesforceParam salesforceParam = param.clone();
                     salesforceParam.setApi(t.getName());
                     salesforceParam.setBeginCreateDate(t.getSyncStartDate());
                     salesforceParam.setEndCreateDate(t.getSyncEndDate());
                     salesforceParam.setBatch(batch.getAndIncrement());
-                                return salesforceParam;
-                            }
-                            return null;
-                        }).collect(Collectors.toList());
-                    }
 
-                    // 手动任务优先执行
-                    for (SalesforceParam salesforceParam : salesforceParams) {
                     Future<?> future = salesforceExecutor.execute(() -> {
                         try {
                             UpdateSfDataNew(salesforceParam, partnerConnection,dataObject);
@@ -684,9 +662,6 @@ public class DataImportNewServiceImpl implements DataImportNewService {
                         }
                     }, salesforceParam.getBatch(), 0);
                     futures.add(future);
-                    }
-                    // 等待当前所有线程执行完成
-                    salesforceExecutor.waitForFutures(futures.toArray(new Future<?>[]{}));
                 } catch (Exception e) {
                     throw e;
                 } finally {
@@ -744,7 +719,6 @@ public class DataImportNewServiceImpl implements DataImportNewService {
         }
         //表内数据总量
         Integer count = customMapper.countBySQL(api, sql);
-        log.info("总Update数据 count:{}；-开始时间:{}；-结束时间:{}；-api:{}；", count, beginDateStr, endDateStr, api);
 
         if(count == 0){
             return;
@@ -768,6 +742,9 @@ public class DataImportNewServiceImpl implements DataImportNewService {
         int targetCount = 0;
         //批量插入200一次
         int page = count%200 == 0 ? count/200 : (count/200) + 1;
+
+        log.info("总Update数据 count:{}；总批次：{}；-开始时间:{}；-结束时间:{}；-api:{}；", count, page,beginDateStr, endDateStr, api);
+
         for (int i = 0; i < page; i++) {
             List<Map<String, Object>> mapList = customMapper.list("*", api,  sql2+ i * 200 + ",200");
             SObject[] accounts = new SObject[mapList.size()];
@@ -1162,13 +1139,14 @@ public class DataImportNewServiceImpl implements DataImportNewService {
 
         Integer count = customMapper.countBySQL(api, "where is_dump = 0 and CreatedDate >= '" + beginDateStr + "' and CreatedDate < '" + endDateStr +"'"+ extraSql );
 
-        log.info("api：{}；总文件数 count:{}；-开始时间:{}；-结束时间:{}；",api, count, beginDateStr, endDateStr);
-
         if (count == 0) {
             return;
         }
 
         int page = count%1000 == 0 ? count/1000 : (count/1000) + 1;
+
+        log.info("api：{}；总文件数 count:{}；总批次 ：{}；-开始时间:{}；-结束时间:{}；",api, count, page,beginDateStr, endDateStr);
+
         //总插入数
         for (int i = 0; i < page; i++) {
 
@@ -1178,6 +1156,7 @@ public class DataImportNewServiceImpl implements DataImportNewService {
             List<Map<String, Object>> list = customMapper.list("Id, " + name, api, " is_dump = 0 and CreatedDate >= '" + beginDateStr + "' and CreatedDate < '" + endDateStr +"'"+ extraSql + "limit " + i * 1000 + ",1000");
 
             for (Map<String, Object> map : list) {
+                int failCount = 0;
                 String id = null;
                 // 上传完毕 更新附件信息
                 List<Map<String, Object>> maps = Lists.newArrayList();
@@ -1189,25 +1168,18 @@ public class DataImportNewServiceImpl implements DataImportNewService {
                     }else {
                         fileName = (String) map.get("Name");
                     }
-//                    log.info("------------文件名:" + id + "_" + fileName);
                     // 判断路径是否为空
                     if (StringUtils.isNotBlank(fileName)) {
                         String filePath = api + "/" + id + "_" + fileName;
                         // 拼接url
                         String url = downloadUrl + String.format(Const.SF_FILE_URL, api, id, field);
 
-//                        log.info("文件下载请求地址：{}",url);
+                        log.info("文件下载请求地址：{}",url);
+                        while (failCount < Const.MAX_FAIL_COUNT){
                             Response response = HttpUtil.doGet(url, null, headers);
                             if (response.body() != null && response.code() == 200) {
                                 InputStream inputStream = response.body().byteStream();
-//                            log.info("文件下载返回状态码：{}，返回信息：{}", response.code(),response.message());
-                            switch (Const.FILE_TYPE) {
-                                case OSS:
-                                    // 上传到oss
-                                    OssUtil.upload(inputStream, filePath);
-                                default:
                                 dumpToServer(headers, id, filePath, url, response, inputStream);
-                            }
                                 Map<String, Object> paramMap = Maps.newHashMap();
                                 if ("Document".equals(api)) {
                                     paramMap.put("key", "localUrl");
@@ -1216,9 +1188,19 @@ public class DataImportNewServiceImpl implements DataImportNewService {
                                 }
                                 paramMap.put("value", filePath);
                                 maps.add(paramMap);
+                                break;
                             }else {
-                            log.error("文件下载失败！, id: "+ id + "，返回体信息：" + response.message());
+                                failCount++;
-                            EmailUtil.send("File Dump ERROR", "文件下载失败！, id: "+ id + "，返回体信息：" + response.message());
+                                log.error("文件下载失败，第"+ failCount +"次！, id: "+ id + "，返回体信息：" + response.message());
+                            }
+                        }
+
+                        if (failCount == Const.MAX_FAIL_COUNT) {
+                            Map<String, Object> paramMap = Maps.newHashMap();
+                            paramMap.put("key", "is_dump");
+                            paramMap.put("value", 2);
+                            maps.add(paramMap);
+                            customMapper.updateById(api, maps, id);
                         }
                     }
                     Map<String, Object> paramMap = Maps.newHashMap();
@@ -1262,6 +1244,8 @@ public class DataImportNewServiceImpl implements DataImportNewService {
         RandomAccessFile accessFile = new RandomAccessFile(path, "rw");
         while (true) {
             try {
+                TimeUnit.MILLISECONDS.sleep(1);
+
                 // 保存到本地
                 byte[] buf = new byte[8192];
                 int len = 0;
@@ -1274,7 +1258,9 @@ public class DataImportNewServiceImpl implements DataImportNewService {
                     offset += len;
                 }
                 break;
-            } catch (Exception e) {
+            }catch (InterruptedException interruptedException){
+                return;
+            }catch (Exception e) {
                 if (offset <= 0) {
                     throw e;
                 }
@@ -1720,14 +1706,20 @@ public class DataImportNewServiceImpl implements DataImportNewService {
 
         for (String api : apis) {
             try {
-                List<SalesforceParam> salesforceParams = null;
+                QueryWrapper<LinkConfig> dbQw = new QueryWrapper<>();
-                QueryWrapper<DataBatch> dbQw = new QueryWrapper<>();
+                dbQw.eq("api", param.getApi()).eq("is_link",1).eq("is_create",1);
-                dbQw.eq("name", api);
+                List<LinkConfig> linkConfigs = linkConfigService.list(dbQw);
-                if (StringUtils.isNotEmpty(beginDateStr) && StringUtils.isNotEmpty(endDateStr)) {
+                if (linkConfigs.isEmpty()){
-                    dbQw.eq("sync_start_date", beginDateStr);  // 等于开始时间
+                    continue;
-                    dbQw.eq("sync_end_date", endDateStr);    // 等于结束时间
                 }
-                List<DataBatch> list = dataBatchService.list(dbQw);
+                List<SalesforceParam> salesforceParams = null;
+                QueryWrapper<DataBatch> dbQw1 = new QueryWrapper<>();
+                dbQw1.eq("name", api);
+                if (StringUtils.isNotEmpty(beginDateStr) && StringUtils.isNotEmpty(endDateStr)) {
+                    dbQw1.eq("sync_start_date", beginDateStr);  // 等于开始时间
+                    dbQw1.eq("sync_end_date", endDateStr);    // 等于结束时间
+                }
+                List<DataBatch> list = dataBatchService.list(dbQw1);
                 AtomicInteger batch = new AtomicInteger(1);
                 if (CollectionUtils.isNotEmpty(list)) {
                     salesforceParams = list.stream().map(t -> {
@@ -1739,12 +1731,13 @@ public class DataImportNewServiceImpl implements DataImportNewService {
                         return salesforceParam;
                     }).collect(Collectors.toList());
                 }
+                Set<String> safeSet = ConcurrentHashMap.newKeySet();
 
                 // 手动任务优先执行
                 for (SalesforceParam salesforceParam : salesforceParams) {
                     Future<?> future = salesforceExecutor.execute(() -> {
                         try {
-                            updateLinkBatch(salesforceParam, resultMap);
+                            updateLinkBatch(salesforceParam, resultMap,safeSet,linkConfigs);
                         } catch (Throwable throwable) {
                             log.error("salesforceExecutor error", throwable);
                             throw new RuntimeException(throwable);
@@ -1752,23 +1745,27 @@ public class DataImportNewServiceImpl implements DataImportNewService {
                     }, salesforceParam.getBatch(), 1);
                     futures.add(future);
                 }
+
+                if (!safeSet.isEmpty()){
+                    String format = String.format("更新关联类型 error, api name: %s,  \ncause:\n%s 前三位编码对象不存在！！",  api, com.alibaba.fastjson2.JSON.toJSONString(safeSet));
+                    EmailUtil.send("DataDump ERROR", format);
+                }
+
                 // 等待当前所有线程执行完成
                 salesforceExecutor.waitForFutures(futures.toArray(new Future<?>[]{}));
             } catch (Exception e) {
                 throw e;
             }
         }
-        return null;
+        return ReturnT.SUCCESS;
     }
 
     /**
      * 执行数据updateLinkType
      */
-    private void updateLinkBatch(SalesforceParam param, Map<String, String> resultMap) throws Exception {
+    private void updateLinkBatch(SalesforceParam param, Map<String, String> resultMap,Set<String> safeSet,List<LinkConfig> linkConfigs) throws Exception {
+
 
-        QueryWrapper<LinkConfig> dbQw = new QueryWrapper<>();
-        dbQw.eq("api", param.getApi());
-        List<LinkConfig> linkConfigs = linkConfigService.list(dbQw);
 
         String beginDateStr = null;
         String endDateStr = null;
@@ -1778,22 +1775,44 @@ public class DataImportNewServiceImpl implements DataImportNewService {
             beginDateStr = DateUtil.format(beginDate, "yyyy-MM-dd HH:mm:ss");
             endDateStr = DateUtil.format(endDate, "yyyy-MM-dd HH:mm:ss");
         }
-        // 表内数据总量
-        Integer count = customMapper.countBySQL(param.getApi(), "where CreatedDate >= '" + beginDateStr + "' and CreatedDate < '" + endDateStr + "'");
 
-        log.info("表api：{} 存在" +count+ "条数据！", param.getApi());
+        String sql = "CreatedDate >= '" + beginDateStr + "' and CreatedDate < '" + endDateStr + "'";
+        String allFiled = "Id";
+
+        if (linkConfigs.size() == 1){
+            for (LinkConfig linkConfig : linkConfigs) {
+                sql = sql + " and " + linkConfig.getField() + " is not null";
+                allFiled = allFiled + "," + linkConfig.getField();
+            }
+        }else {
+            String info = null;
+            for (LinkConfig linkConfig : linkConfigs) {
+                if (StringUtils.isBlank(info)){
+                    info = linkConfig.getField() + " is not null" ;
+                }else {
+                    info = info + " or " + linkConfig.getField() + " is not null";
+                }
+                allFiled = allFiled + "," + linkConfig.getField();
+            }
+            sql = sql +" and ("+ info +")";
+        }
+
+        // 表内数据总量
+        Integer count = customMapper.countBySQL(param.getApi(), "where " + sql);
+
+        log.info("表api：{} 存在" +count+ "条需更新数据！开始时间:{},结束时间：{}", param.getApi(), beginDateStr, endDateStr);
 
         if (count >0 ) {
             int page = count % 2000 == 0 ? count / 2000 : (count / 2000) + 1;
 
             for (int i = 0; i < page; i++) {
 
-                log.info("表api：{}，数据量：{}，执行数据更新!", param.getApi(), 2000* (i+1));
+                log.info("表api：{}，批次：{}，单批次数：2000，开始时间：{}，结束时间：{}，执行数据更新!", param.getApi(), i,beginDateStr, endDateStr);
 
-                List<Map<String, Object>> mapList = customMapper.list("*", param.getApi(), "1=1 order by Id limit " + i * 2000 + ",2000");
+                List<Map<String, Object>> mapList = customMapper.list(allFiled, param.getApi(), sql + " order by Id limit " + i * 2000 + ",2000");
 
-                List<Map<String, Object>> updateMapList = new ArrayList<>();
                 for (int j = 1; j <= mapList.size(); j++) {
+                    List<Map<String, Object>> updateMapList = new ArrayList<>();
                     Map<String, Object> map = mapList.get(j - 1);
                     for (LinkConfig config : linkConfigs) {
                         if (map.get(config.getField()) != null){
@@ -1802,13 +1821,19 @@ public class DataImportNewServiceImpl implements DataImportNewService {
                             paramMap.put("key", config.getLinkField());
                             paramMap.put("value", type);
                             updateMapList.add(paramMap);
-                        }
+                            if (StringUtils.isBlank(type)){
-                    }
+                                safeSet.add(map.get(config.getField()).toString().substring(0, 3));
-                    customMapper.updateById(param.getApi(), updateMapList, String.valueOf(mapList.get(j - 1).get("Id") != null?mapList.get(j - 1).get("Id") : mapList.get(j - 1).get("id")));
+                            }else {
 
                             }
                         }
                     }
+                    if (!updateMapList.isEmpty()) {
+                        customMapper.updateById(param.getApi(), updateMapList, String.valueOf(mapList.get(j - 1).get("Id") != null?mapList.get(j - 1).get("Id") : mapList.get(j - 1).get("id")));
+                    }
+                }
+            }
+        }
 
     }
 
@@ -1831,16 +1856,29 @@ public class DataImportNewServiceImpl implements DataImportNewService {
                         DataObject::getName,
                         (existing, replacement) -> replacement));
 
+        Set<String> safeSet = ConcurrentHashMap.newKeySet();
+
         for (String api : apis) {
+            QueryWrapper<LinkConfig> dbQw = new QueryWrapper<>();
+            dbQw.eq("api", param.getApi()).eq("is_link",1).eq("is_create",1);
+            List<LinkConfig> linkConfigs = linkConfigService.list(dbQw);
+            if (linkConfigs.isEmpty()){
+                continue;
+            }
             Future<?> future = salesforceExecutor.execute(() -> {
                 try {
-                    updateLink(api,param, resultMap);
+                    updateLink(api,param, resultMap,safeSet,linkConfigs);
                 } catch (Throwable throwable) {
                     log.error("salesforceExecutor error", throwable);
                     throw new RuntimeException(throwable);
                 }
             }, 0, 1);
             futures.add(future);
+
+            if (!safeSet.isEmpty()){
+                String format = String.format("更新关联类型 error, api name: %s,  \ncause:\n%s 前三位编码对象不存在！！",  api, com.alibaba.fastjson2.JSON.toJSONString(safeSet));
+                EmailUtil.send("DataDump ERROR", format);
+            }
         }
         // 等待当前所有线程执行完成
         salesforceExecutor.waitForFutures(futures.toArray(new Future<?>[]{}));
@@ -1852,11 +1890,7 @@ public class DataImportNewServiceImpl implements DataImportNewService {
     /**
      * 执行数据updateLinkType
      */
-    private void updateLink(String api,SalesforceParam param, Map<String, String> resultMap) throws Exception {
+    private void updateLink(String api,SalesforceParam param, Map<String, String> resultMap,Set<String> safeSet,List<LinkConfig> linkConfigs) throws Exception {
-
-        QueryWrapper<LinkConfig> dbQw = new QueryWrapper<>();
-        dbQw.eq("api", param.getApi());
-        List<LinkConfig> linkConfigs = linkConfigService.list(dbQw);
 
         String beginDateStr = null;
         Date beginDate = param.getBeginModifyDate();
@@ -1888,12 +1922,12 @@ public class DataImportNewServiceImpl implements DataImportNewService {
 
             for (int i = 0; i < page; i++) {
 
-                log.info("表api：{}，数据量：{}，执行数据更新!", api, 2000* (i+1));
+                log.info("表api：{}，批次：{}，批次量：{}，执行数据更新!", api,i, 2000);
 
                 List<Map<String, Object>> mapList = customMapper.list("*", api, sql2 + i * 2000 + ",2000" );
 
-                List<Map<String, Object>> updateMapList = new ArrayList<>();
                 for (int j = 1; j <= mapList.size(); j++) {
+                    List<Map<String, Object>> updateMapList = new ArrayList<>();
                     Map<String, Object> map = mapList.get(j - 1);
                     for (LinkConfig config : linkConfigs) {
                         if (map.get(config.getField()) != null){
@@ -1902,6 +1936,9 @@ public class DataImportNewServiceImpl implements DataImportNewService {
                             paramMap.put("key", config.getLinkField());
                             paramMap.put("value", type);
                             updateMapList.add(paramMap);
+                            if (StringUtils.isBlank(type)){
+                                safeSet.add(map.get(config.getField()).toString().substring(0, 3));
+                            }
                         }
                         customMapper.updateById(api, updateMapList, String.valueOf(mapList.get(j - 1).get("Id") != null?mapList.get(j - 1).get("Id") : mapList.get(j - 1).get("id")));
                     }
@@ -1922,4 +1959,377 @@ public class DataImportNewServiceImpl implements DataImportNewService {
         return ReturnT.SUCCESS;
     }
 
+    @Override
+    public ReturnT<String> insertSingle(SalesforceParam param) throws Exception {
+        List<Future<?>> futures = Lists.newArrayList();
+        try {
+            if (StringUtils.isNotBlank(param.getApi())) {
+                return insertSingleData(param, futures);
+            } else {
+                return autoInsertSingleData(param, futures);
+            }
+        }  catch (InterruptedException e) {
+            throw e;
+        }  catch (Throwable throwable) {
+            salesforceExecutor.remove(futures.toArray(new Future<?>[]{}));
+            log.error("insertSingle error", throwable);
+            throw throwable;
+        }
+    }
+
+    /**
+     * 组装【单表】 一次性Insert 参数
+     */
+    public ReturnT<String> insertSingleData(SalesforceParam param, List<Future<?>> futures) throws Exception {
+        List<String> apis = DataUtil.toIdList(param.getApi());
+
+        String beginDateStr = null;
+        String endDateStr = null;
+        if (param.getBeginCreateDate() != null && param.getEndCreateDate() != null){
+            Date beginDate = param.getBeginCreateDate();
+            Date endDate = param.getEndCreateDate();
+            beginDateStr = DateUtil.format(beginDate, "yyyy-MM-dd HH:mm:ss");
+            endDateStr = DateUtil.format(endDate, "yyyy-MM-dd HH:mm:ss");
+        }
+
+        // 全量的时候 检测是否有自动任务锁住的表
+        boolean isFull = CollectionUtils.isEmpty(param.getIds());
+        if (isFull) {
+            QueryWrapper<DataObject> qw = new QueryWrapper<>();
+            qw.eq("data_lock", 1).in("name", apis);
+            List<DataObject> list = dataObjectService.list(qw);
+            if (CollectionUtils.isNotEmpty(list)) {
+                String apiNames = list.stream().map(DataObject::getName).collect(Collectors.joining());
+                String message = "api:" + apiNames + " is locked";
+                String format = String.format("数据导入 error, api name: %s, \nparam: %s, \ncause:\n%s",  apiNames, com.alibaba.fastjson2.JSON.toJSONString(param, DataDumpParam.getFilter()), message);
+                EmailUtil.send("DataDump ERROR", format);
+                return new ReturnT<>(500, message);
+            }
+        }
+
+        PartnerConnection partnerConnection = salesforceTargetConnect.createConnect();
+        for (String api : apis) {
+            DataObject update = dataObjectService.getById(api);
+            try {
+                List<SalesforceParam> salesforceParams = null;
+                QueryWrapper<DataBatch> dbQw = new QueryWrapper<>();
+                dbQw.eq("name", api);
+                if (StringUtils.isNotEmpty(beginDateStr) && StringUtils.isNotEmpty(endDateStr)) {
+                    dbQw.eq("sync_start_date", beginDateStr);  // 等于开始时间
+                    dbQw.eq("sync_end_date", endDateStr);    // 等于结束时间
+                }
+                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());
+                }
+
+                // 手动任务优先执行
+                for (SalesforceParam salesforceParam : salesforceParams) {
+                    Future<?> future = salesforceExecutor.execute(() -> {
+                        try {
+                            insertSingleExecute(salesforceParam, partnerConnection,update);
+                        } 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.setNeedUpdate(false);
+            } catch (Exception e) {
+                throw e;
+            } finally {
+                if (isFull) {
+                    update.setName(api);
+                    update.setDataLock(0);
+                    dataObjectService.updateById(update);
+                }
+            }
+        }
+        return ReturnT.SUCCESS;
+    }
+
+    /**
+     * 组装【单表】一次性Insert 参数
+     */
+    public ReturnT<String> autoInsertSingleData(SalesforceParam param, List<Future<?>> futures) throws Exception {
+        List<String> apis = DataUtil.toIdList(param.getApi());
+
+        // 全量的时候 检测是否有自动任务锁住的表
+        boolean isFull = CollectionUtils.isEmpty(param.getIds());
+        if (isFull) {
+            QueryWrapper<DataObject> qw = new QueryWrapper<>();
+            qw.eq("data_lock", 1).in("name", apis);
+            List<DataObject> list = dataObjectService.list(qw);
+            if (CollectionUtils.isNotEmpty(list)) {
+                String apiNames = list.stream().map(DataObject::getName).collect(Collectors.joining());
+                return new ReturnT<>(500, "api:" + apiNames + " is locked");
+            }
+        }
+
+        PartnerConnection partnerConnection = salesforceTargetConnect.createConnect();
+        for (String api : apis) {
+            DataObject update = dataObjectService.getById(api);
+            try {
+                QueryWrapper<DataObject> qw = new QueryWrapper<>();
+                qw.eq("name", api);
+                DataObject dataObject = dataObjectService.getOne(qw);
+
+                List<SalesforceParam> salesforceParams = null;
+                QueryWrapper<DataBatch> dbQw = new QueryWrapper<>();
+                dbQw.eq("name", api);
+                List<DataBatch> list = dataBatchService.list(dbQw);
+                AtomicInteger batch = new AtomicInteger(1);
+                if (CollectionUtils.isNotEmpty(list)) {
+                    salesforceParams = list.stream().map(t -> {
+                        if(t.getSyncEndDate().toInstant().isAfter(t.getLastModifiedDate().toInstant())){
+                            SalesforceParam salesforceParam = param.clone();
+                            salesforceParam.setApi(t.getName());
+                            salesforceParam.setBeginCreateDate(t.getSyncStartDate());
+                            salesforceParam.setEndCreateDate(t.getSyncEndDate());
+                            salesforceParam.setBatch(batch.getAndIncrement());
+                            return salesforceParam;
+                        }
+                        return null;
+                    }).collect(Collectors.toList());
+                }
+
+                // 手动任务优先执行
+                for (SalesforceParam salesforceParam : salesforceParams) {
+                    Future<?> future = salesforceExecutor.execute(() -> {
+                        try {
+                            insertSingleExecute(salesforceParam, partnerConnection,update);
+                        } catch (Throwable throwable) {
+                            log.error("salesforceExecutor error", throwable);
+                            throw new RuntimeException(throwable);
+                        }
+                    }, salesforceParam.getBatch(), 1);
+                    futures.add(future);
+                }
+                // 等待当前所有线程执行完成
+                salesforceExecutor.waitForFutures(futures.toArray(new Future<?>[]{}));
+
+            } catch (Exception e) {
+                throw e;
+            } finally {
+                if (isFull) {
+                    update.setNeedUpdate(false);
+                    update.setName(api);
+                    update.setDataLock(0);
+                    dataObjectService.updateById(update);
+                }
+            }
+        }
+        return ReturnT.SUCCESS;
+    }
+
+    /**
+     * 执行一次性Insert数据
+     */
+    private void insertSingleExecute(SalesforceParam param, PartnerConnection partnerConnection,DataObject dataObject) throws Exception {
+        Map<String, Object> infoFlag = customMapper.list("code,value","system_config","code ='"+SystemConfigCode.INFO_FLAG+"'").get(0);
+
+        String api = param.getApi();
+        TimeUnit.MILLISECONDS.sleep(1);
+        QueryWrapper<DataField> dbQw = new QueryWrapper<>();
+        dbQw.eq("api", api);
+        List<DataField> list = dataFieldService.list(dbQw);
+
+        String sql = "";
+        String sql2 = "";
+
+        String beginDateStr = null;
+        String endDateStr = null;
+        Date beginDate = param.getBeginCreateDate();
+        Date endDate = param.getEndCreateDate();
+        if (param.getBeginCreateDate() != null && param.getEndCreateDate() != null){
+            beginDateStr = DateUtil.format(beginDate, "yyyy-MM-dd HH:mm:ss");
+            endDateStr = DateUtil.format(endDate, "yyyy-MM-dd HH:mm:ss");
+        }
+
+        if (1 == param.getType()) {
+            if (api.endsWith("Share")){
+                sql = "where RowCause = 'Manual' and new_id is null and CreatedDate >= '" + beginDateStr + "' and CreatedDate < '" + endDateStr + "'";
+                sql2 = "RowCause = 'Manual' and new_id is null and CreatedDate >= '" + beginDateStr + "' and CreatedDate < '" + endDateStr + "' order by Id asc limit ";
+            }else {
+                sql = "where new_id is null and CreatedDate >= '" + beginDateStr + "' and CreatedDate < '" + endDateStr + "'";
+                sql2 = "new_id is null and CreatedDate >= '" + beginDateStr + "' and CreatedDate < '" + endDateStr + "' order by Id asc limit ";
+            }
+        }else {
+            if (api.endsWith("Share")){
+                sql = "where RowCause = 'Manual' and new_id is null and LastModifiedDate >= '" + beginDateStr + "' ";
+                sql2 = "RowCause = 'Manual' and new_id is null and LastModifiedDate >= '" + beginDateStr + "' order by Id asc limit ";
+            }else {
+                sql = "where new_id is null and LastModifiedDate >= '" + beginDateStr + "' ";
+                sql2 = "new_id is null and LastModifiedDate >= '" + beginDateStr + "' order by Id asc limit ";
+            }
+        }
+        //表内数据总量
+        Integer count = customMapper.countBySQL(api, sql);
+
+        if(count == 0){
+            return;
+        }
+
+        //查询当前对象多态字段映射
+        QueryWrapper<LinkConfig> queryWrapper = new QueryWrapper<>();
+        queryWrapper.eq("api",api).eq("is_create",1).eq("is_link",true);
+        List<LinkConfig> configs = linkConfigService.list(queryWrapper);
+        Map<String, String> fieldMap = new HashMap<>();
+        if (!configs.isEmpty())  {
+            fieldMap = configs.stream()
+                    .collect(Collectors.toMap(
+                            LinkConfig::getField,      // Key提取器
+                            LinkConfig::getLinkField,  // Value提取器
+                            (oldVal, newVal) -> newVal // 解决重复Key冲突（保留新值）
+                    ));
+        }
+
+        int targetCount = 0;
+        //批量插入200一次
+        int page = count%200 == 0 ? count/200 : (count/200) + 1;
+        log.info("总Insert数据 count:{}；总批次:{}；-开始时间:{}；-结束时间:{}；-api:{}；", count,page, beginDateStr, endDateStr, api);
+
+        for (int i = 0; i < page; i++) {
+            try {
+                List<Map<String, Object>> data = customMapper.list("*", api,  sql2+ i * 200 + ",200");
+                int size = data.size();
+                log.info("执行api:{}, 执行批次:{}, 执行数据量:{}，开始时间：{}，结束时间：{}", api, i+1, size,beginDateStr,endDateStr);
+                SObject[] accounts = new SObject[size];
+                String[] ids = new String[size];
+                for (int j = 1; j <= size; j++) {
+                    SObject account = new SObject();
+                    account.setType(api);
+                    //给对象赋值
+                    for (DataField dataField : list) {
+                        String field = dataField.getField();
+                        String reference_to = dataField.getReferenceTo();
+
+                        String value = String.valueOf(data.get(j - 1).get(field));
+                        //根据旧sfid查找引用对象新sfid
+                        if (!DataUtil.isUpdate(field) || (dataField.getIsCreateable() != null && !dataField.getIsCreateable())) {
+                            continue;
+                        } else if ("reference".equals(dataField.getSfType()) && data.get(j - 1).get(dataField.getField()) != null) {
+
+                            if (!"null".equals(value) && StringUtils.isNotEmpty(value)) {
+                                //引用类型字段
+                                String linkfield = fieldMap.get(dataField.getField());
+                                if (StringUtils.isNotBlank(linkfield)){
+                                    reference_to = data.get(j - 1).get(linkfield)!=null?data.get(j - 1).get(linkfield).toString():null;
+                                }
+                                if (reference_to == null){
+                                    continue;
+                                }
+//                            log.info("----------" + dataField.getField() + "的引用类型 ------------" + reference_to);
+                                //判断reference_to内是否包含User字符串
+                                if (reference_to.contains(",User") || reference_to.contains("User,")) {
+                                    reference_to = "User";
+                                }
+                                Map<String, Object> m = customMapper.getById("new_id", reference_to, value);
+                                if (m != null && !m.isEmpty()) {
+                                    account.setField(field, m.get("new_id"));
+                                }else {
+                                    String message = "对象类型:" + api + "的数据:"+ data.get(j - 1).get("Id") +"的引用对象:" + reference_to + "的数据:"+ data.get(j - 1).get(field) +"不存在！";
+                                    EmailUtil.send("DataDump ERROR", message);
+                                    log.info(message);
+                                    return;
+                                }
+                            }
+                        } else {
+                            if (data.get(j - 1).get(field) != null && StringUtils.isNotBlank(dataField.getSfType())) {
+                                account.setField(field, DataUtil.localDataToSfData(dataField.getSfType(), value));
+                            }else {
+                                account.setField(field, data.get(j - 1).get(field));
+                            }
+                        }
+                    }
+                    if (dataObject.getIsEditable()){
+                        account.setField("old_owner_id__c", data.get(j - 1).get("OwnerId"));
+                        account.setField("old_sfdc_id__c", data.get(j - 1).get("Id"));
+                    }
+
+                    ids[j-1] = data.get(j-1).get("Id").toString();
+                    accounts[j-1] = account;
+                    if (i*200+j == count){
+                        break;
+                    }
+                }
+
+                if (infoFlag != null && "1".equals(infoFlag.get("value"))){
+                    printlnAccountsDetails(accounts, list);
+                }
+                SaveResult[] saveResults = partnerConnection.create(accounts);
+                log.info("sf return saveResults------"+ JSONArray.toJSONString(saveResults));
+                int index = 0;
+                for (SaveResult saveResult : saveResults){
+                    if (saveResult.getSuccess()) {
+                        List<Map<String, Object>> maps = new ArrayList<>();
+                        Map<String, Object> m = new HashMap<>();
+                        m.put("key", "new_id");
+                        m.put("value", saveResult.getId());
+                        maps.add(m);
+                        customMapper.updateById(api, maps, ids[index]);
+                        index++;
+                    }else{
+                        log.error("-------------saveResults: {}", JSON.toJSONString(saveResult));
+                    }
+                }
+                TimeUnit.MILLISECONDS.sleep(1);
+            } catch (Exception e) {
+                log.error("insertSingle error api:{},错误信息：{}", api, e.getMessage());
+            }
+        }
+        SalesforceParam countParam = new SalesforceParam();
+        countParam.setApi(api);
+        countParam.setBeginCreateDate(beginDate);
+        countParam.setEndCreateDate(DateUtils.addSeconds(endDate, -1));
+        // 存在isDeleted 只查询IsDeleted为false的
+        if (dataFieldService.hasDeleted(countParam.getApi())) {
+            countParam.setIsDeleted(false);
+        } else {
+            // 不存在 过滤
+            countParam.setIsDeleted(null);
+        }
+        // sf count
+        Integer sfNum = commonService.countSfNum(partnerConnection, countParam);
+
+        UpdateWrapper<DataBatchHistory> updateQw = new UpdateWrapper<>();
+        updateQw.eq("name", api)
+                .eq("sync_start_date", beginDate)
+                .eq("sync_end_date", DateUtils.addSeconds(endDate, -1))
+                .set("target_sf_num", sfNum);
+        dataBatchHistoryService.update(updateQw);
+
+        UpdateWrapper<DataBatch> updateQw2 = new UpdateWrapper<>();
+        updateQw2.eq("name", api)
+                .eq("sync_start_date", beginDate)
+                .eq("sync_end_date", endDate)
+                .set("sf_add_num", sfNum);
+        dataBatchService.update(updateQw2);
+
+        UpdateWrapper<DataBatchHistory> updateQw3 = new UpdateWrapper<>();
+        updateQw3.eq("name", api)
+                .eq("sync_start_date", beginDate)
+                .eq("sync_end_date", DateUtils.addSeconds(endDate, -1))
+                .set("target_update_num", targetCount);
+        dataBatchHistoryService.update(updateQw3);
+
+        UpdateWrapper<DataBatch> updateQw4 = new UpdateWrapper<>();
+        updateQw4.eq("name", api)
+                .eq("sync_start_date", beginDate)
+                .eq("sync_end_date", endDate)
+                .set("sf_update_num", targetCount);
+        dataBatchService.update(updateQw4);
+    }
+
 }

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

@@ -328,7 +328,7 @@ public class DataImportServiceImpl implements DataImportService {
                         }
                         if ("CreatedById".equals(dataField.getField()) && dataField.getIsCreateable()){
                             Map<String, Object> CreatedByIdMap = customMapper.getById("new_id", "User", data.get(j-1).get("CreatedById").toString());
-                            if(CreatedByIdMap.get("new_id") != null && StringUtils.isNotEmpty(CreatedByIdMap.get("new_id").toString())){
+                            if(CreatedByIdMap != null ){
                                 account.setField("CreatedById", CreatedByIdMap.get("new_id"));
                             }
                             continue;
@@ -348,7 +348,6 @@ public class DataImportServiceImpl implements DataImportService {
                                 if (reference == null){
                                     continue;
                                 }
-//                                log.info("----------" + dataField.getField() + "的引用类型 ------------" + reference);
 
                                 List<Map<String, Object>> referenceMap = customMapper.list("new_id", reference, "new_id is not null limit 1");
                                 if (referenceMap.isEmpty()){
@@ -402,7 +401,7 @@ public class DataImportServiceImpl implements DataImportService {
                 }
                 TimeUnit.MILLISECONDS.sleep(1);
             } catch (Exception e) {
-                log.error("manualCreatedNewId error api:{},错误信息：{}", api, e.getMessage());
+                log.info(JSON.toJSONString(e));
             }
         }
         SalesforceParam countParam = new SalesforceParam();

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

@@ -43,12 +43,12 @@ public class CsvConverterUtil {
 
         if (isEditable){
             header = Stream.concat(
-                    list.stream().filter(DataField::getIsCreateable)
+                    list.stream().filter(dataField ->(dataField.getIsCreateable() != null && dataField.getIsCreateable()))
                             .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 = list.stream().filter(DataField::getIsCreateable)
+            header = list.stream().filter(dataField ->(dataField.getIsCreateable() != null && dataField.getIsCreateable()))
                     .map(DataField::getField).toArray(String[]::new);
         }
 

src/main/resources/mapper/SalesforceMapper.xml

@@ -86,7 +86,7 @@
         select COUNT(Id) num from ${param.api}
         <where>
             <if test="param != null">
-                <if test="param.ids != null">
+                <if test="param.ids  != null and param.ids.size()  > 0">
                     AND id IN <foreach item="id" collection="param.ids"  open="(" separator="," close=")">#{id}</foreach>
                 </if>
                 <if test="param.beginModifyDate != null">