新单跟进136

yd-csf-api/src/main/java/com/yd/csf/api/controller/ApiCommissionController.java

@@ -12,24 +12,22 @@ import com.yd.common.result.Result;
 import com.yd.csf.api.dto.CommissionExcelDTO;
 import com.yd.csf.service.common.ErrorCode;
 import com.yd.csf.service.dto.*;
-import com.yd.csf.service.enums.CommissionStatusEnum;
 import com.yd.csf.service.model.Commission;
 import com.yd.csf.service.model.CommissionCompareRecord;
 import com.yd.csf.service.model.CommissionEditRecord;
-import com.yd.csf.service.model.CommissionExpected;
-import com.yd.csf.service.service.CommissionCompareRecordService;
-import com.yd.csf.service.service.CommissionEditRecordService;
-import com.yd.csf.service.service.CommissionExpectedService;
-import com.yd.csf.service.service.CommissionService;
+import com.yd.csf.service.service.*;
 import com.yd.csf.service.vo.CommissionStatisticsVO;
 import io.swagger.v3.oas.annotations.Operation;
 import io.swagger.v3.oas.annotations.tags.Tag;
 import org.apache.commons.collections4.CollectionUtils;
 import org.apache.commons.lang3.StringUtils;
+import org.redisson.api.RLock;
+import org.redisson.api.RedissonClient;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
-import org.springframework.scheduling.annotation.Async;
 import org.springframework.transaction.annotation.Transactional;
+import org.springframework.transaction.support.TransactionSynchronization;
+import org.springframework.transaction.support.TransactionSynchronizationManager;
 import org.springframework.web.bind.annotation.*;
 import org.springframework.web.multipart.MultipartFile;
 
@@ -37,10 +35,8 @@ import javax.annotation.Resource;
 import javax.servlet.http.HttpServletRequest;
 import java.io.IOException;
 import java.math.BigDecimal;
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
+import java.util.*;
+import java.util.concurrent.TimeUnit;
 import java.util.stream.Collectors;
 
 /**
@@ -58,16 +54,16 @@ public class ApiCommissionController {
 
     @Resource
     private CommissionService commissionService;
-
     @Resource
     private CommissionExpectedService commissionExpectedService;
-
     @Resource
     private CommissionEditRecordService commissionEditRecordService;
-
     @Resource
     private CommissionCompareRecordService commissionCompareRecordService;
-
+    @Resource
+    private RedissonClient redissonClient;
+    @Resource
+    private CommissionAsyncService commissionAsyncService;
 
 
     @PostMapping("/upload/excel")
@@ -93,7 +89,23 @@ public class ApiCommissionController {
         String loginUserId = currentLoginUser.getId().toString();
 
         // 数据处理
-        processData(dataList, loginUserId);
+        List<Commission> entities = processData(dataList, loginUserId);
+
+        // 批量比对
+        TransactionSynchronizationManager.registerSynchronization(
+                new TransactionSynchronization() {
+                    @Override
+                    public void afterCommit() {
+                        try {
+                            // 重新查询最新的数据，获取已提交的数据
+                            commissionAsyncService.commissionCompareBatch(entities);
+                        } catch (Exception e) {
+                            // 比对失败不影响主事务，记录日志即可
+                                log.error("批量导入, 比对操作执行失败, error: {}", e.getMessage());
+                        }
+                    }
+                }
+        );
 
         return Result.success(true);
     }
@@ -130,7 +142,7 @@ public class ApiCommissionController {
     /**
      * 1.处理导入的数据 2.比对预计来佣 3.保存到数据库
      */
