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-08-20 10:40:44 +08:00
parent 12e4a5a5e5
commit e3e1f58482
8 changed files with 128 additions and 90 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

68
src/main/java/com/celnet/datadump/aspect/OperateLogAspect.java
View File

@ -3,9 +3,8 @@ package com.celnet.datadump.aspect;
import com.alibaba.fastjson.JSON;
import com.celnet.datadump.annotation.LogServiceAnnotation;
import com.celnet.datadump.entity.DataLog;
import com.celnet.datadump.service.DataLogService;
import com.google.common.collect.Lists;
import org.aspectj.lang.ProceedingJoinPoint;
import org.aspectj.lang.Signature;
import org.aspectj.lang.annotation.Around;
@ -66,19 +65,10 @@ public class OperateLogAspect {
//执行程序
result = joinPoint.proceed();
//初始化日志记录
DataLog dataLog = initializeOperateLog(joinPoint,args,startTime,logServiceAnno,result,null);
//保存日志
dataLogService.save(dataLog);
} catch (Throwable throwable) {
//初始化日志记录
DataLog dataLog = initializeOperateLog(joinPoint,args,startTime,logServiceAnno,null,throwable);
//保存日志
dataLogService.save(dataLog);
log.error("日志拦截异常："+throwable);
}
@ -86,62 +76,6 @@ public class OperateLogAspect {
return result;
}
/**
* 初始化操作日志
* @param joinPoint 节点
*/
private DataLog initializeOperateLog(ProceedingJoinPoint joinPoint , Object[] args , Date startTime , LogServiceAnnotation logServiceAnno , Object result , Throwable throwable) {
if(logServiceAnno == null){
return null;
}
//request请求
RequestAttributes requestAttributes = RequestContextHolder.getRequestAttributes();
HttpServletRequest request = (HttpServletRequest) requestAttributes.resolveReference(RequestAttributes.REFERENCE_REQUEST);
//ip
String ip = request.getHeader("HTTP_X_FORWARDED_FOR");
//请求参数
Object[] arguments = new Object[args.length];
for (int i = 0; i < args.length; i++) {
if (args[i] instanceof ServletRequest || args[i] instanceof ServletResponse || args[i] instanceof MultipartFile) {
//ServletRequest不能序列化，从入参里排除，否则报异常：java.lang.IllegalStateException: It is illegal to call this method if the current request is not in asynchronous mode (i.e. isAsyncStarted() returns false)
//ServletResponse不能序列化 从入参里排除，否则报异常：java.lang.IllegalStateException: getOutputStream() has already been called for this response
continue;
}
arguments[i] = args[i];
}
String argStr = arguments.length <= 0 ? "" : JSON.toJSONString(arguments);
//响应结果
String resultStr = result == null ? "" : JSON.toJSONString(result);
//异常信息
String exceptionStr = throwable == null ? "" : JSON.toJSONString(throwable.getMessage());
//结束时间
Date endTime = new Date();
DataLog dataLog = new DataLog();
dataLog.setIp(ip);
dataLog.setStartTime(startTime);
dataLog.setEndTime(endTime);
if(resultStr.contains("200")){
dataLog.setCode("200");
dataLog.setStatus("成功");
dataLog.setMessage(resultStr);
} else {
dataLog.setCode("500");
dataLog.setStatus("失败");
dataLog.setMessage(exceptionStr);
}
dataLog.setRequestUrl(request.getRequestURI());
dataLog.setRequestType(logServiceAnno.operateType());
dataLog.setRequestData(argStr);
dataLog.setRequestMethod(logServiceAnno.remark());
return dataLog;
}
/**
* 获取注解

13
src/main/java/com/celnet/datadump/job/DataDumpByParentJob.java
View File

@ -1,8 +1,11 @@
package com.celnet.datadump.job;
import com.alibaba.fastjson.JSON;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.celnet.datadump.entity.DataObject;
import com.celnet.datadump.param.DataDumpByParentParam;
import com.celnet.datadump.service.DataDumpByParentService;
import com.celnet.datadump.service.DataObjectService;
import com.xxl.job.core.biz.model.ReturnT;
import com.xxl.job.core.handler.annotation.XxlJob;
import lombok.extern.slf4j.Slf4j;
@ -22,6 +25,9 @@ public class DataDumpByParentJob {
@Autowired
private DataDumpByParentService dataDumpByParentService;
@Autowired
private DataObjectService dataObjectService;
/**
* 增量任务
*
@ -47,6 +53,13 @@ public class DataDumpByParentJob {
param.setParentApi(StringUtils.deleteWhitespace(param.getParentApi()));
param.setParentField(StringUtils.deleteWhitespace(param.getParentField()));
if (!param.getType().equals("1")) {
QueryWrapper<DataObject> qw = new QueryWrapper<>();
qw.eq("name", param.getApi());
DataObject update = dataObjectService.getOne(qw);
param.setBeginModifyDate(update.getLastUpdateDate());
}
return dataDumpByParentService.dataDumpByParent(param);
}

3
src/main/java/com/celnet/datadump/mapper/CustomMapper.java
View File

@ -152,4 +152,7 @@ public interface CustomMapper {
public Integer countBySQL(@Param("api") String api, @Param("sql") String sql);
public void delete(@Param("tableName") String tableName, @Param("ids") List<String> ids);
public void deleteOne(@Param("tableName") String tableName, @Param("id") String id);
}

9
src/main/java/com/celnet/datadump/service/impl/DataDumpByParentServiceImpl.java
View File

@ -58,8 +58,7 @@ public class DataDumpByParentServiceImpl implements DataDumpByParentService {
@Override
public ReturnT<String> dataDumpByParent(DataDumpByParentParam param) throws Throwable {
try {
int startPage = 1;
int size = 100, page = 0;
// 检测父表
if (StringUtils.isBlank(customMapper.checkTable(param.getParentApi()))) {
return new ReturnT<>(500, ("表" + param.getParentApi() + "不存在"));
@ -77,8 +76,9 @@ public class DataDumpByParentServiceImpl implements DataDumpByParentService {
dataObjectService.updateById(update);
}
int page = 0;
while (true) {
List<String> strings = customMapper.listById(param.getParentField(), param.getParentApi(), "order by " + param.getParentField() + " asc limit " + page + "," + size);
List<String> strings = customMapper.listById(param.getParentField(), param.getParentApi(), "order by " + param.getParentField() + " asc limit " + page * 100 + ",100");
if (CollectionUtils.isEmpty(strings)) {
break;
}
@ -144,8 +144,7 @@ public class DataDumpByParentServiceImpl implements DataDumpByParentService {
TimeUnit.MINUTES.sleep(1);
}
}
startPage++;
page = (startPage - 1) * size;
page ++;
}
return ReturnT.SUCCESS;
} catch (Throwable throwable) {

4
src/main/java/com/celnet/datadump/service/impl/DataDumpSpecialServiceImpl.java
View File

@ -7,6 +7,7 @@ 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.DataLog;
import com.celnet.datadump.entity.DataObject;
import com.celnet.datadump.global.Const;
import com.celnet.datadump.param.DataDumpSpecialParam;
@ -172,13 +173,14 @@ public class DataDumpSpecialServiceImpl implements DataDumpSpecialService {
salesforceParam.setSelect(StringUtils.join(fields, ","));
}
JSONArray objects = null;
int count = 0, failCount = 0;
int count = 0, failCount = 0, batch = 0;
// 遍历
while (true) {
try {
// 判断是否存在要排除的id
salesforceParam.setMaxId(maxId);
map.put("param", salesforceParam);
DataLog dataLog = new DataLog(api, "拉取" + api + "对象数据", new Date(), null, "拉取第" + batch + "批数据", "Pull");
String sql = SqlUtil.showSql("com.celnet.datadump.mapper.SalesforceMapper.listOrderById", map);
log.info("query sql: {}", sql);
XxlJobLogger.log("query sql: {}", sql);

110
src/main/java/com/celnet/datadump/service/impl/DataImportNewServiceImpl.java
View File

@ -97,6 +97,10 @@ public class DataImportNewServiceImpl implements DataImportNewService {
boolean mkdir = excel.mkdir();
}
}
@Autowired
private EmailUtil emailUtil;
/**
* Get返写个人客户联系人入口
*/
@ -1165,7 +1169,7 @@ public class DataImportNewServiceImpl implements DataImportNewService {
//批量插入200一次
int page = count % 200 == 0 ? count / 200 : (count / 200) + 1;
for (int i = 0; i < page; i++) {
List<Map<String, Object>> linkList = customMapper.list("Id,LinkedEntityId,ContentDocumentId,LinkedEntity_Type,ShareType,Visibility", api, "ShareType = 'V' and new_id = '0' order by Id limit " + i * 200 + ",200");
List<Map<String, Object>> linkList = customMapper.list("Id,LinkedEntityId,ContentDocumentId,LinkedEntity_Type,ShareType,Visibility", api, "ShareType = 'V' and new_id = '0' order by Id limit 200");
SObject[] accounts = new SObject[linkList.size()];
String[] ids = new String[linkList.size()];
int index = 0;
@ -1184,11 +1188,20 @@ public class DataImportNewServiceImpl implements DataImportNewService {
if (!objects.isEmpty()) {
Map<String, Object> dMap = customMapper.getById("new_id", "ContentDocument", contentDocumentId);
Map<String, Object> lMap = customMapper.getById("new_id", linkedEntityType, linkedEntityId);
SObject account = new SObject();
account.setType(api);
account.setField("ContentDocumentId", dMap.get("new_id").toString());
account.setField("LinkedEntityId", lMap.get("new_id").toString());
if (dMap != null){
account.setField("ContentDocumentId", dMap.get("new_id").toString());
}else {
log.info("ContentDocumentLink Id: {}，对应的ContentDocumentId: {} 数据不存在！ " , id, contentDocumentId);
continue;
}
if (lMap != null){
account.setField("LinkedEntityId", lMap.get("new_id").toString());
}else {
log.info("ContentDocumentLink Id: {}，对应的LinkedEntityId: {} 数据不存在！ " , id, linkedEntityId);
continue;
}
account.setField("ShareType", shareType);
account.setField("Visibility", Visibility);
ids[index] = id;
@ -1925,6 +1938,11 @@ public class DataImportNewServiceImpl implements DataImportNewService {
for (String api : apis) {
try {
if (!dataFieldService.hasCreatedDate(api)){
log.info("当前对象："+api +"不存在批次，请执行参数更改type为1，按对象更新关联！");
EmailUtil.send("DataDump ERROR", "当前对象："+api +"不存在批次，请执行参数更改type为1，按对象更新关联！");
break;
}
QueryWrapper<LinkConfig> dbQw = new QueryWrapper<>();
dbQw.eq("api", param.getApi()).eq("is_link",1).eq("is_create",1);
List<LinkConfig> linkConfigs = linkConfigService.list(dbQw);
@ -2121,9 +2139,10 @@ public class DataImportNewServiceImpl implements DataImportNewService {
if (beginDateStr != null){
String updateDateField = dataFieldService.returnUpdateDateField(api);
sql1 = "where " + updateDateField + " >= '" + beginDateStr + "'";
sql2 = updateDateField + " >= '"+ beginDateStr +"' order by Id limit " ;
sql2 = updateDateField + " >= '"+ beginDateStr +"' " ;
}else {
sql2 = "1=1 order by Id limit " ;
sql1 = "where 1=1";
sql2 = "1=1 " ;
}
String allFiled = "Id";
@ -2161,7 +2180,7 @@ public class DataImportNewServiceImpl implements DataImportNewService {
log.info("表api：{}，批次：{}，批次量：{}，执行数据更新!", api,i, 10000);
List<Map<String, Object>> mapList = customMapper.list(allFiled, api, sql2 + i * 10000 + ",10000" );
List<Map<String, Object>> mapList = customMapper.list(allFiled, api, sql2 +" order by Id limit "+ + i * 10000 + ",10000" );
for (int j = 1; j <= mapList.size(); j++) {
List<Map<String, Object>> updateMapList = new ArrayList<>();
@ -2204,6 +2223,76 @@ public class DataImportNewServiceImpl implements DataImportNewService {
private ReturnT<String> munalCheckDeletedData(SalesforceParam param) throws Exception {
String api = "DeleteEvent";
PartnerConnection connect = salesforceTargetConnect.createConnect();
if ((param.getEndCreateDate() == null || param.getBeginCreateDate() == null) && param.getIds().isEmpty()){
log.info("手动删除数据必须指定参数：开始与结束时间/Ids !!!!");
return ReturnT.SUCCESS;
}
String sql = "";
if (!param.getIds().isEmpty()){
sql = "where Record in (" + param.getIds().stream().map(id -> "'" + id + "'").collect(Collectors.joining(",")) + ")";
}else {
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");
sql = "where CreatedDate >= '" + beginDateStr + "' and CreatedDate < '" + endDateStr + "'";
}
Integer count = customMapper.countBySQL(api, sql);
int page = count%200 == 0 ? count/200 : (count/200) + 1;
log.info("删除数据，总批次：{}",page);
for (int i = 0; i < page; i++) {
List<Map<String, Object>> list = customMapper.list("Record, SobjectName", api, sql + " order by Id asc limit " + i * 200 + ",200");
log.info("删除数据，当前批次：{}，批次数：{}", i,list.size());
Map<String, String> idNameMap = new HashMap<>();
String[] ids = new String[list.size()];
for (int z = 0; z < list.size(); z++) {
Map<String, Object> objectMap = customMapper.getById("*", list.get(z).get("SobjectName").toString(), list.get(z).get("Record").toString());
if (objectMap != null && objectMap.get("new_id") != null) {
ids[z] = objectMap.get("new_id").toString();
}
idNameMap.put(objectMap.get("new_id").toString(), list.get(z).get("SobjectName").toString());
}
int index = 0;
DeleteResult[] deleteResults = connect.delete(ids);
for (DeleteResult deleteResult : deleteResults) {
if (deleteResult.isSuccess()){
String typeName = idNameMap.get(ids[index]);
Map<String, Object> objectMap = customMapper.getById("*", typeName, ids[index]);
List<Map<String, Object>> maps = new ArrayList<>();
Map<String, Object> m1 = new HashMap<>();
m1.put("key", "is_delete");
m1.put("value", 1);
Map<String, Object> m2 = new HashMap<>();
m2.put("key", "all_data");
m2.put("value", JSON.toJSONString(objectMap));
maps.add(m1);
maps.add(m2);
customMapper.updateById(api, maps, ids[index]);
}else {
log.error("ID:{},删除数据失败：{}", deleteResult.getId(),deleteResult.getErrors());
}
index++;
}
}
return ReturnT.SUCCESS;
}
@ -2843,7 +2932,6 @@ public class DataImportNewServiceImpl implements DataImportNewService {
account.setField(field, m.get("new_id"));
}else {
String message = "对象类型:" + api + "的数据:"+ data.get(j - 1).get("Id") +"的引用对象:" + reference_to + "的数据:"+ data.get(j - 1).get(field) +"不存在！";
EmailUtil.send("DataDump ERROR", message);
log.info(message);
break;
}
@ -3039,7 +3127,6 @@ public class DataImportNewServiceImpl implements DataImportNewService {
account.setField(field, m.get("new_id"));
}else {
String message = "对象类型:" + api + "的数据:"+ data.get(j - 1).get("Id") +"的引用对象:" + reference_to + "的数据:"+ data.get(j - 1).get(field) +"不存在！";
EmailUtil.send("DataDump ERROR", message);
log.info(message);
break;
}
@ -3077,13 +3164,16 @@ public class DataImportNewServiceImpl implements DataImportNewService {
m.put("value", saveResult.getId());
maps.add(m);
customMapper.updateById(api, maps, ids[index]);
log.info("Created Success row with id " + saveResult.getId());
}else{
log.error("Id:{},saveResults: {}",ids[index], JSON.toJSONString(saveResult));
}
index++;
}
TimeUnit.MILLISECONDS.sleep(1);
} catch (Exception e) {
}catch (InterruptedException e){
return;
}catch (Exception e) {
log.error("insertSingle error api:{},错误信息：{}", api, JSON.toJSONString(e));
}
}

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

@ -1,10 +1,7 @@
package com.celnet.datadump.util;
import com.alibaba.fastjson.JSON;
import com.celnet.datadump.entity.DataLog;
import com.celnet.datadump.mapper.CustomMapper;
import com.celnet.datadump.service.DataLogService;
import com.google.common.collect.Lists;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang3.ArrayUtils;
@ -117,10 +114,6 @@ public class EmailUtil implements ApplicationContextAware {
helper.setText(text, true);
JAVA_MAIL_SENDER.send(message);
DataLogService dataLogService = EmailUtil.getBean(DataLogService.class);
DataLog dataLog = new DataLog();
dataLog.setMessage(text);
dataLogService.save(dataLog);
} catch (MessagingException e) {
log.error("send mine email error", e);

4
src/main/resources/mapper/CustomMapper.xml
View File

@ -216,4 +216,8 @@
</foreach>
</if>
</delete>
<delete id="deleteOne">
delete from `${tableName}` where Id = #{d}
</delete>
</mapper>