From 468aba25152fbfab5add59ae857d25949c48549c Mon Sep 17 00:00:00 2001 From: Kris <2893855659@qq.com> Date: Fri, 29 Aug 2025 14:30:42 +0800 Subject: [PATCH] =?UTF-8?q?=E3=80=90feat=E3=80=91=E6=95=B0=E6=8D=AE?= =?UTF-8?q?=E6=9B=B4=E6=96=B0=E5=A4=B1=E8=B4=A5=E8=AE=B0=E5=BD=95=E5=8F=8A?= =?UTF-8?q?=E6=95=B0=E6=8D=AE=E6=9B=B4=E6=96=B0=E5=AD=97=E6=AE=B5=E5=88=A4?= =?UTF-8?q?=E6=96=AD=E6=9D=A1=E4=BB=B6?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .run/data-dump（亚数）.run.xml | 14 ++ .run/data-dump（启德）.run.xml | 14 ++ .../com/celnet/datadump/entity/DataField.java | 4 +- .../celnet/datadump/job/DataDumpNewJob.java | 34 ++- .../datadump/service/CommonBatchService.java | 11 + .../service/impl/CommonBatchServiceImpl.java | 237 ++++++++++++++++++ .../service/impl/CommonServiceImpl.java | 11 +- .../impl/DataImportNewServiceImpl.java | 20 +- 8 files changed, 325 insertions(+), 20 deletions(-) create mode 100644 .run/data-dump（亚数）.run.xml create mode 100644 .run/data-dump（启德）.run.xml create mode 100644 src/main/java/com/celnet/datadump/service/CommonBatchService.java create mode 100644 src/main/java/com/celnet/datadump/service/impl/CommonBatchServiceImpl.java diff --git a/.run/data-dump（亚数）.run.xml b/.run/data-dump（亚数）.run.xml new file mode 100644 index 0000000..62a9c87 --- /dev/null +++ b/.run/data-dump（亚数）.run.xml @@ -0,0 +1,14 @@ + + + + + + + + + \ No newline at end of file diff --git a/.run/data-dump（启德）.run.xml b/.run/data-dump（启德）.run.xml new file mode 100644 index 0000000..6484b6b --- /dev/null +++ b/.run/data-dump（启德）.run.xml @@ -0,0 +1,14 @@ + + + + + + + + + \ No newline at end of file diff --git a/src/main/java/com/celnet/datadump/entity/DataField.java b/src/main/java/com/celnet/datadump/entity/DataField.java index 0ac7b23..03bb633 100644 --- a/src/main/java/com/celnet/datadump/entity/DataField.java +++ b/src/main/java/com/celnet/datadump/entity/DataField.java @@ -100,9 +100,9 @@ public class DataField implements Serializable { /** * 是否更新（目标库） */ - @TableField("is_update") + @TableField("is_updateable") @ApiModelProperty(value = "是否更新") - private Boolean isUpdate; + private Boolean isUpdateable; /** * 是否为空（目标库） diff --git a/src/main/java/com/celnet/datadump/job/DataDumpNewJob.java b/src/main/java/com/celnet/datadump/job/DataDumpNewJob.java index 854bac5..7db6881 100644 --- a/src/main/java/com/celnet/datadump/job/DataDumpNewJob.java +++ b/src/main/java/com/celnet/datadump/job/DataDumpNewJob.java @@ -4,10 +4,7 @@ import com.alibaba.fastjson.JSON; import com.celnet.datadump.config.SalesforceConnect; import com.celnet.datadump.entity.DataObject; import com.celnet.datadump.param.SalesforceParam; -import com.celnet.datadump.service.CommonService; -import com.celnet.datadump.service.DataImportBatchService; -import com.celnet.datadump.service.DataImportNewService; -import com.celnet.datadump.service.DataObjectService; +import com.celnet.datadump.service.*; import com.celnet.datadump.util.DataUtil; import com.xxl.job.core.biz.model.ReturnT; import com.xxl.job.core.handler.annotation.XxlJob; @@ -40,6 +37,9 @@ public class DataDumpNewJob { @Autowired private DataObjectService dataObjectService; + @Autowired + private CommonBatchService commonBatchService; + /** * bulk批量大数据生成newSFID @@ -75,7 +75,7 @@ public class DataDumpNewJob { */ @XxlJob("dataUpdateBatchJob") public ReturnT dataUpdateBatchJob(String paramStr) throws Exception { - log.info("dataImportBatchJob execute start .................."); + log.info("dataUpdateBatchJob execute start .................."); SalesforceParam param = new SalesforceParam(); try { if (StringUtils.isNotBlank(paramStr)) { @@ -90,6 +90,30 @@ public class DataDumpNewJob { return dataImportBatchService.immigrationUpdateBatch(param); } + /** + * 存量任务（大批量） + * + * @param paramStr 参数json + * @return result + */ + @XxlJob("dataDumpManualBatchJob") + public ReturnT dataDumpManualBatchJob(String paramStr) throws Exception { + log.info("dataDumpManualBatchJob execute start .................."); + SalesforceParam param = new SalesforceParam(); + try { + if (StringUtils.isNotBlank(paramStr)) { + param = JSON.parseObject(paramStr, SalesforceParam.class); + } + } catch (Throwable throwable) { + return new ReturnT<>(500, "参数解析失败！"); + } + // 参数转换 + param.setBeginCreateDate(param.getBeginDate()); + param.setEndCreateDate(param.getEndDate()); + + return commonBatchService.dumpBatch(param); + } + /** * 返写NewId * @param paramStr 参数json diff --git a/src/main/java/com/celnet/datadump/service/CommonBatchService.java b/src/main/java/com/celnet/datadump/service/CommonBatchService.java new file mode 100644 index 0000000..c7e41d6 --- /dev/null +++ b/src/main/java/com/celnet/datadump/service/CommonBatchService.java @@ -0,0 +1,11 @@ +package com.celnet.datadump.service; + +import com.celnet.datadump.param.SalesforceParam; +import com.xxl.job.core.biz.model.ReturnT; + +public interface CommonBatchService { + + ReturnT dumpBatch(SalesforceParam param) throws Exception; + + +} diff --git a/src/main/java/com/celnet/datadump/service/impl/CommonBatchServiceImpl.java b/src/main/java/com/celnet/datadump/service/impl/CommonBatchServiceImpl.java new file mode 100644 index 0000000..5cbd9e2 --- /dev/null +++ b/src/main/java/com/celnet/datadump/service/impl/CommonBatchServiceImpl.java @@ -0,0 +1,237 @@ +package com.celnet.datadump.service.impl; + +import com.alibaba.fastjson2.JSON; +import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper; +import com.celnet.datadump.config.SalesforceConnect; +import com.celnet.datadump.config.SalesforceExecutor; +import com.celnet.datadump.entity.DataBatch; +import com.celnet.datadump.entity.DataBatchHistory; +import com.celnet.datadump.entity.DataObject; +import com.celnet.datadump.entity.DataReport; +import com.celnet.datadump.global.Const; +import com.celnet.datadump.param.DataDumpParam; +import com.celnet.datadump.param.DataDumpSpecialParam; +import com.celnet.datadump.param.SalesforceParam; +import com.celnet.datadump.service.*; +import com.celnet.datadump.util.DataUtil; +import com.celnet.datadump.util.EmailUtil; +import com.celnet.datadump.util.SqlUtil; +import com.google.common.collect.Lists; +import com.google.common.collect.Maps; +import com.sforce.async.BulkConnection; +import com.sforce.soap.partner.PartnerConnection; +import com.sforce.soap.partner.QueryResult; +import com.sforce.soap.partner.sobject.SObject; +import com.xxl.job.core.biz.model.ReturnT; +import com.xxl.job.core.log.XxlJobLogger; +import lombok.extern.slf4j.Slf4j; +import org.apache.commons.collections.CollectionUtils; +import org.apache.commons.lang3.StringUtils; +import org.apache.commons.lang3.time.DateFormatUtils; +import org.apache.commons.lang3.time.DateUtils; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.stereotype.Service; + +import java.util.Date; +import java.util.List; +import java.util.Map; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.stream.Collectors; + +@Service +@Slf4j +public class CommonBatchServiceImpl implements CommonBatchService { + + @Autowired + private SalesforceExecutor salesforceExecutor; + @Autowired + private DataObjectService dataObjectService; + @Autowired + private DataBatchService dataBatchService; + @Autowired + private CommonService commonService; + @Autowired + private SalesforceConnect salesforceConnect; + @Autowired + private DataFieldService dataFieldService; + + @Override + public ReturnT dumpBatch(SalesforceParam param) throws Exception { +// List> futures = Lists.newArrayList(); +// try { +// if (StringUtils.isNotBlank(param.getApi())) { +// // 手动任务 +// ReturnT result = manualBatchDump(param, futures); +// if (result != null) { +// return result; +// } +// } else { +// // 自动任务 +// autoDump(param, futures); +// } +// return ReturnT.SUCCESS; +// } catch (Throwable throwable) { +// salesforceExecutor.remove(futures.toArray(new Future[]{})); +// log.error("dump error", throwable); +// throw throwable; +// } + return ReturnT.SUCCESS; + } + +// private ReturnT manualBatchDump(SalesforceParam param, List> futures) throws InterruptedException { +// List apis = DataUtil.toIdList(param.getApi()); +// String join = StringUtils.join(apis, ","); +// log.info("dump apis: {}", join); +// +// // 根据参数获取sql +// for (String api : apis) { +// DataObject update = new DataObject(); +// TimeUnit.MILLISECONDS.sleep(1); +// try { +// // 判断当前表是否存在 +// commonService.checkApi(api, true); +// +//// if (!hasCreatedDate(api)) { +//// DataDumpSpecialParam dataDumpSpecialParam = new DataDumpSpecialParam(); +//// dataDumpSpecialParam.setApi(api); +//// Future future = dataDumpSpecialService.getData(dataDumpSpecialParam, salesforceConnect.createConnect()); +//// // 等待当前所有线程执行完成 +//// salesforceExecutor.waitForFutures(future); +//// continue; +//// } +// +// List salesforceParams = null; +// update.setName(api); +// update.setDataLock(1); +// dataObjectService.updateById(update); +// QueryWrapper dbQw = new QueryWrapper<>(); +// dbQw.eq("name", api) +// .isNull("first_sync_date"); +// +// 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()); +// } else { +// salesforceParams = DataUtil.splitTask(param); +// } +// // 手动任务优先执行 +// for (SalesforceParam salesforceParam : salesforceParams) { +// Future future = salesforceExecutor.execute(() -> { +// try { +// dumpBatchData(salesforceParam); +// } catch (Throwable throwable) { +// log.error("salesforceExecutor error", throwable); +// throw new RuntimeException(throwable); +// } +// }, salesforceParam.getBatch(), 1); +// futures.add(future); +// } +// +// // 等待当前所有线程执行完成 +// salesforceExecutor.waitForFutures(futures.toArray(new Future[]{})); +// +// update.setDataWork(0); +// } catch (Throwable e) { +// log.error("manualDump error", e); +// throw new RuntimeException(e); +// } finally { +// update.setName(api); +// update.setDataLock(0); +// dataObjectService.updateById(update); +// } +// } +// return null; +// } +// +// +// private void dumpBatchData(SalesforceParam param) throws Throwable { +// String api = param.getApi(); +// BulkConnection bulkConnect = null; +// try { +// DataBatchHistory dataBatchHistory = new DataBatchHistory(); +// dataBatchHistory.setName(api); +// dataBatchHistory.setStartDate(new Date()); +// dataBatchHistory.setSyncStartDate(param.getBeginCreateDate()); +// if (param.getEndCreateDate() != null) { +// dataBatchHistory.setSyncEndDate(DateUtils.addSeconds(param.getEndCreateDate(), -1)); +// } +// dataBatchHistory.setBatch(param.getBatch()); +// if (param.getBeginCreateDate() != null && param.getEndCreateDate() != null) { +// log.info("NO.{} dump {}, date: {} ~ {} start", param.getBatch(), +// param.getApi(), +// DateFormatUtils.format(param.getBeginCreateDate(), "yyyy-MM-dd HH:mm:ss"), +// DateFormatUtils.format(param.getEndCreateDate(), "yyyy-MM-dd HH:mm:ss")); +// } +// bulkConnect = salesforceConnect.createBulkConnect(); +// // 存在isDeleted 只查询IsDeleted为false的 +// if (dataFieldService.hasDeleted(param.getApi())) { +// param.setIsDeleted(false); +// } else { +// // 不存在 过滤 +// param.setIsDeleted(null); +// } +// // 若count数量过多 可能导致超时出不来结果 对该任务做进一步拆分 +// int failCount = 0; +// boolean isSuccess = false; +// while (failCount <= Const.MAX_FAIL_COUNT) { +// try { +// // 不能 count Id +// if (!"FeedItem".equals(api) && !"ContentFolderMember ".equals(api)){ +// dataBatchHistory.setSfNum(commonService.countSfNum(connect, param)); +// } +// isSuccess = true; +// break; +// } catch (Throwable throwable) { +// failCount++; +// } +// } +// // 不成功 做任务拆分 +// if (!isSuccess) { +// if (splitTask(param)) { +// return connect; +// } +// } +// getAllSfData(param, connect, dataReport); +// updateDataBatch(param, dataBatchHistory); +// } catch (Throwable throwable) { +// log.error("dataDumpJob error api:{}", api, throwable); +// String type = param.getType() == 1 ? "存量" : "增量"; +// String format = String.format("%s数据迁移 error, api name: %s, \nparam: %s, \ncause:\n%s", type, api, JSON.toJSONString(param, DataDumpParam.getFilter()), throwable); +// EmailUtil.send("DataDump ERROR", format); +// throw throwable; +// } +// return connect; +// } +// +// /** +// * 统计salesforce数据量 +// * +// * @param connect connect +// * @param param 参数 +// * @return sf统计数量 +// */ +// public Integer countSfNum(PartnerConnection connect, SalesforceParam param) throws Exception { +// Map map = Maps.newHashMap(); +// map.put("param", param); +// String sql = SqlUtil.showSql("com.celnet.datadump.mapper.SalesforceMapper.count", map); +// log.info("count sql: {}", sql); +// XxlJobLogger.log("count sql: {}", sql); +// QueryResult queryResult = connect.queryAll(sql); +// SObject record = queryResult.getRecords()[0]; +// Integer num = (Integer) record.getField("num"); +// log.info("count sf num: {}", num); +// XxlJobLogger.log("count sf num: {}", num); +// return num; +// } + +} 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 b259cb7..fd26f32 100644 --- a/src/main/java/com/celnet/datadump/service/impl/CommonServiceImpl.java +++ b/src/main/java/com/celnet/datadump/service/impl/CommonServiceImpl.java @@ -579,7 +579,9 @@ public class CommonServiceImpl implements CommonService { param.setIsDeleted(null); } - dataBatchHistory.setSfNum(countSfNum(connect, param)); + if (!"FeedItem".equals(api) && !"ContentFolderMember ".equals(api)) { + dataBatchHistory.setSfNum(countSfNum(connect, param)); + } getAllSfData(param, connect, dataReport); @@ -917,7 +919,10 @@ public class CommonServiceImpl implements CommonService { maps.add(paramMap2); } - + Map paramMap3 = Maps.newHashMap(); + paramMap3.put("key", "is_update"); + paramMap3.put("value", 0); + maps.add(paramMap3); if (existsIds.contains(id)) { customMapper.updateById(api, maps, id); @@ -999,7 +1004,7 @@ public class CommonServiceImpl implements CommonService { dataField.setField(field.getName()); dataField.setName(field.getLabel()); dataField.setIsCreateable(field.getCreateable()); - dataField.setIsUpdate(field.getUpdateable()); + dataField.setIsUpdateable(field.getUpdateable()); dataField.setIsNillable(field.getNillable()); dataField.setIsDefaultedOnCreate(field.getDefaultedOnCreate()); String join = null; 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 8949c4a..88f3f12 100644 --- a/src/main/java/com/celnet/datadump/service/impl/DataImportNewServiceImpl.java +++ b/src/main/java/com/celnet/datadump/service/impl/DataImportNewServiceImpl.java @@ -738,20 +738,20 @@ public class DataImportNewServiceImpl implements DataImportNewService { if (1 == param.getType()) { if (api.endsWith("Share")){ - sql = "where RowCause = 'Manual' and is_update != 1 and new_id is not null and CreatedDate >= '" + beginDateStr + "' and CreatedDate < '" + endDateStr + "'"; - sql2 = "RowCause = 'Manual' and is_update != 1 and new_id is not null and CreatedDate >= '" + beginDateStr + "' and CreatedDate < '" + endDateStr + "' order by Id asc limit "; + sql = "where RowCause = 'Manual' and is_update = 0 and new_id is not null and CreatedDate >= '" + beginDateStr + "' and CreatedDate < '" + endDateStr + "'"; + sql2 = "RowCause = 'Manual' and is_update = 0 and new_id is not null and CreatedDate >= '" + beginDateStr + "' and CreatedDate < '" + endDateStr + "' order by Id asc limit "; }else { - sql = "where new_id is not null and is_update != 1 and CreatedDate >= '" + beginDateStr + "' and CreatedDate < '" + endDateStr + "'"; - sql2 = "new_id is not null and is_update != 1 and CreatedDate >= '" + beginDateStr + "' and CreatedDate < '" + endDateStr + "' order by Id asc limit "; + sql = "where new_id is not null and is_update = 0 and CreatedDate >= '" + beginDateStr + "' and CreatedDate < '" + endDateStr + "'"; + sql2 = "new_id is not null and is_update = 0 and CreatedDate >= '" + beginDateStr + "' and CreatedDate < '" + endDateStr + "' order by Id asc limit "; } }else { String updateDateField = dataFieldService.returnUpdateDateField(api); if (api.endsWith("Share")){ - sql = "where RowCause = 'Manual' and is_update != 1 and new_id is not null and "+updateDateField+" >= '" + beginDateStr + "' "; - sql2 = "RowCause = 'Manual' and is_update != 1 and new_id is not null and "+updateDateField+" >= '" + beginDateStr + "' order by Id asc limit "; + sql = "where RowCause = 'Manual' and is_update = 0 and new_id is not null and "+updateDateField+" >= '" + beginDateStr + "' "; + sql2 = "RowCause = 'Manual' and is_update = 0 and new_id is not null and "+updateDateField+" >= '" + beginDateStr + "' order by Id asc limit "; }else { - sql = "where new_id is not null and and is_update != 1 "+updateDateField+" >= '" + beginDateStr + "' "; - sql2 = "new_id is not null and is_update != 1 and "+updateDateField+" >= '" + beginDateStr + "' order by Id asc limit "; + sql = "where new_id is not null and and is_update = 0 "+updateDateField+" >= '" + beginDateStr + "' "; + sql2 = "new_id is not null and is_update = 0 and "+updateDateField+" >= '" + beginDateStr + "' order by Id asc limit "; } } //表内数据总量 @@ -818,7 +818,7 @@ public class DataImportNewServiceImpl implements DataImportNewService { //根据旧sfid查找引用对象新sfid if (field.equals("Id")) { account.setId(String.valueOf(map.get("new_id"))); - } else if (!DataUtil.isUpdate(field) || (dataField.getIsUpdate() != null && !dataField.getIsUpdate())) { + } else if (!DataUtil.isUpdate(field) || (dataField.getIsUpdateable() != null && !dataField.getIsUpdateable())) { continue; } else if ("reference".equals(dataField.getSfType()) && map.get(dataField.getField()) != null) { @@ -1036,7 +1036,7 @@ public class DataImportNewServiceImpl implements DataImportNewService { //根据旧sfid查找引用对象新sfid if (field.equals("Id")) { account.setId(String.valueOf(map.get("new_id"))); - } else if (!DataUtil.isUpdate(field) || (dataField.getIsUpdate() != null && !dataField.getIsUpdate())) { + } else if (!DataUtil.isUpdate(field) || (dataField.getIsUpdateable() != null && !dataField.getIsUpdateable())) { continue; } else if ("reference".equals(dataField.getSfType()) && map.get(dataField.getField()) != null) {