From f951e3627eb7ec85cc44b48738aa55e12fd15b2b Mon Sep 17 00:00:00 2001 From: Kris <2893855659@qq.com> Date: Wed, 20 Aug 2025 11:18:31 +0800 Subject: [PATCH] =?UTF-8?q?=E3=80=90feat=E3=80=91=20=E5=A2=9E=E5=8A=A0?= =?UTF-8?q?=E6=95=B0=E6=8D=AE=E6=8B=89=E5=8F=96=EF=BC=8C=E8=8E=B7=E5=8F=96?= =?UTF-8?q?new=5Fid=E3=80=81=E6=9B=B4=E6=96=B0=E6=95=B0=E6=8D=AE=E3=80=81?= =?UTF-8?q?=E4=B8=80=E6=AC=A1=E6=80=A7=E6=8F=92=E5=85=A5=E5=9F=8B=E7=82=B9?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../com/celnet/datadump/entity/DataLog.java | 9 +++++ .../service/impl/CommonServiceImpl.java | 12 +++++- .../impl/DataDumpSpecialServiceImpl.java | 10 ++++- .../impl/DataImportNewServiceImpl.java | 40 ++++++++++++++++++- .../service/impl/DataImportServiceImpl.java | 14 +++++++ 5 files changed, 82 insertions(+), 3 deletions(-) diff --git a/src/main/java/com/celnet/datadump/entity/DataLog.java b/src/main/java/com/celnet/datadump/entity/DataLog.java index a6d67ff..937097c 100644 --- a/src/main/java/com/celnet/datadump/entity/DataLog.java +++ b/src/main/java/com/celnet/datadump/entity/DataLog.java @@ -6,6 +6,7 @@ import com.baomidou.mybatisplus.annotation.TableId; import com.baomidou.mybatisplus.annotation.TableName; import io.swagger.annotations.ApiModel; import io.swagger.annotations.ApiModelProperty; +import lombok.AllArgsConstructor; import lombok.Getter; import lombok.Setter; @@ -101,5 +102,13 @@ public class DataLog implements Serializable { @ApiModelProperty(value = "状态") private String status; + public DataLog(String requestUrl, String requestData, Date startTime, Date endTime, String requestMethod, String requestType) { + this.requestUrl = requestUrl; + this.requestData = requestData; + this.startTime = startTime; + this.endTime = endTime; + this.requestMethod = requestMethod; + this.requestType = requestType; + } } 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 07757ca..c5bde4c 100644 --- a/src/main/java/com/celnet/datadump/service/impl/CommonServiceImpl.java +++ b/src/main/java/com/celnet/datadump/service/impl/CommonServiceImpl.java @@ -83,6 +83,8 @@ public class CommonServiceImpl implements CommonService { private MetaclassConfigService metaclassConfigService; @Autowired private LinkConfigService linkConfigService; + @Autowired + private DataLogService dataLogService; @Override public ReturnT increment(SalesforceParam param) throws Exception { @@ -686,7 +688,7 @@ public class CommonServiceImpl implements CommonService { } // 获取数据库字段进行比对 List fields = customMapper.getFields(api).stream().map(String::toUpperCase).collect(Collectors.toList()); - int failCount = 0; + int failCount = 0,batch = 0; while (true) { try { // 获取创建时间 @@ -704,10 +706,14 @@ public class CommonServiceImpl implements CommonService { } log.info("query sql: {}", sql); XxlJobLogger.log("query sql: {}", sql); + DataLog dataLog = new DataLog(api, null, new Date(), null, "数据拉取，拉取第" + batch + "批数据", "QuerySF"); QueryResult queryResult = connect.queryAll(sql); if (ObjectUtils.isEmpty(queryResult) || ObjectUtils.isEmpty(queryResult.getRecords())) { break; } + dataLog.setEndTime(new Date()); + dataLogService.save(dataLog); + SObject[] records = queryResult.getRecords(); objects = DataUtil.toJsonArray(records, dsrFields); // 获取最大修改时间和等于该修改时间的数据id @@ -720,8 +726,12 @@ public class CommonServiceImpl implements CommonService { .max(String::compareTo).get(); lastCreatedDate = maxDate; } + DataLog dataLog1 = new DataLog(api, null, new Date(), null, "数据拉取，Upsert第" + batch + "批数据", "UpsertDB"); // 存储更新 saveOrUpdate(api, fields, records, objects, true); + dataLog1.setEndTime(new Date()); + dataLogService.save(dataLog1); + count += records.length; TimeUnit.MILLISECONDS.sleep(1); String format = DateFormatUtils.format(lastCreatedDate, "yyyy-MM-dd HH:mm:ss"); diff --git a/src/main/java/com/celnet/datadump/service/impl/DataDumpSpecialServiceImpl.java b/src/main/java/com/celnet/datadump/service/impl/DataDumpSpecialServiceImpl.java index f27cb16..a614d9d 100644 --- a/src/main/java/com/celnet/datadump/service/impl/DataDumpSpecialServiceImpl.java +++ b/src/main/java/com/celnet/datadump/service/impl/DataDumpSpecialServiceImpl.java @@ -64,6 +64,8 @@ public class DataDumpSpecialServiceImpl implements DataDumpSpecialService { private SalesforceConnect salesforceConnect; @Autowired private SalesforceExecutor salesforceExecutor; + @Autowired + private DataLogService dataLogService; @Override public ReturnT dataDumpSpecial(DataDumpSpecialParam param) throws Throwable { @@ -180,7 +182,7 @@ public class DataDumpSpecialServiceImpl implements DataDumpSpecialService { // 判断是否存在要排除的id salesforceParam.setMaxId(maxId); map.put("param", salesforceParam); - DataLog dataLog = new DataLog(api, "拉取" + api + "对象数据", new Date(), null, "拉取第" + batch + "批数据", "Pull"); + DataLog dataLog = new DataLog(api, null, new Date(), null, "数据拉取，拉取第" + batch + "批数据", "QuerySF"); String sql = SqlUtil.showSql("com.celnet.datadump.mapper.SalesforceMapper.listOrderById", map); log.info("query sql: {}", sql); XxlJobLogger.log("query sql: {}", sql); @@ -188,11 +190,17 @@ public class DataDumpSpecialServiceImpl implements DataDumpSpecialService { if (ObjectUtils.isEmpty(queryResult) || ObjectUtils.isEmpty(queryResult.getRecords())) { break; } + dataLog.setEndTime(new Date()); + dataLogService.save(dataLog); + SObject[] records = queryResult.getRecords(); objects = DataUtil.toJsonArray(records, dsrFields); maxId = ((JSONObject) objects.get(objects.size() - 1)).getString(Const.ID); + DataLog dataLog1 = new DataLog(api, null, new Date(), null, "数据拉取，Upsert第" + batch + "批数据", "UpsertDB"); // 存储更新 commonService.saveOrUpdate(api, fields, records, objects, true); + dataLog1.setEndTime(new Date()); + dataLogService.save(dataLog1); count += records.length; TimeUnit.MILLISECONDS.sleep(1); log.info("dump success count: {}", count); 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 b38dc7c..5c57078 100644 --- a/src/main/java/com/celnet/datadump/service/impl/DataImportNewServiceImpl.java +++ b/src/main/java/com/celnet/datadump/service/impl/DataImportNewServiceImpl.java @@ -88,6 +88,9 @@ public class DataImportNewServiceImpl implements DataImportNewService { @Autowired private LinkConfigService linkConfigService; + @Autowired + private DataLogService dataLogService; + private static final String TEMP_FILE_PATH = "verify/"; static { @@ -790,6 +793,7 @@ public class DataImportNewServiceImpl implements DataImportNewService { for (int i = 0; i < page; i++) { + DataLog dataLog = new DataLog(api, "开始时间："+beginDateStr+";结束时间："+endDateStr, new Date(), null, "数据更新，查询并组装第" + z*50 + i + "批数据", "QueryDB"); int startIndex = 200 * i; int endIndex = Math.min(startIndex + 200, size); if (startIndex >= endIndex) { @@ -853,11 +857,13 @@ public class DataImportNewServiceImpl implements DataImportNewService { } accounts[j++] = account; } + dataLog.setEndTime(new Date()); + dataLogService.save(dataLog); if (infoFlag != null && "1".equals(infoFlag.get("value"))){ printlnAccountsDetails(accounts,list); } - + DataLog dataLog1 = new DataLog(api, "开始时间："+beginDateStr+";结束时间："+endDateStr, new Date(), null, "数据更新，更新SF第" + z*50 + i + "批数据", "UpdateSF"); SaveResult[] saveResults = partnerConnection.update(accounts); for (SaveResult saveResult : saveResults) { if (!saveResult.getSuccess()) { @@ -869,6 +875,8 @@ public class DataImportNewServiceImpl implements DataImportNewService { targetCount ++; } } + dataLog1.setEndTime(new Date()); + dataLogService.save(dataLog1); } catch (Throwable e) { log.info(JSON.toJSONString(e)); @@ -2888,6 +2896,8 @@ public class DataImportNewServiceImpl implements DataImportNewService { int page = size%200 == 0 ? size/200 : (size/200) + 1; for (int i = 0; i < page; i++) { + + DataLog dataLog = new DataLog(api, "开始时间："+beginDateStr+";结束时间："+endDateStr, new Date(), null, "一次性插入，查询并组装" + (z*50+i) + "批数据", "QueryBD"); int startIndex = 200 * i; int endIndex = Math.min(startIndex + 200, size); if (startIndex >= endIndex) { @@ -2955,11 +2965,20 @@ public class DataImportNewServiceImpl implements DataImportNewService { break; } } + dataLog.setEndTime(new Date()); + dataLogService.save(dataLog); if (infoFlag != null && "1".equals(infoFlag.get("value"))){ printlnAccountsDetails(accounts, list); } + + DataLog dataLog1 = new DataLog(api, "开始时间："+beginDateStr+";结束时间："+endDateStr, new Date(), null, "一次性插入，插入SF第" + (z*50+i) + "批数据", "InsertSF"); SaveResult[] saveResults = partnerConnection.create(accounts); + dataLog1.setEndTime(new Date()); + dataLogService.save(dataLog1); + + DataLog dataLog2 = new DataLog(api, "开始时间："+beginDateStr+";结束时间："+endDateStr, new Date(), null, "一次性插入，更新DB第" + (z*50+i) + "批数据", "UpdateDB"); + int index = 0; for (SaveResult saveResult : saveResults){ if (saveResult.getSuccess()) { @@ -2975,6 +2994,12 @@ public class DataImportNewServiceImpl implements DataImportNewService { } index++; } + dataLog2.setEndTime(new Date()); + dataLogService.save(dataLog2); + + TimeUnit.MILLISECONDS.sleep(1); + }catch (InterruptedException e){ + return; } catch (Exception e) { log.error("insertSingle error api:{},错误信息：{}", api, JSON.toJSONString(e)); } @@ -3085,6 +3110,8 @@ public class DataImportNewServiceImpl implements DataImportNewService { for (int i = 0; i < page; i++) { try { + DataLog dataLog = new DataLog(api, null, new Date(), null, "一次性插入，查询并组装" + (z*50+i) + "批数据", "QueryBD"); + int startIndex = 200 * i; int endIndex = Math.min(startIndex + 200, size); if (startIndex >= endIndex) { @@ -3150,11 +3177,19 @@ public class DataImportNewServiceImpl implements DataImportNewService { break; } } + dataLog.setEndTime(new Date()); + dataLogService.save(dataLog); if (infoFlag != null && "1".equals(infoFlag.get("value"))){ printlnAccountsDetails(accounts, list); } + DataLog dataLog1 = new DataLog(api, null, new Date(), null, "一次性插入，插入SF第" + (z*50+i) + "批数据", "InsertSF"); SaveResult[] saveResults = partnerConnection.create(accounts); + dataLog1.setEndTime(new Date()); + dataLogService.save(dataLog1); + + DataLog dataLog2 = new DataLog(api, null, new Date(), null, "一次性插入，更新DB第" + (z*50+i) + "批数据", "UpdateDB"); + int index = 0; for (SaveResult saveResult : saveResults){ if (saveResult.getSuccess()) { @@ -3170,6 +3205,9 @@ public class DataImportNewServiceImpl implements DataImportNewService { } index++; } + dataLog2.setEndTime(new Date()); + dataLogService.save(dataLog2); + TimeUnit.MILLISECONDS.sleep(1); }catch (InterruptedException e){ return; 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 756b71d..b83f169 100644 --- a/src/main/java/com/celnet/datadump/service/impl/DataImportServiceImpl.java +++ b/src/main/java/com/celnet/datadump/service/impl/DataImportServiceImpl.java @@ -315,6 +315,8 @@ public class DataImportServiceImpl implements DataImportService { if (startIndex >= endIndex) { break; } + DataLog dataLog = new DataLog(api, "开始时间："+beginDateStr+";结束时间："+endDateStr, new Date(), null, "获取new_id，查询并组装第" + z*50 + i + "批数据", "QueryDB"); + List> mapList = mapsList.subList(startIndex, endIndex); int sized = mapList.size(); SObject[] accounts = new SObject[sized]; @@ -398,11 +400,20 @@ public class DataImportServiceImpl implements DataImportService { } } + dataLog.setEndTime(new Date()); + dataLogService.save(dataLog); + + DataLog dataLog1 = new DataLog(api, "开始时间："+beginDateStr+";结束时间："+endDateStr, new Date(), null, "获取new_id，插入SF第" + z*50 + i + "批数据", "InserSF"); if (infoFlag != null && "1".equals(infoFlag.get("value"))){ printAccountsDetails(accounts, list); } SaveResult[] saveResults = partnerConnection.create(accounts); log.info("sf return saveResults------"+ JSONArray.toJSONString(saveResults)); + dataLog1.setEndTime(new Date()); + dataLogService.save(dataLog1); + + DataLog dataLog2 = new DataLog(api, "开始时间："+beginDateStr+";结束时间："+endDateStr, new Date(), null, "获取new_id，更新DB第" + z*50 + i + "批数据", "UpdateDB"); + int index = 0; for (SaveResult saveResult : saveResults){ if (saveResult.getSuccess()) { @@ -417,6 +428,9 @@ public class DataImportServiceImpl implements DataImportService { } index++; } + dataLog2.setEndTime(new Date()); + dataLogService.save(dataLog2); + TimeUnit.MILLISECONDS.sleep(1); } catch (Exception e) { log.info(JSON.toJSONString(e));