-    private void processData(List<CommissionExcelDTO> dataList, String loginUserId) {
+    private List<Commission> processData(List<CommissionExcelDTO> dataList, String loginUserId) {
         List<Commission> entities = new ArrayList<>();
         for (CommissionExcelDTO data : dataList) {
             // 数据验证
@@ -141,52 +153,7 @@ public class ApiCommissionController {
         // 保存来佣数据
         commissionService.saveBatch(entities);
 
-        // todo 6. 异步触发比对任务（事务提交后执行）
-//        TransactionSynchronizationManager.registerSynchronization(
-//                new TransactionSynchronization() {
-//                    @Override
-//                    public void afterCommit() {
-//                        // 主事务提交后再执行异步任务
-//                        commissionComareBatch(entities);
-//                    }
-//                }
-//        );
-
-    }
-
-    /**
-     * 异步执行来佣对比（有独立事务）
-     */
-    @Async("commonAsyncExecutor")
-    @Transactional(rollbackFor = Exception.class)
-    public void commissionComareBatch(List<Commission> entities) {
-        // 根据导入的来佣获取保单号集合
-        List<String> policyNoList = entities.stream()
-                .map(Commission::getPolicyNo)
-                .collect(Collectors.toList());
-        // 根据保单号查询预计来佣
-        List<CommissionExpected> expectedList = commissionExpectedService.lambdaQuery()
-                .in(CommissionExpected::getPolicyNo, policyNoList)
-                .list();
-
-        for (Commission commission : entities) {
-            // 查询预计来佣
-            CommissionExpected commissionExpected;
-            try {
-                commissionExpected = commissionService.queryByCommission(commission);
-                // 对比预计来佣
-                commissionService.compareWithExpected(commission.getPremium(), commission, commissionExpected);
-                commissionExpectedService.updateById(commissionExpected);
-            } catch (Exception e) {
-                // 预计来佣不存在，来佣状态设置为 比对失败
-                commission.setStatus(CommissionStatusEnum.COMPARE_FAIL.getItemValue());
-                commission.setRemark(e.getMessage());
-            }
-        }
-        // 保存来佣数据
-        commissionService.saveBatch(entities);
-        // 更新预计来佣状态
-        commissionExpectedService.updateBatchById(expectedList);
+        return entities;
     }
 
     /**
@@ -288,7 +255,30 @@ public class ApiCommissionController {
         if (commissionUpdateRequest == null || commissionUpdateRequest.getCommissionBizId() == null) {
             return Result.fail(ErrorCode.PARAMS_ERROR.getCode(), ErrorCode.PARAMS_ERROR.getMessage());
         }
-        return Result.success(commissionService.updateCommission(commissionUpdateRequest));
+        // 校验入参
+        commissionService.validateCommissionUpdateRequest(commissionUpdateRequest);
+
+        String lockKey = "COMMISSION_LOCK:" + commissionUpdateRequest.getCommissionBizId();
+        RLock lock = redissonClient.getLock(lockKey);
+        try {
+            // 竞争锁
+            boolean res = lock.tryLock(3, 15, TimeUnit.SECONDS);
+            if (!res) {
+                log.warn("获取分布式锁失败，commissionBizId: {}", commissionUpdateRequest.getCommissionBizId());
+                throw new BusinessException("系统繁忙，请稍后重试");
+            }
+            // 更新来佣记录
+            Result.success(commissionService.updateCommission(commissionUpdateRequest));
+        } catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
+            log.error("更新来佣记录时，获取锁被中断", e);
+            return Result.fail(ErrorCode.OPERATION_ERROR.getCode(), "更新来佣记录时，获取锁被中断");
+        } finally {
+            if (lock != null && lock.isLocked() && lock.isHeldByCurrentThread()) {
+                lock.unlock();
+            }
+        }
+        return Result.success(true);
     }
 
     /**

yd-csf-service/src/main/java/com/yd/csf/service/service/CommissionService.java

 package com.yd.csf.service.service;
 
-import com.baomidou.mybatisplus.core.conditions.Wrapper;
 import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
 import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
+import com.yd.auth.core.dto.AuthUserDto;
 import com.yd.csf.service.dto.*;
 import com.yd.csf.service.model.Commission;
 import com.baomidou.mybatisplus.extension.service.IService;
+import com.yd.csf.service.model.CommissionCompareRecord;
 import com.yd.csf.service.model.CommissionExpected;
 import com.yd.csf.service.vo.CommissionStatisticsVO;
 import com.yd.csf.service.vo.CommissionVO;
 
 import java.util.List;
-import java.util.Map;
 
 /**
 * @author Zhang Jianan
@@ -40,5 +40,15 @@ public interface CommissionService extends IService<Commission> {
 
     CommissionStatisticsVO getCommissionStatistics(List<Long> commissionIds);
 
-    CommissionExpected queryByCommission(Commission commission);
+    CommissionExpected queryByCommission(String policyNo, String commissionName, Integer commissionPeriod, String currency, String premium);
+
+    void calculatePaidAmountByCommissionList(Commission commission, CommissionExpected one);
+
+    void validateCommissionUpdateRequest(CommissionUpdateRequest commissionUpdateRequest);
+
+    void saveUpdateCommissionRecord(Commission commission, CommissionUpdateRequest commissionUpdateRequest, AuthUserDto currentLoginUser);
+
+    void saveCompareRecord(Commission commission, CommissionExpected commissionExpected, AuthUserDto currentLoginUser);
+
+    CommissionCompareRecord getNewCompareRecord(Commission existingCommission, CommissionExpected commissionExpected, AuthUserDto currentLoginUser);
 }