diff --git a/src/main/java/com/celnet/datadump/controller/JobController.java b/src/main/java/com/celnet/datadump/controller/JobController.java index 4ade6dc..0b66cf7 100644 --- a/src/main/java/com/celnet/datadump/controller/JobController.java +++ b/src/main/java/com/celnet/datadump/controller/JobController.java @@ -41,6 +41,8 @@ public class JobController { private DataCheckDeletedService dataCheckDeletedService; @Autowired private DataImportService dataImportService; + @Autowired + private DataImportBatchService dataImportBatchService; @PostMapping("/fileTransform") @ApiOperation("附件解析") @@ -213,6 +215,34 @@ public class JobController { // 参数转换 param.setBeginCreateDate(param.getBeginDate()); param.setEndCreateDate(param.getEndDate()); - return dataImportService.immigrationBatch(param); + return dataImportBatchService.immigrationBatch(param); } + + /** + * bulk批量更新大数据量数据 + * @param paramStr + * @author kris + * @return + * @throws Exception + */ + @PostMapping("/dataUpdateBatchJob") + @ApiOperation("更新数据（大数据量）") + public ReturnT dataUpdateBatchJob(String paramStr) throws Exception { + log.info("dataImportBatchJob 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.setType(1); + // 参数转换 + param.setBeginCreateDate(param.getBeginDate()); + param.setEndCreateDate(param.getEndDate()); + return dataImportBatchService.immigrationUpdateBatch(param); + } + + } diff --git a/src/main/java/com/celnet/datadump/job/DataDumpJob.java b/src/main/java/com/celnet/datadump/job/DataDumpJob.java index 3e17114..3422164 100644 --- a/src/main/java/com/celnet/datadump/job/DataDumpJob.java +++ b/src/main/java/com/celnet/datadump/job/DataDumpJob.java @@ -3,6 +3,7 @@ package com.celnet.datadump.job; import com.alibaba.fastjson.JSON; import com.celnet.datadump.param.SalesforceParam; import com.celnet.datadump.service.CommonService; +import com.celnet.datadump.service.DataImportBatchService; import com.celnet.datadump.service.DataImportService; import com.celnet.datadump.service.impl.DataImportServiceImpl; import com.xxl.job.core.biz.model.ReturnT; @@ -27,6 +28,9 @@ public class DataDumpJob { @Autowired private DataImportService dataImportService; + @Autowired + private DataImportBatchService dataImportBatchService; + /** * 创建api @@ -144,7 +148,7 @@ public class DataDumpJob { // 参数转换 param.setBeginCreateDate(param.getBeginDate()); param.setEndCreateDate(param.getEndDate()); - return dataImportService.immigrationBatch(param); + return dataImportBatchService.immigrationBatch(param); } @@ -173,6 +177,31 @@ public class DataDumpJob { return dataImportService.immigrationUpdate(param); } + /** + * bulk批量大数据更新数据 + * @param paramStr + * @author kris + * @return + * @throws Exception + */ + @XxlJob("dataUpdateBatchJob") + public ReturnT dataUpdateBatchJob(String paramStr) throws Exception { + log.info("dataImportBatchJob 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.setType(1); + // 参数转换 + param.setBeginCreateDate(param.getBeginDate()); + param.setEndCreateDate(param.getEndDate()); + return dataImportBatchService.immigrationUpdateBatch(param); + } + /** * 获取文件关联表 * diff --git a/src/main/java/com/celnet/datadump/service/DataImportBatchService.java b/src/main/java/com/celnet/datadump/service/DataImportBatchService.java new file mode 100644 index 0000000..f739519 --- /dev/null +++ b/src/main/java/com/celnet/datadump/service/DataImportBatchService.java @@ -0,0 +1,13 @@ +package com.celnet.datadump.service; + +import com.celnet.datadump.param.SalesforceParam; +import com.xxl.job.core.biz.model.ReturnT; + +public interface DataImportBatchService { + + + ReturnT immigrationBatch(SalesforceParam param) throws Exception; + + ReturnT immigrationUpdateBatch(SalesforceParam param) throws Exception; + +} diff --git a/src/main/java/com/celnet/datadump/service/DataImportService.java b/src/main/java/com/celnet/datadump/service/DataImportService.java index 34785fc..e0aca12 100644 --- a/src/main/java/com/celnet/datadump/service/DataImportService.java +++ b/src/main/java/com/celnet/datadump/service/DataImportService.java @@ -14,7 +14,6 @@ public interface DataImportService { ReturnT immigration(SalesforceParam param) throws Exception; - ReturnT immigrationBatch(SalesforceParam param) throws Exception; - ReturnT immigrationUpdate(SalesforceParam param) throws Exception; + } diff --git a/src/main/java/com/celnet/datadump/service/impl/DataImportBatchServiceImpl.java b/src/main/java/com/celnet/datadump/service/impl/DataImportBatchServiceImpl.java new file mode 100644 index 0000000..6d9c50f --- /dev/null +++ b/src/main/java/com/celnet/datadump/service/impl/DataImportBatchServiceImpl.java @@ -0,0 +1,609 @@ +package com.celnet.datadump.service.impl; + + +import cn.hutool.core.lang.UUID; +import cn.hutool.json.JSONObject; +import com.alibaba.fastjson.JSON; +import com.alibaba.fastjson.JSONArray; +import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper; +import com.baomidou.mybatisplus.core.conditions.update.UpdateWrapper; +import com.celnet.datadump.config.SalesforceExecutor; +import com.celnet.datadump.config.SalesforceTargetConnect; +import com.celnet.datadump.entity.DataBatch; +import com.celnet.datadump.entity.DataBatchHistory; +import com.celnet.datadump.entity.DataField; +import com.celnet.datadump.entity.DataObject; +import com.celnet.datadump.global.SystemConfigCode; +import com.celnet.datadump.mapper.CustomMapper; +import com.celnet.datadump.param.DataDumpParam; +import com.celnet.datadump.param.SalesforceParam; +import com.celnet.datadump.service.*; +import com.celnet.datadump.util.BulkUtil; +import com.celnet.datadump.util.CsvConverterUtil; +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; +import lombok.extern.slf4j.Slf4j; +import org.apache.commons.collections.CollectionUtils; +import org.apache.commons.lang.time.DateUtils; +import org.apache.commons.lang3.StringUtils; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.stereotype.Service; + +import java.io.IOException; +import java.time.LocalDateTime; +import java.time.ZoneId; +import java.time.ZonedDateTime; +import java.time.format.DateTimeFormatter; +import java.util.*; +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 DataImportBatchServiceImpl implements DataImportBatchService { + + @Autowired + private SalesforceTargetConnect salesforceTargetConnect; + + @Autowired + private SalesforceExecutor salesforceExecutor; + + @Autowired + private DataObjectService dataObjectService; + + @Autowired + private DataBatchService dataBatchService; + + @Autowired + private DataFieldService dataFieldService; + + @Autowired + private CustomMapper customMapper; + + @Autowired + private DataBatchHistoryService dataBatchHistoryService; + + + /** + * Insert入口 + */ + @Override + public ReturnT immigrationBatch(SalesforceParam param) throws Exception { + List> futures = Lists.newArrayList(); + try { + if (StringUtils.isNotBlank(param.getApi())) { + // 手动任务 + ReturnT result = manualImmigrationBatch(param, futures); + if (result != null) { + return result; + } + } + return ReturnT.SUCCESS; + } catch (Exception exception) { + salesforceExecutor.remove(futures.toArray(new Future[]{})); + log.error("immigration error", exception); + throw exception; + } + } + + /** + * 组装执行参数 + */ + public ReturnT manualImmigrationBatch(SalesforceParam param, List> futures) throws Exception { + List apis; + apis = DataUtil.toIdList(param.getApi()); + String join = StringUtils.join(apis, ","); + log.info("immigration apis: {}", join); + XxlJobLogger.log("immigration apis: {}", join); + + TimeUnit.MILLISECONDS.sleep(1); + + // 全量的时候 检测是否有自动任务锁住的表 + 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"); + } + } + + BulkConnection bulkConnection = salesforceTargetConnect.createBulkConnect(); + for (String api : apis) { + DataObject update = new DataObject(); + TimeUnit.MILLISECONDS.sleep(1); + try { + List salesforceParams = null; + + update.setName(api); + update.setDataLock(1); + dataObjectService.updateById(update); + + 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 -> { + 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 { + manualCreatedNewIdBatch(salesforceParam, bulkConnection); + } 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 (InterruptedException e) { + throw e; + } catch (Throwable e) { + log.error("manualImmigration error", e); + throw new RuntimeException(e); + } finally { + if (isFull) { + update.setName(api); + update.setDataLock(0); + dataObjectService.updateById(update); + } + } + } + return null; + } + + /** + * 执行数据Insert + */ + public void manualCreatedNewIdBatch(SalesforceParam param, BulkConnection bulkConnection) throws Exception { + String api = param.getApi(); + QueryWrapper dbQw = new QueryWrapper<>(); + dbQw.eq("api", api); + List list = dataFieldService.list(dbQw); + TimeUnit.MILLISECONDS.sleep(1); + + Date beginDate = param.getBeginCreateDate(); + Date endDate = param.getEndCreateDate(); + String beginDateStr = DateUtil.format(beginDate, "yyyy-MM-dd HH:mm:ss"); + String endDateStr = DateUtil.format(endDate, "yyyy-MM-dd HH:mm:ss"); + + //表内数据总量 + 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; + } + //批量插入2000一次 + int page = count%2000 == 0 ? count/2000 : (count/2000) + 1; + //总插入数 + 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 2000"); + int size = data.size(); + + log.info("执行api:{}, 执行page:{}, 执行size:{}", api, i+1, size); + List insertList = new ArrayList<>(); + + //判断引用对象是否存在new_id + DataObject update = new DataObject(); + update.setName(api); + + //更新对象的new_id + String[] ids = new String[size]; + + // 定义输入/输出格式 + DateTimeFormatter inputFormatter = DateTimeFormatter.ISO_LOCAL_DATE_TIME; + DateTimeFormatter outputFormatter = DateTimeFormatter.ofPattern("yyyy-MM-dd'T'HH:mm:ss.SSSxx"); + + for (int j = 1; j <= size; j++) { + JSONObject account = new JSONObject(); + for (DataField dataField : list) { + if ("OwnerId".equals(dataField.getField()) || "Owner_Type".equals(dataField.getField()) + || "Id".equals(dataField.getField())){ + continue; + } + if (dataField.getIsCreateable() && !dataField.getIsNillable() && !dataField.getIsDefaultedOnCreate()) { + if ("reference".equals(dataField.getSfType())){ + String reference = dataField.getReferenceTo(); + if (reference == null){ + reference = data.get(j-1).get("Parent_Type").toString(); + } + List> referenceMap = customMapper.list("new_id", reference, "new_id is not null limit 1"); + if (referenceMap.isEmpty()){ + QueryWrapper maxIndex = new QueryWrapper<>(); + maxIndex.select("IFNULL(max(data_index),0) as data_index"); + maxIndex.ne("name", api); + Map map = dataObjectService.getMap(maxIndex); + //如果必填lookup字段没有值，跳过 + update.setDataIndex(Integer.parseInt(map.get("data_index").toString()+1)); + dataObjectService.updateById(update); + return; + }else{ + account.put(dataField.getField(), referenceMap.get(0).get("new_id")); + continue; + } + } + if ("picklist".equals(dataField.getSfType())){ + List> pickList = customMapper.list("value", "data_picklist", "api = '"+api+"' and field = '"+dataField.getField()+"' limit 1"); + account.put(dataField.getField(), pickList.get(0).get("value")); + continue; + } + account.put(dataField.getField(), DataUtil.fieldTypeToSf(dataField)); + } + + // 转换为UTC时间并格式化 + LocalDateTime localDateTime = LocalDateTime.parse(String.valueOf(data.get(j - 1).get("CreatedDate")), inputFormatter); + + ZonedDateTime utcDateTime = localDateTime.atZone(ZoneId.of("UTC")); + + String convertedTime = utcDateTime.format(outputFormatter); + + account.put("CreatedDate", convertedTime); + 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())){ + account.put("CreatedById", CreatedByIdMap.get("new_id")); + } + + } + + ids[j-1] = data.get(j-1).get("Id").toString(); + insertList.add(account); + if (i*2000+j == count){ + break; + } + } + + try { + + //写入csv文件 + String fullPath = CsvConverterUtil.writeToCsv(insertList, UUID.randomUUID().toString()); + + JobInfo salesforceInsertJob = BulkUtil.createJob(bulkConnection, api, OperationEnum.insert); + + List batchInfos = BulkUtil.createBatchesFromCSVFile(bulkConnection, salesforceInsertJob, fullPath); + + BulkUtil.closeJob(bulkConnection, salesforceInsertJob.getId()); + + BulkUtil.awaitCompletion(bulkConnection, salesforceInsertJob, batchInfos); + + sfNum = sfNum + checkInsertResults(bulkConnection, salesforceInsertJob, batchInfos, api, ids); + + } catch (Exception e) { + log.error("manualCreatedNewId error api:{}", api, e); + throw e; + } + } + + 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); + + } + + /** + * 读写Insert结果 + */ + public int checkInsertResults(BulkConnection connection, JobInfo job, + List batchInfoList,String api,String[] ids) + throws AsyncApiException, IOException { + int index = 0; + // batchInfoList was populated when batches were created and submitted + for (BatchInfo b : batchInfoList) { + CSVReader 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)); + } + boolean success = Boolean.valueOf(resultInfo.get("Success")); + boolean created = Boolean.valueOf(resultInfo.get("Created")); + String id = resultInfo.get("Id"); + String error = resultInfo.get("Error"); + if (success && 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 ++; + log.info("Created row with id " + id); + } else if (!success) { + log.info("Failed with error: " + error); + } + } + } + return index; + } + + /** + * Update入口 + */ + @Override + public ReturnT immigrationUpdateBatch(SalesforceParam param) throws Exception { + List> futures = Lists.newArrayList(); + try { + if (StringUtils.isNotBlank(param.getApi())) { + // 手动任务 + ReturnT result = updateSfDataBatch(param, futures); + if (result != null) { + return result; + } + } + return ReturnT.SUCCESS; + } catch (InterruptedException e) { + throw e; + } catch (Throwable throwable) { + salesforceExecutor.remove(futures.toArray(new Future[]{})); + log.error("immigrationUpdate error", throwable); + throw throwable; + } + } + + /** + * 组装执行参数 + */ + public ReturnT updateSfDataBatch(SalesforceParam param, List> futures) throws Exception { + List apis; + String beginDateStr = null; + String endDateStr = null; + apis = DataUtil.toIdList(param.getApi()); + 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"); + } + String join = StringUtils.join(apis, ","); + log.info("immigration apis: {}", join); + XxlJobLogger.log("immigration apis: {}", join); + // 全量的时候 检测是否有自动任务锁住的表 + 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"); + } + } + + BulkConnection bulkConnection = salesforceTargetConnect.createBulkConnect(); + for (String api : apis) { + DataObject update = new DataObject(); + try { + TimeUnit.MILLISECONDS.sleep(1); + 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 { + manualUpdateSfDataBatch(salesforceParam, bulkConnection); + } 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 (Exception e) { + throw e; + } finally { + if (isFull) { + update.setName(api); + update.setDataLock(0); + dataObjectService.updateById(update); + } + } + } + return null; + } + + /** + * 执行数据Update + */ + private void manualUpdateSfDataBatch(SalesforceParam param, BulkConnection bulkConnection) throws Exception { + + String api = param.getApi(); + QueryWrapper dbQw = new QueryWrapper<>(); + dbQw.eq("api", api); + List list = dataFieldService.list(dbQw); + TimeUnit.MILLISECONDS.sleep(1); + + Date beginDate = param.getBeginCreateDate(); + Date endDate = param.getEndCreateDate(); + String beginDateStr = DateUtil.format(beginDate, "yyyy-MM-dd HH:mm:ss"); + String endDateStr = DateUtil.format(endDate, "yyyy-MM-dd HH:mm:ss"); + + // 表内数据总量 + 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; + } + // 总更新数 + int sfNum = 0; + // 批量更新2000一次 + int page = count%2000 == 0 ? count/2000 : (count/2000) + 1; + 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 * 2000 + ",2000"); + + List updateList = new ArrayList<>(); + + for (Map map : mapList) { + JSONObject account = new JSONObject(); + + //给对象赋值 + for (DataField dataField : list) { + String field = dataField.getField(); + String reference_to = dataField.getReferenceTo(); + + //根据旧sfid查找引用对象新sfid + if (field.equals("Id")) { + account.put("Id",String.valueOf(map.get("new_id"))); + } else if (!DataUtil.isUpdate(field) || (dataField.getIsCreateable() != null && !dataField.getIsCreateable())) { + continue; + } else if (StringUtils.isNotBlank(reference_to) && !"data_picklist".equals(reference_to)) { + if (!"null".equals(map.get(field)) && null != map.get(field) && !"OwnerId".equals(field) + || !"Owner_Type".equals(field)) { + //判断reference_to内是否包含User字符串 + if (reference_to.contains("User")) { + reference_to = "User"; + } + Map m = customMapper.getById("new_id", reference_to, String.valueOf(map.get(field))); + if (m != null && !m.isEmpty()) { + account.put(field, m.get("new_id")); + } + } + } else { + if (map.get(field) != null && StringUtils.isNotBlank(dataField.getSfType())) { + account.put(field, DataUtil.localDataToSfData(dataField.getSfType(), String.valueOf(map.get(field)))); + }else { + account.put(field, map.get(field)); + } + } + } + account.put("old_owner_id__c", map.get("OwnerId")); + account.put("old_sfdc_id__c", map.get("Id")); + + updateList.add(account); + } + + try { + //写入csv文件 + String fullPath = CsvConverterUtil.writeToCsv(updateList, UUID.randomUUID().toString()); + + JobInfo salesforceInsertJob = BulkUtil.createJob(bulkConnection, api, OperationEnum.update); + + List batchInfos = BulkUtil.createBatchesFromCSVFile(bulkConnection, salesforceInsertJob, fullPath); + + BulkUtil.closeJob(bulkConnection, salesforceInsertJob.getId()); + + BulkUtil.awaitCompletion(bulkConnection, salesforceInsertJob, batchInfos); + + sfNum = sfNum + checkUpdateResults(bulkConnection, salesforceInsertJob, batchInfos); + + } catch (Throwable e) { + log.info(e.getMessage()); + throw e; + } + } + + UpdateWrapper updateQw = new UpdateWrapper<>(); + updateQw.eq("name", api) + .eq("sync_start_date", beginDate) + .eq("sync_end_date", DateUtils.addSeconds(endDate, -1)) + .set("target_update_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_update_num", sfNum); + dataBatchService.update(updateQw2); + + } + + /** + * 读写update结果 + */ + public int checkUpdateResults(BulkConnection connection, JobInfo job, + List batchInfoList) + throws AsyncApiException, IOException { + int index = 0; + // batchInfoList was populated when batches were created and submitted + for (BatchInfo b : batchInfoList) { + CSVReader 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)); + } + boolean success = Boolean.valueOf(resultInfo.get("Success")); + String id = resultInfo.get("Id"); + String error = resultInfo.get("Error"); + if (success) { + index ++; + log.info("Update row with id " + id); + } else if (!success) { + log.info("Failed with error: " + error); + } + } + } + return index; + } + + + +} 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 aea5bbc..66c12a5 100644 --- a/src/main/java/com/celnet/datadump/service/impl/DataImportServiceImpl.java +++ b/src/main/java/com/celnet/datadump/service/impl/DataImportServiceImpl.java @@ -34,28 +34,15 @@ import org.apache.commons.collections.CollectionUtils; import org.apache.commons.lang.time.DateUtils; import org.apache.commons.lang3.StringUtils; import org.springframework.beans.factory.annotation.Autowired; -import org.springframework.boot.autoconfigure.batch.BatchProperties; import org.springframework.stereotype.Service; - -import javax.swing.text.Utilities; -import java.io.ByteArrayInputStream; -import java.io.FileInputStream; -import java.io.IOException; -import java.io.InputStream; import java.text.ParseException; import java.text.SimpleDateFormat; -import java.time.LocalDateTime; -import java.time.ZoneId; -import java.time.ZonedDateTime; -import java.time.format.DateTimeFormatter; import java.util.*; import java.util.concurrent.Future; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; import java.util.stream.Collectors; -import static com.celnet.datadump.global.SystemConfigCode.EXECUTOR_SIZE; - @Service @Slf4j public class DataImportServiceImpl implements DataImportService { @@ -113,24 +100,7 @@ public class DataImportServiceImpl implements DataImportService { } } - @Override - public ReturnT immigrationBatch(SalesforceParam param) throws Exception { - List> futures = Lists.newArrayList(); - try { - if (StringUtils.isNotBlank(param.getApi())) { - // 手动任务 - ReturnT result = manualImmigrationBatch(param, futures); - if (result != null) { - return result; - } - } - return ReturnT.SUCCESS; - } catch (Exception exception) { - salesforceExecutor.remove(futures.toArray(new Future[]{})); - log.error("immigration error", exception); - throw exception; - } - } + @Override public ReturnT manualImmigration(SalesforceParam param, List> futures) throws Exception { @@ -212,84 +182,6 @@ public class DataImportServiceImpl implements DataImportService { return null; } - public ReturnT manualImmigrationBatch(SalesforceParam param, List> futures) throws Exception { - List apis; - apis = DataUtil.toIdList(param.getApi()); - String join = StringUtils.join(apis, ","); - log.info("immigration apis: {}", join); - XxlJobLogger.log("immigration apis: {}", join); - - TimeUnit.MILLISECONDS.sleep(1); - - // 全量的时候 检测是否有自动任务锁住的表 - 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"); - } - } - - BulkConnection bulkConnection = salesforceTargetConnect.createBulkConnect(); - for (String api : apis) { - DataObject update = new DataObject(); - TimeUnit.MILLISECONDS.sleep(1); - try { - List salesforceParams = null; - - update.setName(api); - update.setDataLock(1); - dataObjectService.updateById(update); - - 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 -> { - 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 { - manualCreatedNewIdBatch(salesforceParam, bulkConnection); - } 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 (InterruptedException e) { - throw e; - } catch (Throwable e) { - log.error("manualImmigration error", e); - throw new RuntimeException(e); - } finally { - if (isFull) { - update.setName(api); - update.setDataLock(0); - dataObjectService.updateById(update); - } - } - } - return null; - } @Override @@ -534,205 +426,6 @@ public class DataImportServiceImpl implements DataImportService { dataBatchService.update(updateQw2); } - private void manualCreatedNewIdBatch(SalesforceParam param, BulkConnection bulkConnection) throws Exception { - String api = param.getApi(); - QueryWrapper dbQw = new QueryWrapper<>(); - dbQw.eq("api", api); - List list = dataFieldService.list(dbQw); - TimeUnit.MILLISECONDS.sleep(1); - - Date beginDate = param.getBeginCreateDate(); - Date endDate = param.getEndCreateDate(); - String beginDateStr = DateUtil.format(beginDate, "yyyy-MM-dd HH:mm:ss"); - String endDateStr = DateUtil.format(endDate, "yyyy-MM-dd HH:mm:ss"); - - //表内数据总量 - 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; - } - - //批量插入2000一次 - int page = count%2000 == 0 ? count/2000 : (count/2000) + 1; - - //总插入数 - 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 2000"); - int size = data.size(); - - log.info("执行api:{}, 执行page:{}, 执行size:{}", api, i+1, size); - List insertList = new ArrayList<>(); - - //判断引用对象是否存在new_id - DataObject update = new DataObject(); - update.setName(api); - - //更新对象的new_id - String[] ids = new String[size]; - - // 定义输入/输出格式 - DateTimeFormatter inputFormatter = DateTimeFormatter.ISO_LOCAL_DATE_TIME; - DateTimeFormatter outputFormatter = DateTimeFormatter.ofPattern("yyyy-MM-dd'T'HH:mm:ss.SSSxx"); - - for (int j = 1; j <= size; j++) { - JSONObject account = new JSONObject(); - for (DataField dataField : list) { - if ("OwnerId".equals(dataField.getField()) || "Owner_Type".equals(dataField.getField()) - || "Id".equals(dataField.getField())){ - continue; - } - if (dataField.getIsCreateable() && !dataField.getIsNillable() && !dataField.getIsDefaultedOnCreate()) { - if ("reference".equals(dataField.getSfType())){ - String reference = dataField.getReferenceTo(); - if (reference == null){ - reference = data.get(j-1).get("Parent_Type").toString(); - } - List> referenceMap = customMapper.list("new_id", reference, "new_id is not null limit 1"); - if (referenceMap.isEmpty()){ - QueryWrapper maxIndex = new QueryWrapper<>(); - maxIndex.select("IFNULL(max(data_index),0) as data_index"); - maxIndex.ne("name", api); - Map map = dataObjectService.getMap(maxIndex); - //如果必填lookup字段没有值，跳过 - update.setDataIndex(Integer.parseInt(map.get("data_index").toString()+1)); - dataObjectService.updateById(update); - return; - }else{ - account.put(dataField.getField(), referenceMap.get(0).get("new_id")); - continue; - } - } - if ("picklist".equals(dataField.getSfType())){ - List> pickList = customMapper.list("value", "data_picklist", "api = '"+api+"' and field = '"+dataField.getField()+"' limit 1"); - account.put(dataField.getField(), pickList.get(0).get("value")); - continue; - } - account.put(dataField.getField(), DataUtil.fieldTypeToSf(dataField)); - } - - // 转换为UTC时间并格式化 - LocalDateTime localDateTime = LocalDateTime.parse(String.valueOf(data.get(j - 1).get("CreatedDate")), inputFormatter); - - ZonedDateTime utcDateTime = localDateTime.atZone(ZoneId.of("UTC")); - - String convertedTime = utcDateTime.format(outputFormatter); - - account.put("CreatedDate", convertedTime); - 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())){ - account.put("CreatedById", CreatedByIdMap.get("new_id")); - } - - } - - ids[j-1] = data.get(j-1).get("Id").toString(); - insertList.add(account); - if (i*2000+j == count){ - break; - } - } - - try { - - //写入csv文件 - String fullPath = CsvConverterUtil.writeToCsv(insertList, UUID.randomUUID().toString()); - - JobInfo salesforceInsertJob = createSalesforceJob(bulkConnection, api, OperationEnum.insert); - - List batchInfos = BulkUtil.createBatchesFromCSVFile(bulkConnection, salesforceInsertJob, fullPath); - - BulkUtil.closeJob(bulkConnection, salesforceInsertJob.getId()); - - BulkUtil.awaitCompletion(bulkConnection, salesforceInsertJob, batchInfos); - - sfNum = sfNum + checkInsertResults(bulkConnection, salesforceInsertJob, batchInfos, api, ids); - - } catch (Exception e) { - log.error("manualCreatedNewId error api:{}", api, e); - throw e; - } - } - - 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); - - } - - /** - * Gets the results of the operation and checks for errors. - */ - private int checkInsertResults(BulkConnection connection, JobInfo job, - List batchInfoList,String api,String[] ids) - throws AsyncApiException, IOException { - int index = 0; - // batchInfoList was populated when batches were created and submitted - for (BatchInfo b : batchInfoList) { - CSVReader 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)); - } - boolean success = Boolean.valueOf(resultInfo.get("Success")); - boolean created = Boolean.valueOf(resultInfo.get("Created")); - String id = resultInfo.get("Id"); - String error = resultInfo.get("Error"); - if (success && 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 ++; - log.info("Created row with id " + id); - } else if (!success) { - log.info("Failed with error: " + error); - } - } - } - return index; - } - - - - /** - * 创建任务 - * @param bulkConn - * @param objectName - * @param operation - * @return - * @throws AsyncApiException - */ - private static JobInfo createSalesforceJob(BulkConnection bulkConn, String objectName, OperationEnum operation) - throws AsyncApiException { - - JobInfo job = new JobInfo(); - job.setObject(objectName); - job.setOperation(operation); - job.setContentType(ContentType.CSV); // 使用 CSV 格式 - return bulkConn.createJob(job); - } @@ -923,6 +616,7 @@ public class DataImportServiceImpl implements DataImportService { } } + public ReturnT updateSfData(SalesforceParam param, List> futures) throws Exception { List apis; String beginDateStr = null; @@ -1193,22 +887,6 @@ public class DataImportServiceImpl implements DataImportService { } - public void printAccountsDetail(SObject[] accounts,List list) { - SObject account = accounts[0]; - - // 获取对象所有字段名 - for (DataField dataField : list) { - try { - Object value = account.getField(dataField.getField()); - System.out.println(dataField.getField() + ": " + (value != null ? value.toString() : "null")); - } catch (Exception e) { - System.out.println(dataField.getField() + ": [权限不足或字段不存在]"); - } - } - System.out.println("old_owner_id__c: " + (account.getField("old_owner_id__c") != null ? account.getField("old_owner_id__c").toString() : "null")); - System.out.println("old_sfdc_id__c: " + (account.getField("old_sfdc_id__c") != null ? account.getField("old_sfdc_id__c").toString() : "null")); - - } public void printAccountsDetails(SObject[] accounts,List list) { for (int i = 0; i < accounts.length; i++) { diff --git a/src/main/java/com/celnet/datadump/util/BulkUtil.java b/src/main/java/com/celnet/datadump/util/BulkUtil.java index cf56f35..9c308fa 100644 --- a/src/main/java/com/celnet/datadump/util/BulkUtil.java +++ b/src/main/java/com/celnet/datadump/util/BulkUtil.java @@ -70,17 +70,18 @@ public class BulkUtil { * @return The JobInfo for the new job. * @throws AsyncApiException */ - public static JobInfo createJob(String sobjectType, BulkConnection connection) + public static JobInfo createJob( BulkConnection connection,String sobjectType,OperationEnum operation) throws AsyncApiException { JobInfo job = new JobInfo(); job.setObject(sobjectType); - job.setOperation(OperationEnum.insert); + job.setOperation(operation); job.setContentType(ContentType.CSV); job = connection.createJob(job); - System.out.println(job); return job; } + + /** * Create the BulkConnection used to call Bulk API operations. */