admin/smartDriveEE
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

同步用户并构建上级链条

Browse Source
This commit is contained in:
cst61
2026-05-19 17:56:46 +08:00
parent 893af21ce2
commit 1313269710
10 changed files with 338 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files

12
src/main/java/com/rj/controller/LbDepartmentUserController.java
View File

@@ -108,4 +108,16 @@ public class LbDepartmentUserController {
}
return ResponseEntity.badRequest().body(result);
}
@PostMapping("/syncFromLbUser")
@Operation(summary = "从lb_user同步到部门用户并构建上级链条")
public ResponseEntity<Map<String, Object>> syncFromLbUser(
@Parameter(description = "租户id", required = true) @RequestParam String tenantId) {
Map<String, Object> result = lbDepartmentUserService.syncFromLbUser(tenantId);
Boolean success = (Boolean) result.get("success");
if (success != null && success) {
return ResponseEntity.ok(result);
}
return ResponseEntity.badRequest().body(result);
}
}

4
src/main/java/com/rj/entity/LbDepartmentUser.java
View File

@@ -55,6 +55,10 @@ public class LbDepartmentUser implements Serializable {
@Schema(description = "从事的行业")
private String industry;
@TableField("org_chain")
@Schema(description = "上级链条（从直接上级到根上级，逗号分隔的用户ID）")
private String orgChain;
@TableField("join_date")
@Schema(description = "加入日期（不含时分秒）")
private LocalDate joinDate;

6
src/main/java/com/rj/service/ILbDailyUserTradeReportService.java
View File

@@ -6,6 +6,7 @@ import com.rj.entity.LbDailyUserTrade;
import com.rj.entity.LbDailyUserTradeReport;
import java.time.LocalDate;
import java.util.List;
import java.util.Map;
/**
@@ -21,6 +22,11 @@ public interface ILbDailyUserTradeReportService extends IService<LbDailyUserTrad
Map<String, Object> calculateReportSumByDateAndTenant(LocalDate reportDate, String tenantId);
/**
* 替换指定租户、报表日期的明细报表（report_detail）：先删后插，须在单事务内执行。
*/
void replaceDetailReportsForDate(String tenantId, LocalDate reportDate, List<LbDailyUserTradeReport> reports);
/**
* 按日期范围与租户统计每人四类金额（昨日买入、当日卖出、当日买入、差额）：本人汇总与递归下级汇总。
* 用户范围由 lb_daily_user_trade_report 在条件内出现过的 user_id 圈定；上下级由 lb_department_user 确定。

5
src/main/java/com/rj/service/ILbDepartmentUserService.java
View File

@@ -30,4 +30,9 @@ public interface ILbDepartmentUserService extends IService<LbDepartmentUser> {
String tenantId);
Map<String, Object> syncFromDailyUserTrade(LocalDate startDate, LocalDate endDate);
/**
* 从 lb_user 同步到 lb_department_user，并按 pid 构建 org_chain 上级链条。
*/
Map<String, Object> syncFromLbUser(String tenantId);
}

45
src/main/java/com/rj/service/impl/LbDailyUserTradeReportServiceImpl.java
View File

