kris/data-dump
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

【feat】 调整代码结构，新增更新大数据量数据方法

Browse Source
This commit is contained in:
Kris 2025-06-06 15:24:29 +08:00
parent 7048ea42b2
commit 5ea0ff490b
7 changed files with 690 additions and 331 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

32
src/main/java/com/celnet/datadump/controller/JobController.java
View File

@ -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<String> 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);
}
}

31
src/main/java/com/celnet/datadump/job/DataDumpJob.java
View File

@ -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<String> 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);
}
/**
* 获取文件关联表
*

13
src/main/java/com/celnet/datadump/service/DataImportBatchService.java Normal file
View File

@ -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<String> immigrationBatch(SalesforceParam param) throws Exception;
ReturnT<String> immigrationUpdateBatch(SalesforceParam param) throws Exception;
}

3
src/main/java/com/celnet/datadump/service/DataImportService.java
View File

@ -14,7 +14,6 @@ public interface DataImportService {
ReturnT<String> immigration(SalesforceParam param) throws Exception;
ReturnT<String> immigrationBatch(SalesforceParam param) throws Exception;
ReturnT<String> immigrationUpdate(SalesforceParam param) throws Exception;
}

609
src/main/java/com/celnet/datadump/service/impl/DataImportBatchServiceImpl.java Normal file
View File

@ -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<String> immigrationBatch(SalesforceParam param) throws Exception {
List<Future<?>> futures = Lists.newArrayList();
try {
if (StringUtils.isNotBlank(param.getApi())) {
// 手动任务
ReturnT<String> 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<String> manualImmigrationBatch(SalesforceParam param, List<Future<?>> futures) throws Exception {
List<String> 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<DataObject> qw = new QueryWrapper<>();
qw.eq("data_lock", 1).in("name", apis);
List<DataObject> 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<SalesforceParam> salesforceParams = null;
update.setName(api);
update.setDataLock(1);
dataObjectService.updateById(update);
QueryWrapper<DataBatch> dbQw = new QueryWrapper<>();
dbQw.eq("name", api);
List<DataBatch> 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<DataField> dbQw = new QueryWrapper<>();
dbQw.eq("api", api);
List<DataField> 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<JSONObject> 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<JSONObject> 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<Map<String, Object>> referenceMap = customMapper.list("new_id", reference, "new_id is not null limit 1");
if (referenceMap.isEmpty()){
QueryWrapper<DataObject> maxIndex = new QueryWrapper<>();
maxIndex.select("IFNULL(max(data_index),0) as data_index");
maxIndex.ne("name", api);
Map<String, Object> 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<Map<String, Object>> 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<String, Object> 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<BatchInfo> 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<DataBatchHistory> 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<DataBatch> 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<BatchInfo> 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<String> resultHeader = rdr.nextRecord();
int resultCols = resultHeader.size();
List<String> row;
while ((row = rdr.nextRecord()) != null) {
Map<String, String> resultInfo = new HashMap<String, String>();
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<Map<String, Object>> maps = new ArrayList<>();
Map<String, Object> 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<String> immigrationUpdateBatch(SalesforceParam param) throws Exception {
List<Future<?>> futures = Lists.newArrayList();
try {
if (StringUtils.isNotBlank(param.getApi())) {
// 手动任务
ReturnT<String> 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<String> updateSfDataBatch(SalesforceParam param, List<Future<?>> futures) throws Exception {
List<String> 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<DataObject> qw = new QueryWrapper<>();
qw.eq("data_lock", 1).in("name", apis);
List<DataObject> 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<SalesforceParam> salesforceParams = null;
QueryWrapper<DataBatch> 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<DataBatch> 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<DataField> dbQw = new QueryWrapper<>();
dbQw.eq("api", api);
List<DataField> 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<Map<String, Object>> mapList = customMapper.list("*", api, "new_id is not null and CreatedDate >= '" + beginDateStr + "' and CreatedDate < '" + endDateStr + "' order by Id asc limit " + i * 2000 + ",2000");
List<JSONObject> updateList = new ArrayList<>();
for (Map<String, Object> 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<String, Object> 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<BatchInfo> 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<DataBatchHistory> 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<DataBatch> 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<BatchInfo> 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<String> resultHeader = rdr.nextRecord();
int resultCols = resultHeader.size();
List<String> row;
while ((row = rdr.nextRecord()) != null) {
Map<String, String> resultInfo = new HashMap<String, String>();
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;
}
}

326
src/main/java/com/celnet/datadump/service/impl/DataImportServiceImpl.java
View File

@ -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<String> immigrationBatch(SalesforceParam param) throws Exception {
List<Future<?>> futures = Lists.newArrayList();
try {
if (StringUtils.isNotBlank(param.getApi())) {
// 手动任务
ReturnT<String> 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<String> manualImmigration(SalesforceParam param, List<Future<?>> futures) throws Exception {
@ -212,84 +182,6 @@ public class DataImportServiceImpl implements DataImportService {
return null;
}
public ReturnT<String> manualImmigrationBatch(SalesforceParam param, List<Future<?>> futures) throws Exception {
List<String> 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<DataObject> qw = new QueryWrapper<>();
qw.eq("data_lock", 1).in("name", apis);
List<DataObject> 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<SalesforceParam> salesforceParams = null;
update.setName(api);
update.setDataLock(1);
dataObjectService.updateById(update);
QueryWrapper<DataBatch> dbQw = new QueryWrapper<>();
dbQw.eq("name", api);
List<DataBatch> 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<DataField> dbQw = new QueryWrapper<>();
dbQw.eq("api", api);
List<DataField> 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<JSONObject> 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<JSONObject> 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<Map<String, Object>> referenceMap = customMapper.list("new_id", reference, "new_id is not null limit 1");
if (referenceMap.isEmpty()){
QueryWrapper<DataObject> maxIndex = new QueryWrapper<>();
maxIndex.select("IFNULL(max(data_index),0) as data_index");
maxIndex.ne("name", api);
Map<String, Object> 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<Map<String, Object>> 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<String, Object> 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<BatchInfo> 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<DataBatchHistory> 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<DataBatch> 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<BatchInfo> 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<String> resultHeader = rdr.nextRecord();
int resultCols = resultHeader.size();
List<String> row;
while ((row = rdr.nextRecord()) != null) {
Map<String, String> resultInfo = new HashMap<String, String>();
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<Map<String, Object>> maps = new ArrayList<>();
Map<String, Object> 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<String> updateSfData(SalesforceParam param, List<Future<?>> futures) throws Exception {
List<String> apis;
String beginDateStr = null;
@ -1193,22 +887,6 @@ public class DataImportServiceImpl implements DataImportService {
}
public void printAccountsDetail(SObject[] accounts,List<DataField> 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<DataField> list) {
for (int i = 0; i < accounts.length; i++) {

7
src/main/java/com/celnet/datadump/util/BulkUtil.java
View File

@ -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.
*/