165 lines
7.8 KiB
Java
165 lines
7.8 KiB
Java
package com.yunhai.bschclassmanage.service.impl;
|
||
|
||
import com.alibaba.fastjson.JSON;
|
||
import com.yunhai.bschclassmanage.Repository.SyncLogRepository;
|
||
import com.yunhai.bschclassmanage.entity.SyncLog;
|
||
import com.yunhai.bschclassmanage.model.vo.PageVO;
|
||
import com.yunhai.bschclassmanage.service.ISyncLogService;
|
||
import com.yunhai.bschclassmanage.utils.ReflectRetryUtil;
|
||
import com.yunhai.bschclassmanage.utils.SyncLogRetryContext;
|
||
import com.yunhai.platform.utils.exception.DException;
|
||
import lombok.extern.slf4j.Slf4j;
|
||
import org.apache.commons.lang3.StringUtils;
|
||
import org.springframework.beans.factory.annotation.Autowired;
|
||
import org.springframework.context.ApplicationContext;
|
||
import org.springframework.data.domain.Page;
|
||
import org.springframework.data.domain.PageRequest;
|
||
import org.springframework.data.domain.Pageable;
|
||
import org.springframework.data.domain.Sort;
|
||
import org.springframework.data.jpa.domain.Specification;
|
||
import org.springframework.scheduling.annotation.Async;
|
||
import org.springframework.stereotype.Service;
|
||
|
||
import javax.persistence.criteria.Predicate;
|
||
import java.lang.reflect.InvocationTargetException;
|
||
import java.lang.reflect.Method;
|
||
import java.lang.reflect.Parameter;
|
||
import java.time.LocalDateTime;
|
||
import java.util.ArrayList;
|
||
import java.util.LinkedHashMap;
|
||
import java.util.List;
|
||
import java.util.Map;
|
||
|
||
@Slf4j
|
||
@Service
|
||
public class SyncLogServiceImpl implements ISyncLogService {
|
||
|
||
@Autowired
|
||
private SyncLogRepository repository;
|
||
|
||
@Autowired
|
||
private ApplicationContext applicationContext;
|
||
|
||
@Async("logExecutor")
|
||
@Override
|
||
public void saveAsync(SyncLog entity) {
|
||
try {
|
||
repository.save(entity);
|
||
} catch (Exception e) {
|
||
log.error("同步日志异步落库失败, classMethod={}", entity.getClassMethod(), e);
|
||
}
|
||
}
|
||
|
||
@Override
|
||
public PageVO<SyncLog> pageQuery(Integer schId, String syncType, String syncTypeName,
|
||
String classMethod, Integer status, String userName,
|
||
Pageable pageable) {
|
||
Specification<SyncLog> spec = (root, query, cb) -> {
|
||
List<Predicate> ps = new ArrayList<>();
|
||
if (schId != null) ps.add(cb.equal(root.get("schId").as(Integer.class), schId));
|
||
if (StringUtils.isNotBlank(syncType)) ps.add(cb.equal(root.get("syncType").as(String.class), syncType));
|
||
if (StringUtils.isNotBlank(syncTypeName))
|
||
ps.add(cb.like(root.get("syncTypeName").as(String.class), "%" + syncTypeName + "%"));
|
||
if (StringUtils.isNotBlank(classMethod))
|
||
ps.add(cb.like(root.get("classMethod").as(String.class), "%" + classMethod + "%"));
|
||
if (status != null) ps.add(cb.equal(root.get("status").as(Integer.class), status));
|
||
if (StringUtils.isNotBlank(userName))
|
||
ps.add(cb.like(root.get("userName").as(String.class), "%" + userName.trim() + "%"));
|
||
return cb.and(ps.toArray(new Predicate[0]));
|
||
};
|
||
Pageable page = pageable;
|
||
if (!page.getSort().isSorted()) {
|
||
page = PageRequest.of(page.getPageNumber(), page.getPageSize(),
|
||
Sort.by(Sort.Direction.DESC, "createTime"));
|
||
}
|
||
Page<SyncLog> p = repository.findAll(spec, page);
|
||
return new PageVO<>(p.getNumber() + 1, p.getSize(), p.getTotalElements(), p.getContent());
|
||
}
|
||
|
||
@Override
|
||
public SyncLog getById(Long id) {
|
||
return repository.findById(id).orElse(null);
|
||
}
|
||
|
||
@Override
|
||
public String getDownloadData(Long id) {
|
||
SyncLog e = getById(id);
|
||
if (e == null) return "未找到该同步日志(id=" + id + ")";
|
||
if (StringUtils.isNotBlank(e.getJsonData())) return e.getJsonData();
|
||
if (StringUtils.isNotBlank(e.getErrorMsg())) return e.getErrorMsg();
|
||
return "该条同步日志未记录请求数据/异常信息";
|
||
}
|
||
|
||
/**
|
||
* 重试：按 classMethod + jsonData 反射重放原始同步/推送调用。
|
||
* - 仅允许失败/未判定（status != 1）的记录重试；
|
||
* - 经 Spring 代理调用，会再次走 @SyncOpLog 切面、生成新的同步日志；
|
||
* - 入参按位置映射（切面用 LinkedHashMap 按入参顺序生成 jsonData），覆盖基本类型/枚举/List/POJO；
|
||
* - 代理安全：支持 CGLIB / JDK 动态代理（见 ReflectRetryUtil）；
|
||
* - 注意：原日志已对 pwd/idCard/phone 等脱敏为 ***，用户类增改重试可能因此失败（best-effort）。
|
||
*/
|
||
@Override
|
||
public Object retry(Long id) {
|
||
SyncLog e = getById(id);
|
||
if (e == null) throw new DException("同步日志不存在(id=" + id + ")");
|
||
if (e.getStatus() != null && e.getStatus() == 1) throw new DException("该同步已成功，无需重试");
|
||
if (StringUtils.isBlank(e.getClassMethod())) throw new DException("该日志缺少调用信息(classMethod)，无法重试");
|
||
if (StringUtils.isBlank(e.getJsonData())) throw new DException("该日志缺少入参(jsonData)，无法重试");
|
||
|
||
// 记录本次重试元数据（重试次数 +1、重试时间）
|
||
e.setRetryCount((e.getRetryCount() == null ? 0 : e.getRetryCount()) + 1);
|
||
e.setLastRetryTime(LocalDateTime.now());
|
||
|
||
try {
|
||
@SuppressWarnings("unchecked")
|
||
Map<String, Object> argMap = JSON.parseObject(e.getJsonData(), LinkedHashMap.class);
|
||
if (argMap == null) throw new DException("入参解析失败，无法重试");
|
||
|
||
// 定位目标 Bean + Method（代理安全：支持 CGLIB / JDK 代理兜底扫描）
|
||
ReflectRetryUtil.Resolved resolved = ReflectRetryUtil.resolveBeanAndMethod(
|
||
applicationContext, e.getClassMethod(), argMap.size());
|
||
|
||
Method target = resolved.method;
|
||
Parameter[] params = target.getParameters();
|
||
Object[] args = new Object[params.length];
|
||
if (params.length > 0 && params[0].isNamePresent()) {
|
||
// 编译期保留参数名：按参数名匹配
|
||
for (int i = 0; i < params.length; i++) {
|
||
args[i] = ReflectRetryUtil.convert(argMap.get(params[i].getName()), params[i]);
|
||
}
|
||
} else {
|
||
// 按位置匹配（切面保证 jsonData 与入参同序）
|
||
List<Object> values = new ArrayList<>(argMap.values());
|
||
if (values.size() != params.length) {
|
||
throw new DException("入参个数不匹配，无法重试(日志=" + values.size() + ", 方法=" + params.length + ")");
|
||
}
|
||
for (int i = 0; i < params.length; i++) {
|
||
args[i] = ReflectRetryUtil.convert(values.get(i), params[i]);
|
||
}
|
||
}
|
||
|
||
Method invokeMethod = ReflectRetryUtil.getInvokeMethod(resolved.bean, target);
|
||
SyncLogRetryContext.mark(); // 标记为重试，切面生成新日志时在同步类型名追加"(重试)"
|
||
Object result = invokeMethod.invoke(resolved.bean, args);
|
||
e.setLastRetryStatus(1); // 本次重试成功
|
||
return result;
|
||
} catch (DException de) {
|
||
e.setLastRetryStatus(0); // 本次重试失败
|
||
throw de;
|
||
} catch (InvocationTargetException | IllegalAccessException ex) {
|
||
e.setLastRetryStatus(0); // 本次重试失败
|
||
throw ReflectRetryUtil.rethrow(ex);
|
||
} catch (RuntimeException re) {
|
||
e.setLastRetryStatus(0); // 本次重试失败（入参转换 / 定位等）
|
||
throw re;
|
||
} finally {
|
||
SyncLogRetryContext.clear(); // 清除重试标记，避免线程池复用泄漏
|
||
try {
|
||
repository.save(e); // 同步落库重试元数据（次数 / 上次状态 / 时间）
|
||
} catch (Exception saveEx) {
|
||
log.error("重试元数据落库失败, id={}", e.getId(), saveEx);
|
||
}
|
||
}
|
||
}
|
||
}
|