@@ -10,10 +10,15 @@ import com.rj.mapper.LbDailyUserTradeMapper;
import com.rj.mapper.LbDailyUserTradeReportMapper;
import com.rj.service.ILbDailyUserTradeReportService;
import com.rj.service.ILbDepartmentUserService;
import com.rj.service.support.LbDailyUserTradeReportWriteSupport;
import com.rj.tenant.TenantContextHolder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.transaction.PlatformTransactionManager;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.transaction.support.TransactionTemplate;
import java.math.BigDecimal;
import java.time.LocalDate;
@@ -46,6 +51,8 @@ public class LbDailyUserTradeReportServiceImpl extends ServiceImpl<LbDailyUserTr
private ILbDepartmentUserService lbDepartmentUserService;
@Autowired
private LbDailyUserTradeMapper lbDailyUserTradeMapper;
@Autowired
private PlatformTransactionManager transactionManager;
@Override
public Page<LbDailyUserTrade> pageDailyUserTradeByUserIdAndDateRange(
@@ -82,8 +89,46 @@ public class LbDailyUserTradeReportServiceImpl extends ServiceImpl<LbDailyUserTr
return lbDailyUserTradeMapper.selectPage(new Page<>(current, size), queryWrapper);
}
@Override
@Transactional(rollbackFor = Exception.class)
public void replaceDetailReportsForDate(String tenantId, LocalDate reportDate, List<LbDailyUserTradeReport> reports) {
String normalizedTenantId = tenantId.trim();
LocalDateTime reportDateTime = reportDate.atStartOfDay();
LambdaQueryWrapper<LbDailyUserTradeReport> deleteWrapper = new LambdaQueryWrapper<>();
deleteWrapper.eq(LbDailyUserTradeReport::getTenantId, normalizedTenantId)
.eq(LbDailyUserTradeReport::getReportDate, reportDateTime)
.eq(LbDailyUserTradeReport::getDataType, DATA_TYPE_REPORT_DETAIL);
remove(deleteWrapper);
if (reports != null && !reports.isEmpty()) {
if (!saveBatch(reports)) {
throw new IllegalStateException("报表明细批量插入失败");
}
}
}
@Override
public Map<String, Object> calculateReportSumByDateAndTenant(LocalDate reportDate, String tenantId) {
String normalizedTenantId = tenantId.trim();
TransactionTemplate tx = new TransactionTemplate(transactionManager);
return LbDailyUserTradeReportWriteSupport.runExclusiveReturning(
normalizedTenantId,
reportDate,
() -> LbDailyUserTradeReportWriteSupport.runWithDeadlockRetry(() -> tx.execute(status -> {
String previousTenantId = TenantContextHolder.getTenantId();
TenantContextHolder.setTenantId(normalizedTenantId);
try {
return doCalculateReportSumByDateAndTenant(reportDate, normalizedTenantId);
} finally {
if (previousTenantId != null) {
TenantContextHolder.setTenantId(previousTenantId);
} else {
TenantContextHolder.clear();
}
}
})));
}
private Map<String, Object> doCalculateReportSumByDateAndTenant(LocalDate reportDate, String tenantId) {
Map<String, Object> result = new HashMap<>();
LocalDateTime start = reportDate.atStartOfDay();
LocalDateTime end = reportDate.plusDays(1).atStartOfDay();

28
src/main/java/com/rj/service/impl/LbDailyUserTradeServiceImpl.java
View File

@@ -8,6 +8,7 @@ import com.rj.entity.LbDailyUserTradeReport;
import com.rj.mapper.LbDailyUserTradeMapper;
import com.rj.service.ILbDailyUserTradeService;
import com.rj.service.ILbDailyUserTradeReportService;
import com.rj.service.support.LbDailyUserTradeReportWriteSupport;
import org.apache.poi.poifs.filesystem.FileMagic;
import org.apache.poi.ss.usermodel.Cell;
import org.apache.poi.ss.usermodel.CellType;
@@ -161,17 +162,26 @@ public class LbDailyUserTradeServiceImpl extends ServiceImpl<LbDailyUserTradeMap
report.setUpdatedAt(now);
reportList.add(report);
}
// 保存前先删除同租户同报表日期的历史报表，避免重复插入
LambdaQueryWrapper<LbDailyUserTradeReport> reportDeleteWrapper = new LambdaQueryWrapper<>();
reportDeleteWrapper.eq(LbDailyUserTradeReport::getTenantId, normalizedTenantId)
.eq(LbDailyUserTradeReport::getReportDate, reportDate.atStartOfDay());
lbDailyUserTradeReportService.remove(reportDeleteWrapper);
boolean saved = lbDailyUserTradeReportService.saveBatch(reportList);
if (!saved) {
String previousTenantId = TenantContextHolder.getTenantId();
try {
LbDailyUserTradeReportWriteSupport.runExclusive(normalizedTenantId, reportDate, () ->
LbDailyUserTradeReportWriteSupport.runWithDeadlockRetry(() -> {
TenantContextHolder.setTenantId(normalizedTenantId);
lbDailyUserTradeReportService.replaceDetailReportsForDate(
normalizedTenantId, reportDate, reportList);
}));
} catch (RuntimeException ex) {
log.error("calculateDailyTradeReport write failed, tenantId={}, reportDate={}",
normalizedTenantId, reportDate, ex);
result.put("success", false);
result.put("message", "报表插入失败");
result.put("message", "报表写入失败：" + ex.getMessage());
return result;
} finally {
if (previousTenantId != null) {
TenantContextHolder.setTenantId(previousTenantId);
} else {
TenantContextHolder.clear();
}
}
result.put("success", true);

156
src/main/java/com/rj/service/impl/LbDepartmentUserServiceImpl.java
View File

@@ -6,12 +6,15 @@ import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.rj.entity.LbDailyUserTrade;
import com.rj.entity.LbDepartmentUser;
import com.rj.entity.LbPurchaseApply;
import com.rj.entity.LbUser;
import com.rj.mapper.LbDailyUserTradeMapper;
import com.rj.mapper.LbDepartmentUserMapper;
import com.rj.mapper.LbPurchaseApplyMapper;
import com.rj.mapper.LbUserMapper;
import com.rj.service.ILbDepartmentUserService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import java.util.ArrayList;
import java.util.HashSet;
@@ -33,6 +36,8 @@ public class LbDepartmentUserServiceImpl extends ServiceImpl<LbDepartmentUserMap
private LbDailyUserTradeMapper lbDailyUserTradeMapper;
@Autowired
private LbPurchaseApplyMapper lbPurchaseApplyMapper;
@Autowired
private LbUserMapper lbUserMapper;
@Override
public Map<String, Object> add(LbDepartmentUser entity) {
@@ -257,13 +262,12 @@ public class LbDepartmentUserServiceImpl extends ServiceImpl<LbDepartmentUserMap
}
List<LbDailyUserTrade> tradeList = lbDailyUserTradeMapper.selectList(
new LambdaQueryWrapper<LbDailyUserTrade>()
.ge(LbDailyUserTrade::getReportDate, startDate)
.le(LbDailyUserTrade::getReportDate, endDate)
.ge(LbDailyUserTrade::getDailyBuyAmt, 1)
.isNotNull(LbDailyUserTrade::getNickname)
.orderByDesc(LbDailyUserTrade::getUpdatedAt)
.orderByDesc(LbDailyUserTrade::getCreatedAt)
);
if (tradeList == null || tradeList.isEmpty()) {
result.put("success", true);
@@ -396,4 +400,152 @@ public class LbDepartmentUserServiceImpl extends ServiceImpl<LbDepartmentUserMap
String userPart = applyUser == null ? "" : applyUser.trim();
return tenantPart + "#" + userPart;
}
@Override
@Transactional(rollbackFor = Exception.class)
public Map<String, Object> syncFromLbUser(String tenantId) {
Map<String, Object> result = new HashMap<>();
try {
if (tenantId == null || tenantId.trim().isEmpty()) {
result.put("success", false);
result.put("message", "租户id不能为空");
return result;
}
String tenantIdTrim = tenantId.trim();
List<LbUser> lbUsers = lbUserMapper.selectList(
new LambdaQueryWrapper<LbUser>().eq(LbUser::getTenantId, tenantIdTrim));
if (lbUsers == null || lbUsers.isEmpty()) {
result.put("success", true);
result.put("message", "无可同步用户数据");
result.put("insertedCount", 0);
result.put("updatedCount", 0);
result.put("sourceCount", 0);
result.put("tenantId", tenantIdTrim);
return result;
}
Map<Long, LbUser> userById = new HashMap<>();
for (LbUser user : lbUsers) {
if (user.getId() != null) {
userById.put(user.getId(), user);
}
}
List<LbDepartmentUser> existedUsers = this.list(
new LambdaQueryWrapper<LbDepartmentUser>().eq(LbDepartmentUser::getTenantId, tenantIdTrim));
Map<String, LbDepartmentUser> existingByUserId = new HashMap<>();
if (existedUsers != null) {
for (LbDepartmentUser deptUser : existedUsers) {
if (deptUser.getUserId() != null && !deptUser.getUserId().trim().isEmpty()) {
existingByUserId.put(deptUser.getUserId().trim(), deptUser);
}
}
}
List<LbDepartmentUser> toInsert = new ArrayList<>();
List<LbDepartmentUser> toUpdate = new ArrayList<>();
LocalDateTime now = LocalDateTime.now();
for (LbUser lbUser : lbUsers) {
if (lbUser.getId() == null) {
continue;
}
String userId = String.valueOf(lbUser.getId());
LbDepartmentUser entity = existingByUserId.get(userId);
boolean isInsert = false;
if (entity == null) {
entity = new LbDepartmentUser();
entity.setId(UUID.randomUUID().toString());
entity.setCreateTime(now);
isInsert = true;
}
entity.setTenantId(tenantIdTrim);
entity.setUserId(userId);
entity.setName(resolveDisplayName(lbUser));
entity.setPhone(lbUser.getMobile());
if (lbUser.getJoinTime() != null) {
entity.setJoinDate(lbUser.getJoinTime().toLocalDate());
}
Long pid = lbUser.getPid();
if (pid != null && pid > 0) {
entity.setParentId(String.valueOf(pid));
LbUser parent = userById.get(pid);
if (parent != null) {
entity.setParentName(resolveDisplayName(parent));
} else if (lbUser.getPname() != null && !lbUser.getPname().trim().isEmpty()) {
entity.setParentName(lbUser.getPname().trim());
}
} else {
entity.setParentId(null);
entity.setParentName(null);
}
entity.setOrgChain(buildOrgChain(lbUser.getId(), userById));
entity.setUpdateTime(now);
if (isInsert) {
toInsert.add(entity);
} else {
toUpdate.add(entity);
}
}
if (!toInsert.isEmpty()) {
this.saveBatch(toInsert);
}
if (!toUpdate.isEmpty()) {
this.updateBatchById(toUpdate);
}
result.put("success", true);
result.put("message", "同步完成");
result.put("insertedCount", toInsert.size());
result.put("updatedCount", toUpdate.size());
result.put("sourceCount", lbUsers.size());
result.put("tenantId", tenantIdTrim);
return result;
} catch (Exception e) {
result.put("success", false);
result.put("message", "同步异常：" + e.getMessage());
return result;
}
}
/**
* 沿 lb_user.pid 向上追溯，构建从直接上级到根上级的用户ID链条。
*/
private String buildOrgChain(Long userId, Map<Long, LbUser> userById) {
LbUser current = userById.get(userId);
if (current == null) {
return null;
}
List<String> chain = new ArrayList<>();
Set<Long> visited = new HashSet<>();
Long pid = current.getPid();
while (pid != null && pid > 0) {
if (!visited.add(pid)) {
break;
}
chain.add(String.valueOf(pid));
LbUser parent = userById.get(pid);
if (parent == null) {
break;
}
pid = parent.getPid();
}
return chain.isEmpty() ? null : String.join(",", chain);
}
private String resolveDisplayName(LbUser user) {
if (user.getNickname() != null && !user.getNickname().trim().isEmpty()) {
return user.getNickname().trim();
}
if (user.getUsername() != null && !user.getUsername().trim().isEmpty()) {
return user.getUsername().trim();
}
return user.getId() == null ? null : String.valueOf(user.getId());
}
}

