diff --git a/src/main/java/com/celnet/datadump/service/impl/CommonBatchServiceImpl.java b/src/main/java/com/celnet/datadump/service/impl/CommonBatchServiceImpl.java index 008e463..01d8916 100644 --- a/src/main/java/com/celnet/datadump/service/impl/CommonBatchServiceImpl.java +++ b/src/main/java/com/celnet/datadump/service/impl/CommonBatchServiceImpl.java @@ -1119,6 +1119,7 @@ public class CommonBatchServiceImpl implements CommonBatchService { Map map = Maps.newHashMap(); String dateName = param.getType() == 1 ? Const.CREATED_DATE : param.getUpdateField(); int count = 0; + int batch = 0; Date lastCreatedDate = null; String maxId = null; @@ -1155,7 +1156,9 @@ public class CommonBatchServiceImpl implements CommonBatchService { } log.info("query sql: {}", sql); XxlJobLogger.log("query sql: {}", sql); - + + DataLog dataLog = new DataLog(api, null, new Date(), null, "数据拉取，拉取第" + batch + "批数据", "QuerySF"); + job = BulkUtil.createJob(bulkConnect, api, OperationEnum.queryAll); log.info("创建Bulk作业成功, 作业ID: {}", job.getId()); @@ -1165,6 +1168,8 @@ public class CommonBatchServiceImpl implements CommonBatchService { int completionOne = BulkUtil.awaitCompletionOne(bulkConnect, job, batchInfo); log.info("批次处理完成, 处理记录数: {}", completionOne); + dataLogService.save(dataLog); + if (completionOne == 0){ log.info("无更多数据, 结束处理"); BulkUtil.closeJob(bulkConnect, job.getId()); @@ -1173,6 +1178,8 @@ public class CommonBatchServiceImpl implements CommonBatchService { List> batchRecords = new ArrayList<>(); + DataLog dataLog1 = new DataLog(api, null, new Date(), null, "数据拉取，Upsert第" + batch + "批数据", "UpsertDB"); + QueryResultList queryResultList = bulkConnect.getQueryResultList(job.getId(), batchInfo.getId()); log.info("获取查询结果列表, 结果数量: {}", queryResultList.getResult().length); @@ -1194,6 +1201,8 @@ public class CommonBatchServiceImpl implements CommonBatchService { } } + dataLogService.save(dataLog1); + if (!batchRecords.isEmpty()){ log.info("开始保存或更新数据, 记录数: {}", batchRecords.size()); sfData = saveOrUpdate(api, batchRecords, true); @@ -1206,6 +1215,7 @@ public class CommonBatchServiceImpl implements CommonBatchService { } BulkUtil.closeJob(bulkConnect, job.getId()); + batch ++; log.info("关闭作业成功, 作业ID: {}", job.getId()); }catch (InterruptedException interruptedException){ return 0; 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 cd2da47..d5a95e9 100644 --- a/src/main/java/com/celnet/datadump/service/impl/DataImportBatchServiceImpl.java +++ b/src/main/java/com/celnet/datadump/service/impl/DataImportBatchServiceImpl.java @@ -21,9 +21,6 @@ import com.celnet.datadump.util.DataUtil; import com.celnet.datadump.util.EmailUtil; import com.google.common.collect.Lists; import com.sforce.async.*; -import com.sforce.soap.partner.PartnerConnection; -import com.sforce.soap.partner.SaveResult; -import com.sforce.soap.partner.sobject.SObject; import com.xxl.job.core.biz.model.ReturnT; import com.xxl.job.core.log.XxlJobLogger; import com.xxl.job.core.util.DateUtil; @@ -81,6 +78,9 @@ public class DataImportBatchServiceImpl implements DataImportBatchService { @Autowired private LinkConfigService linkConfigService; + @Autowired + private DataLogService dataLogService; + /** * Insert入口 */ @@ -326,6 +326,8 @@ public class DataImportBatchServiceImpl implements DataImportBatchService { for (int i = 0; i < page; i++) { + DataLog dataLog = new DataLog(api, "开始时间："+beginDateStr+";结束时间："+endDateStr, new Date(), null, "数据更新（BULK），查询并组装第" + i + "批数据", "QueryDB"); + List data = customMapper.listJsonObject("*", api, "new_id is null and CreatedDate >= '" + beginDateStr + "' and CreatedDate < '" + endDateStr + "' limit 8000"); int size = data.size(); @@ -423,17 +425,30 @@ public class DataImportBatchServiceImpl implements DataImportBatchService { break; } } + dataLog.setEndTime(new Date()); + dataLogService.save(dataLog); + JobInfo salesforceInsertJob = null; try { + DataLog dataLog1 = new DataLog(api, "开始时间："+beginDateStr+";结束时间："+endDateStr, new Date(), null, "数据更新（BULK），更新SF第" + i + "批数据", "UpdateSF"); + salesforceInsertJob = BulkUtil.createJob(bulkConnection, api, OperationEnum.insert); BatchInfo batchInfo = CsvConverterUtil.writeToCsvNewOne(insertList, UUID.randomUUID().toString(), list, true, bulkConnection, salesforceInsertJob); BulkUtil.awaitOneCompletion(bulkConnection, salesforceInsertJob, batchInfo); + dataLog1.setEndTime(new Date()); + dataLogService.save(dataLog1); + + DataLog dataLog2 = new DataLog(api, "开始时间："+beginDateStr+";结束时间："+endDateStr, new Date(), null, "数据更新（BULK），更新DB第" + i + "批数据", "UpdateDB"); + sfNum = sfNum + checkInsertResultsOne(bulkConnection, salesforceInsertJob, batchInfo, api, ids); + dataLog2.setEndTime(new Date()); + dataLogService.save(dataLog2); + TimeUnit.MILLISECONDS.sleep(1); } catch (InterruptedException interruptedException){ @@ -511,41 +526,58 @@ public class DataImportBatchServiceImpl implements DataImportBatchService { * 读写Insert结果 */ public int checkInsertResultsOne(BulkConnection connection, JobInfo job, - BatchInfo b,String api,String[] ids) - throws AsyncApiException, IOException { + BatchInfo b,String api,String[] ids) + throws Exception { int index = 0; - CSVReader rdr = - new CSVReader(connection.getBatchResultStream(job.getId(), b.getId())); - List resultHeader = rdr.nextRecord(); - int resultCols = resultHeader.size(); + CSVReader rdr = null; + int retryCount = 0; + final int maxRetries = 3; + + while (retryCount <= maxRetries) { + try { + rdr = new CSVReader(connection.getBatchResultStream(job.getId(), b.getId())); + List resultHeader = rdr.nextRecord(); + int resultCols = resultHeader.size(); - List row; - while ((row = rdr.nextRecord()) != null) { - Map resultInfo = new HashMap(); - for (int i = 0; i < resultCols; i++) { - resultInfo.put(resultHeader.get(i), row.get(i)); + List row; + while ((row = rdr.nextRecord()) != null) { + Map resultInfo = new HashMap(); + for (int i = 0; i < resultCols; i++) { + resultInfo.put(resultHeader.get(i), row.get(i)); + } + boolean insertStatus = Boolean.valueOf(resultInfo.get("Success")); + boolean created = Boolean.valueOf(resultInfo.get("Created")); + 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]); + } else{ + List> maps = new ArrayList<>(); + Map linkMap1 = new HashMap<>(); + linkMap1.put("key", "error_message"); + linkMap1.put("value", JSON.toJSONString(error)); + maps.add(linkMap1); + customMapper.updateById(api, maps, ids[index]); + log.info("Id:{},saveResults: {}",ids[index], error); + } + index ++; + } + break; // 成功执行，跳出重试循环 + }catch (Exception e) { + retryCount++; + log.info("checkInsertResultsOne failed, retrying {}/{} for api: {}", retryCount, maxRetries, api); + try { + Thread.sleep(1000 * retryCount); // 简单的退避策略 + } catch (InterruptedException ie) { + Thread.currentThread().interrupt(); + throw new IOException("Interrupted while retrying", ie); + } } - boolean insertStatus = Boolean.valueOf(resultInfo.get("Success")); - boolean created = Boolean.valueOf(resultInfo.get("Created")); - 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]); - } else{ - List> maps = new ArrayList<>(); - Map linkMap1 = new HashMap<>(); - linkMap1.put("key", "error_message"); - linkMap1.put("value", JSON.toJSONString(error)); - maps.add(linkMap1); - customMapper.updateById(api, maps, ids[index]); - log.error("Id:{},saveResults: {}",ids[index], error); - } - index ++; } return index; } @@ -826,6 +858,8 @@ public class DataImportBatchServiceImpl implements DataImportBatchService { for (int i = 0; i < page; i++) { + DataLog dataLog = new DataLog(api, "开始时间："+beginDateStr+";结束时间："+endDateStr, new Date(), null, "数据更新（BULK），查询并组装第" + i + "批数据", "QueryDB"); + List> mapList = customMapper.list("*", api, "new_id is not null and is_update = 0 and CreatedDate >= '" + beginDateStr + "' and CreatedDate < '" + endDateStr + "' order by Id asc limit 8000"); log.error("总Update数据 count:{}；当前批次:{}；-开始时间:{}；-结束时间:{}；-api:{}；", count,i, beginDateStr, endDateStr, api); @@ -911,10 +945,13 @@ public class DataImportBatchServiceImpl implements DataImportBatchService { } updateList.add(account); } + dataLog.setEndTime(new Date()); + dataLogService.save(dataLog); JobInfo salesforceUpdateJob = null; try { + DataLog dataLog1 = new DataLog(api, "开始时间："+beginDateStr+";结束时间："+endDateStr, new Date(), null, "数据更新（BULK），更新SF第" + i + "批数据", "UpdateSF"); salesforceUpdateJob = BulkUtil.createJob(bulkConnection, api, OperationEnum.update); @@ -922,10 +959,18 @@ public class DataImportBatchServiceImpl implements DataImportBatchService { BulkUtil.awaitOneCompletion(bulkConnection, salesforceUpdateJob, batchInfo); + dataLog1.setEndTime(new Date()); + dataLogService.save(dataLog1); + + DataLog dataLog2 = new DataLog(api, "开始时间："+beginDateStr+";结束时间："+endDateStr, new Date(), null, "数据更新（BULK），更新DB第" + i + "批数据", "UpdateDB"); + int updatedCount = checkUpdateResultsOne(bulkConnection, salesforceUpdateJob, batchInfo,api); sfNum = sfNum + updatedCount; + dataLog2.setEndTime(new Date()); + dataLogService.save(dataLog2); + TimeUnit.MILLISECONDS.sleep(1); } catch (InterruptedException interruptedException){ @@ -1100,7 +1145,7 @@ public class DataImportBatchServiceImpl implements DataImportBatchService { update.setDataLock(1); dataObjectService.updateById(update); - if (api.endsWith("Share") || "GroupMember".equals(api)){ + if (dataFieldService.hasCreatedDate(api)){ insertSingleShareData(api,bulkConnection); continue; } @@ -1248,8 +1293,19 @@ public class DataImportBatchServiceImpl implements DataImportBatchService { endDateStr = DateUtil.format(endDate, "yyyy-MM-dd HH:mm:ss"); } + String sql = ""; + String sql2 = ""; + + if (api.endsWith("Share")){ + sql = "where RowCause NOT IN ('Owner','Rule','Territory','Team','ImplicitChild','ImplicitParent','TerritoryRule','ImplicitCallCenter','PortalRole','Portal','ImplicitPerson','ImplicitGrant') and new_id is null "; + sql2 = "RowCause NOT IN ('Owner','Rule','Territory','Team','ImplicitChild','ImplicitParent','TerritoryRule','ImplicitCallCenter','PortalRole','Portal','ImplicitPerson','ImplicitGrant') and new_id is null limit 8000"; + }else { + sql = "where new_id is null "; + sql2 = "new_id is null limit 8000"; + } + //表内数据总量 - Integer count = customMapper.countBySQL(api, "where new_id is null and CreatedDate >= '" + beginDateStr + "' and CreatedDate < '" + endDateStr + "'"); + Integer count = customMapper.countBySQL(api, sql); if (count == 0) { return; } @@ -1277,7 +1333,9 @@ public class DataImportBatchServiceImpl implements DataImportBatchService { for (int i = 0; i < page; i++) { - List data = customMapper.listJsonObject("*", api, "new_id is null and CreatedDate >= '" + beginDateStr + "' and CreatedDate < '" + endDateStr + "' limit 8000"); + DataLog dataLog = new DataLog(api, "开始时间："+beginDateStr+";结束时间："+endDateStr, new Date(), null, "一次性插入（BULK），查询并组装" + i + "批数据", "QueryBD"); + + List data = customMapper.listJsonObject("*", api, sql2); int size = data.size(); log.error("总Insert数据 count:{}；当前批次：{}，-开始时间:{}；-结束时间:{}；-api:{}；", count,i, beginDateStr, endDateStr, api); @@ -1374,15 +1432,29 @@ public class DataImportBatchServiceImpl implements DataImportBatchService { JobInfo salesforceInsertJob = null; + dataLog.setEndTime(new Date()); + dataLogService.save(dataLog); + try { + DataLog dataLog1 = new DataLog(api, "开始时间："+beginDateStr+";结束时间："+endDateStr, new Date(), null, "一次性插入（BULK），插入SF第" + i + "批数据", "InsertSF"); + salesforceInsertJob = BulkUtil.createJob(bulkConnection, api, OperationEnum.insert); - BatchInfo batchInfo = CsvConverterUtil.writeToCsvNewOne(insertList, UUID.randomUUID().toString(), list, true, bulkConnection, salesforceInsertJob); + BatchInfo batchInfo = CsvConverterUtil.writeToCsvNewOne(insertList, UUID.randomUUID().toString(), list, dataObject.getIsEditable(), bulkConnection, salesforceInsertJob); BulkUtil.awaitOneCompletion(bulkConnection, salesforceInsertJob, batchInfo); + dataLog1.setEndTime(new Date()); + + dataLogService.save(dataLog1); + + DataLog dataLog2 = new DataLog(api, "开始时间："+beginDateStr+";结束时间："+endDateStr, new Date(), null, "一次性插入（BULK），更新DB第" + i + "批数据", "UpdateDB"); + sfNum = sfNum + checkInsertResultsOne(bulkConnection, salesforceInsertJob, batchInfo, api, ids); + dataLog2.setEndTime(new Date()); + dataLogService.save(dataLog2); + } catch (Exception e) { log.error("createdNewIdBatch error api:{}", api, e); throw e; @@ -1448,6 +1520,8 @@ public class DataImportBatchServiceImpl implements DataImportBatchService { for (int i = 0; i < page; i++) { + DataLog dataLog = new DataLog(api, "开始时间："+beginDateStr+";结束时间："+endDateStr, new Date(), null, "一次性插入（BULK），查询并组装" + i + "批数据", "QueryBD"); + List data = customMapper.listJsonObject("*", api, "new_id is null and RowCause NOT IN ('Owner','Rule','Territory','Team','ImplicitChild','ImplicitParent','TerritoryRule','ImplicitCallCenter','PortalRole','Portal','ImplicitPerson','ImplicitGrant') order by Id asc limit " + i * 10000 + ",10000"); int size = data.size(); @@ -1523,9 +1597,14 @@ public class DataImportBatchServiceImpl implements DataImportBatchService { } } + dataLog.setEndTime(new Date()); + + dataLogService.save(dataLog); + JobInfo salesforceInsertJob = null; try { + DataLog dataLog1 = new DataLog(api, "开始时间："+beginDateStr+";结束时间："+endDateStr, new Date(), null, "一次性插入（BULK），插入SF第" + i + "批数据", "InsertSF"); salesforceInsertJob = BulkUtil.createJob(bulkConnection, api, OperationEnum.insert); @@ -1533,8 +1612,17 @@ public class DataImportBatchServiceImpl implements DataImportBatchService { BulkUtil.awaitCompletion(bulkConnection, salesforceInsertJob, batchInfos); + dataLog1.setEndTime(new Date()); + + dataLogService.save(dataLog1); + + DataLog dataLog2 = new DataLog(api, "开始时间："+beginDateStr+";结束时间："+endDateStr, new Date(), null, "一次性插入（BULK），更新DB第" + i + "批数据", "UpdateDB"); + checkInsertResults(bulkConnection, salesforceInsertJob, batchInfos, api, ids); + dataLog2.setEndTime(new Date()); + + dataLogService.save(dataLog2); } 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 ec2d6f7..e744dd8 100644 --- a/src/main/java/com/celnet/datadump/service/impl/DataImportNewServiceImpl.java +++ b/src/main/java/com/celnet/datadump/service/impl/DataImportNewServiceImpl.java @@ -3663,7 +3663,7 @@ public class DataImportNewServiceImpl implements DataImportNewService { String value = String.valueOf(data.get(j - 1).get(field)); //根据旧sfid查找引用对象新sfid - if (dataField.getIsCreateable() != null && !dataField.getIsCreateable()) { + if (dataField.getIsCreateable() == null || !dataField.getIsCreateable()) { continue; } else if ("reference".equals(dataField.getSfType()) && data.get(j - 1).get(dataField.getField()) != null) { diff --git a/src/main/java/com/celnet/datadump/util/DataUtil.java b/src/main/java/com/celnet/datadump/util/DataUtil.java index 5300df3..079e08f 100644 --- a/src/main/java/com/celnet/datadump/util/DataUtil.java +++ b/src/main/java/com/celnet/datadump/util/DataUtil.java @@ -511,6 +511,7 @@ public class DataUtil { public static Object localDataToSfData(String fieldType, String data) throws ParseException { SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss'Z'"); + sdf.setTimeZone(TimeZone.getTimeZone("UTC")); SimpleDateFormat sd = new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss"); Date date; //date转Calendar类型