From f7e21839897bc8c023dfd5a6291be06ededb1d5b Mon Sep 17 00:00:00 2001 From: Kris <2893855659@qq.com> Date: Tue, 5 Aug 2025 18:02:59 +0800 Subject: [PATCH] =?UTF-8?q?=E3=80=90feat=E3=80=91=2020250805=E5=A4=A7?= =?UTF-8?q?=E7=89=88=E6=9C=AC=EF=BC=88bulk=E7=9B=B8=E5=85=B3=EF=BC=89?= =?UTF-8?q?=E8=B0=83=E6=95=B4?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../celnet/datadump/job/DataDumpNewJob.java | 22 +- .../service/DataImportNewService.java | 4 + .../service/impl/CommonServiceImpl.java | 9 +- .../impl/DataImportBatchServiceImpl.java | 322 ++++----- .../impl/DataImportNewServiceImpl.java | 654 ++++++++++++++---- .../service/impl/DataImportServiceImpl.java | 5 +- .../datadump/util/CsvConverterUtil.java | 6 +- .../resources/mapper/SalesforceMapper.xml | 4 +- 8 files changed, 707 insertions(+), 319 deletions(-) diff --git a/src/main/java/com/celnet/datadump/job/DataDumpNewJob.java b/src/main/java/com/celnet/datadump/job/DataDumpNewJob.java index 8d9d54d..a87f520 100644 --- a/src/main/java/com/celnet/datadump/job/DataDumpNewJob.java +++ b/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 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 diff --git a/src/main/java/com/celnet/datadump/service/DataImportNewService.java b/src/main/java/com/celnet/datadump/service/DataImportNewService.java index 253a524..f6da4da 100644 --- a/src/main/java/com/celnet/datadump/service/DataImportNewService.java +++ b/src/main/java/com/celnet/datadump/service/DataImportNewService.java @@ -49,4 +49,8 @@ public interface DataImportNewService { ReturnT checkDeletedData(SalesforceParam param) throws Exception; + /** + * 一次性插入数据 + */ + ReturnT insertSingle(SalesforceParam param) throws Exception; } diff --git a/src/main/java/com/celnet/datadump/service/impl/CommonServiceImpl.java b/src/main/java/com/celnet/datadump/service/impl/CommonServiceImpl.java index 10e7ae4..8ac7fbc 100644 --- a/src/main/java/com/celnet/datadump/service/impl/CommonServiceImpl.java +++ b/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 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 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 { - dataBatchHistory.setSfNum(countSfNum(connect, param)); + if (!"FeedItem".equals(api) && !"ContentFolderMember ".equals(api)){ + dataBatchHistory.setSfNum(countSfNum(connect, param)); + } isSuccess = true; break; } catch (Throwable throwable) { diff --git a/src/main/java/com/celnet/datadump/service/impl/DataImportBatchServiceImpl.java b/src/main/java/com/celnet/datadump/service/impl/DataImportBatchServiceImpl.java index 16e67a2..b4e00e8 100644 --- a/src/main/java/com/celnet/datadump/service/impl/DataImportBatchServiceImpl.java +++ b/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.IOException; +import java.io.*; +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 dbQw = new QueryWrapper<>(); dbQw.eq("api", api); List list = dataFieldService.list(dbQw); + QueryWrapper 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> referenceMap = customMapper.list("new_id", reference, "new_id is not null limit 1"); if (referenceMap.isEmpty()){ QueryWrapper 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 batchInfos = BulkUtil.createBatchesFromCSVFile(bulkConnection, salesforceInsertJob, fullPath); BulkUtil.awaitCompletion(bulkConnection, salesforceInsertJob, batchInfos); - sfNum = sfNum + checkInsertResults(bulkConnection, salesforceInsertJob, batchInfos, api, ids); + List 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 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(); - - } catch (Exception e) { - log.error("manualCreatedNewId error api:{}", api, e); - throw e; + BulkUtil.closeJob(bulkConnection, salesforceQueryJob.getId()); } } @@ -451,10 +489,10 @@ public class DataImportBatchServiceImpl implements DataImportBatchService { /** * 读写Insert结果 */ - public int checkInsertResults(BulkConnection connection, JobInfo job, + public List checkInsertResults(BulkConnection connection, JobInfo job, List batchInfoList,String api,String[] ids) throws AsyncApiException, IOException { - int index = 0; + ArrayList 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 row; while ((row = rdr.nextRecord()) != null) { - Map resultInfo = new HashMap(); + Map 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> maps = new ArrayList<>(); - Map m = new HashMap<>(); - m.put("key", "new_id"); - m.put("value", id); - maps.add(m); - customMapper.updateById(api, maps, ids[index]); - index ++; + strings.add(id); 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 header = rdr.nextRecord(); + List row; + + while ((row = rdr.nextRecord()) != null) { + Map 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> maps = new ArrayList<>(); + Map 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 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> 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 salesforceParams = null; - update.setName(api); update.setDataLock(1); dataObjectService.updateById(update); - if (api.endsWith("Share")){ - insertSingleShareData(api,bulkConnection); - continue; - } - QueryWrapper dbQw = new QueryWrapper<>(); dbQw.eq("name", api); List 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 salesforceParams = null; QueryWrapper dbQw = new QueryWrapper<>(); dbQw.eq("name", api); @@ -1061,9 +1114,6 @@ public class DataImportBatchServiceImpl implements DataImportBatchService { dbQw.eq("api", api); List list = dataFieldService.list(dbQw); TimeUnit.MILLISECONDS.sleep(1); - QueryWrapper 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 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")); - } + 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 batchInfos = BulkUtil.createBatchesFromCSVFile(bulkConnection, salesforceInsertJob, fullPath); BulkUtil.awaitCompletion(bulkConnection, salesforceInsertJob, batchInfos); - sfNum = sfNum + checkInsertResults(bulkConnection, salesforceInsertJob, batchInfos, api, ids); + List 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 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 dbQw = new QueryWrapper<>(); - dbQw.eq("api", api); - List 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 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 insertList = new ArrayList<>(); - - //查询当前对象多态字段映射 - QueryWrapper queryWrapper = new QueryWrapper<>(); - queryWrapper.eq("api",api).eq("is_create",1).eq("is_link",true); - List configs = linkConfigService.list(queryWrapper); - Map 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 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 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; - } - } - - } } diff --git a/src/main/java/com/celnet/datadump/service/impl/DataImportNewServiceImpl.java b/src/main/java/com/celnet/datadump/service/impl/DataImportNewServiceImpl.java index ab71fd2..b1894ad 100644 --- a/src/main/java/com/celnet/datadump/service/impl/DataImportNewServiceImpl.java +++ b/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> data = customMapper.list("*", api, "new_id is null and IsPersonAccount = 1 and CreatedDate >= '" + beginDateStr + "' and CreatedDate < '" + endDateStr + "' limit 200"); @@ -459,55 +460,42 @@ public class DataImportNewServiceImpl implements DataImportNewService { for (String api : apis) { DataObject update = dataObjectService.getById(api); try { - QueryWrapper qw = new QueryWrapper<>(); - qw.eq("name", api); - DataObject dataObject = dataObjectService.getOne(qw); - - List salesforceParams = null; QueryWrapper dbQw = new QueryWrapper<>(); - dbQw.eq("name", api); - List 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()); - } + dbQw.eq("name", api).orderByDesc("sync_end_date").last("LIMIT 1"); + DataBatch t = dataBatchService.getOne(dbQw); + SalesforceParam salesforceParam = param.clone(); + salesforceParam.setApi(t.getName()); + salesforceParam.setBeginCreateDate(t.getSyncStartDate()); + salesforceParam.setEndCreateDate(t.getSyncEndDate()); + salesforceParam.setBatch(new AtomicInteger(1).getAndIncrement()); // 手动任务优先执行 - for (SalesforceParam salesforceParam : salesforceParams) { - Future future = salesforceExecutor.execute(() -> { - try { - UpdateSfDataNew(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[]{})); + Future future = salesforceExecutor.execute(() -> { + try { + UpdateSfDataNew(salesforceParam, partnerConnection,update); + } catch (Throwable throwable) { + log.error("salesforceExecutor error", throwable); + throw new RuntimeException(throwable); + } + }, salesforceParam.getBatch(), 1); + futures.add(future); + + update.setNeedUpdate(false); } 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,39 +642,26 @@ public class DataImportNewServiceImpl implements DataImportNewService { TimeUnit.MILLISECONDS.sleep(1); try { String api = dataObject.getName(); - List salesforceParams = null; QueryWrapper dbQw = new QueryWrapper<>(); - dbQw.eq("name", api); - List 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()); - } + dbQw.eq("name", api).orderByDesc("sync_end_date").last("LIMIT 1"); + DataBatch t = dataBatchService.getOne(dbQw); - // 手动任务优先执行 - for (SalesforceParam salesforceParam : salesforceParams) { - Future future = salesforceExecutor.execute(() -> { - try { - UpdateSfDataNew(salesforceParam, partnerConnection,dataObject); - } catch (Throwable throwable) { - log.error("salesforceExecutor error", throwable); - throw new RuntimeException(throwable); - } - }, salesforceParam.getBatch(), 0); - futures.add(future); - } - // 等待当前所有线程执行完成 - salesforceExecutor.waitForFutures(futures.toArray(new Future[]{})); + AtomicInteger batch = new AtomicInteger(1); + SalesforceParam salesforceParam = param.clone(); + salesforceParam.setApi(t.getName()); + salesforceParam.setBeginCreateDate(t.getSyncStartDate()); + salesforceParam.setEndCreateDate(t.getSyncEndDate()); + salesforceParam.setBatch(batch.getAndIncrement()); + + Future future = salesforceExecutor.execute(() -> { + try { + UpdateSfDataNew(salesforceParam, partnerConnection,dataObject); + } catch (Throwable throwable) { + log.error("salesforceExecutor error", throwable); + throw new RuntimeException(throwable); + } + }, salesforceParam.getBatch(), 0); + futures.add(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> mapList = customMapper.list("*", api, sql2+ i * 200 + ",200"); SObject[] accounts = new SObject[mapList.size()]; @@ -826,7 +803,7 @@ public class DataImportNewServiceImpl implements DataImportNewService { account.setField("old_sfdc_id__c", map.get("Id")); } accounts[j++] = account; - } + } if (infoFlag != null && "1".equals(infoFlag.get("value"))){ printlnAccountsDetails(accounts,list); @@ -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> list = customMapper.list("Id, " + name, api, " is_dump = 0 and CreatedDate >= '" + beginDateStr + "' and CreatedDate < '" + endDateStr +"'"+ extraSql + "limit " + i * 1000 + ",1000"); for (Map map : list) { + int failCount = 0; String id = null; // 上传完毕 更新附件信息 List> maps = Lists.newArrayList(); @@ -1189,36 +1168,39 @@ 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); - 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); + 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(); + dumpToServer(headers, id, filePath, url, response, inputStream); + Map paramMap = Maps.newHashMap(); + if ("Document".equals(api)) { + paramMap.put("key", "localUrl"); + } else { + paramMap.put("key", "url"); + } + paramMap.put("value", filePath); + maps.add(paramMap); + break; + }else { + failCount++; + log.error("文件下载失败，第"+ failCount +"次！, id: "+ id + "，返回体信息：" + response.message()); } + } + + if (failCount == Const.MAX_FAIL_COUNT) { Map paramMap = Maps.newHashMap(); - if ("Document".equals(api)) { - paramMap.put("key", "localUrl"); - } else { - paramMap.put("key", "url"); - } - paramMap.put("value", filePath); + paramMap.put("key", "is_dump"); + paramMap.put("value", 2); maps.add(paramMap); - }else { - log.error("文件下载失败！, id: "+ id + "，返回体信息：" + response.message()); - EmailUtil.send("File Dump ERROR", "文件下载失败！, id: "+ id + "，返回体信息：" + response.message()); + customMapper.updateById(api, maps, id); } } Map 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 salesforceParams = null; - QueryWrapper 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); // 等于结束时间 + QueryWrapper dbQw = new QueryWrapper<>(); + dbQw.eq("api", param.getApi()).eq("is_link",1).eq("is_create",1); + List linkConfigs = linkConfigService.list(dbQw); + if (linkConfigs.isEmpty()){ + continue; } - List list = dataBatchService.list(dbQw); + List salesforceParams = null; + QueryWrapper 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 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 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 resultMap) throws Exception { + private void updateLinkBatch(SalesforceParam param, Map resultMap,Set safeSet,List linkConfigs) throws Exception { + - QueryWrapper dbQw = new QueryWrapper<>(); - dbQw.eq("api", param.getApi()); - List 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> mapList = customMapper.list("*", param.getApi(), "1=1 order by Id limit " + i * 2000 + ",2000"); + List> mapList = customMapper.list(allFiled, param.getApi(), sql + " order by Id limit " + i * 2000 + ",2000"); - List> updateMapList = new ArrayList<>(); for (int j = 1; j <= mapList.size(); j++) { + List> updateMapList = new ArrayList<>(); Map map = mapList.get(j - 1); for (LinkConfig config : linkConfigs) { if (map.get(config.getField()) != null){ @@ -1802,10 +1821,16 @@ 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)); + }else { + + } } } - 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"))); - + 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 safeSet = ConcurrentHashMap.newKeySet(); + for (String api : apis) { + QueryWrapper dbQw = new QueryWrapper<>(); + dbQw.eq("api", param.getApi()).eq("is_link",1).eq("is_create",1); + List 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 resultMap) throws Exception { - - QueryWrapper dbQw = new QueryWrapper<>(); - dbQw.eq("api", param.getApi()); - List linkConfigs = linkConfigService.list(dbQw); + private void updateLink(String api,SalesforceParam param, Map resultMap,Set safeSet,List linkConfigs) throws Exception { 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> mapList = customMapper.list("*", api, sql2 + i * 2000 + ",2000" ); - List> updateMapList = new ArrayList<>(); for (int j = 1; j <= mapList.size(); j++) { + List> updateMapList = new ArrayList<>(); Map 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 insertSingle(SalesforceParam param) throws Exception { + List> 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 insertSingleData(SalesforceParam param, List> futures) throws Exception { + List 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 qw = new QueryWrapper<>(); + qw.eq("data_lock", 1).in("name", apis); + List 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 salesforceParams = null; + QueryWrapper 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 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 autoInsertSingleData(SalesforceParam param, List> futures) throws Exception { + List apis = DataUtil.toIdList(param.getApi()); + + // 全量的时候 检测是否有自动任务锁住的表 + boolean isFull = CollectionUtils.isEmpty(param.getIds()); + if (isFull) { + QueryWrapper qw = new QueryWrapper<>(); + qw.eq("data_lock", 1).in("name", apis); + List 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 qw = new QueryWrapper<>(); + qw.eq("name", api); + DataObject dataObject = dataObjectService.getOne(qw); + + List salesforceParams = null; + QueryWrapper dbQw = new QueryWrapper<>(); + dbQw.eq("name", api); + List 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 infoFlag = customMapper.list("code,value","system_config","code ='"+SystemConfigCode.INFO_FLAG+"'").get(0); + + String api = param.getApi(); + TimeUnit.MILLISECONDS.sleep(1); + QueryWrapper dbQw = new QueryWrapper<>(); + dbQw.eq("api", api); + List 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 queryWrapper = new QueryWrapper<>(); + queryWrapper.eq("api",api).eq("is_create",1).eq("is_link",true); + List configs = linkConfigService.list(queryWrapper); + Map 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> 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 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> maps = new ArrayList<>(); + Map 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 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 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 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 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); + } + } diff --git a/src/main/java/com/celnet/datadump/service/impl/DataImportServiceImpl.java b/src/main/java/com/celnet/datadump/service/impl/DataImportServiceImpl.java index 90b4bb0..e7894c5 100644 --- a/src/main/java/com/celnet/datadump/service/impl/DataImportServiceImpl.java +++ b/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 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> 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(); diff --git a/src/main/java/com/celnet/datadump/util/CsvConverterUtil.java b/src/main/java/com/celnet/datadump/util/CsvConverterUtil.java index 01b3a78..4c94e5d 100644 --- a/src/main/java/com/celnet/datadump/util/CsvConverterUtil.java +++ b/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); } diff --git a/src/main/resources/mapper/SalesforceMapper.xml b/src/main/resources/mapper/SalesforceMapper.xml index 8fba755..99e4df7 100644 --- a/src/main/resources/mapper/SalesforceMapper.xml +++ b/src/main/resources/mapper/SalesforceMapper.xml @@ -86,8 +86,8 @@ select COUNT(Id) num from ${param.api} - - AND id IN #{id} + + AND id IN #{id} AND SystemModstamp >= #{param.beginModifyDate}