6
src/main/java/com/rj/service/impl/LbUserServiceImpl.java
View File

@@ -285,9 +285,9 @@ public class LbUserServiceImpl extends ServiceImpl<LbUserMapper, LbUser> impleme
if (entity.getTodaySellTotal() == null) {
entity.setTodaySellTotal(BigDecimal.ZERO);
}
if (entity.getPoor() == null) {
entity.setPoor(0);
}
// poor 为 TINYINT；远程可能返回超出 [-128,127] 的值，按贫困标识归一为 0/1
Integer poor = entity.getPoor();
entity.setPoor(poor == null || poor == 0 ? 0 : 1);
}
private static LocalDateTime parseDateTime(String text) {

88
src/main/java/com/rj/service/support/LbDailyUserTradeReportWriteSupport.java Normal file
View File

@@ -0,0 +1,88 @@
package com.rj.service.support;
import java.sql.SQLException;
import java.time.LocalDate;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.locks.ReentrantLock;
import java.util.function.Supplier;
/**
* 同一租户、同一报表日期的 lb_daily_user_trade_report 写操作串行化，降低并发 DELETE/INSERT 死锁概率。
*/
public final class LbDailyUserTradeReportWriteSupport {
private static final ConcurrentHashMap<String, ReentrantLock> LOCKS = new ConcurrentHashMap<>();
private static final int MAX_DEADLOCK_RETRIES = 3;
private LbDailyUserTradeReportWriteSupport() {
}
public static void runExclusive(String tenantId, LocalDate reportDate, Runnable action) {
runExclusiveReturning(tenantId, reportDate, () -> {
action.run();
return null;
});
}
public static <T> T runExclusiveReturning(String tenantId, LocalDate reportDate, Supplier<T> action) {
String key = lockKey(tenantId, reportDate);
ReentrantLock lock = LOCKS.computeIfAbsent(key, ignored -> new ReentrantLock());
lock.lock();
try {
return runWithDeadlockRetry(action);
} finally {
lock.unlock();
}
}
public static void runWithDeadlockRetry(Runnable action) {
runWithDeadlockRetry(() -> {
action.run();
return null;
});
}
public static <T> T runWithDeadlockRetry(Supplier<T> action) {
RuntimeException last = null;
for (int attempt = 1; attempt <= MAX_DEADLOCK_RETRIES; attempt++) {
try {
return action.get();
} catch (RuntimeException ex) {
last = ex;
if (attempt >= MAX_DEADLOCK_RETRIES || !isDeadlock(ex)) {
throw ex;
}
sleepBriefly(attempt);
}
}
throw last;
}
public static boolean isDeadlock(Throwable throwable) {
for (Throwable current = throwable; current != null; current = current.getCause()) {
if (current instanceof SQLException sqlException) {
if (sqlException.getErrorCode() == 1213 || "40001".equals(sqlException.getSQLState())) {
return true;
}
}
String message = current.getMessage();
if (message != null && message.contains("Deadlock found")) {
return true;
}
}
return false;
}
private static String lockKey(String tenantId, LocalDate reportDate) {
return tenantId.trim() + "|" + reportDate;
}
private static void sleepBriefly(int attempt) {
try {
Thread.sleep(50L * attempt);
} catch (InterruptedException interrupted) {
Thread.currentThread().interrupt();
throw new IllegalStateException("报表写入被中断", interrupted);
}
}
}