feat 订单取消/修复/取消/同步等操作添加分布式锁, 防止出现重复操作

_doc/Task.md

@@ -28,19 +28,17 @@
 - 2024-01-05:
   - [x] 支付同步日志记录, 无论同步成功还是失败, 以及修复成功还是失败, 都需要记录日志
   - [x] 超时自动取消功能联调, 先用spring定时任务实现, 通过支付同步实现
+  - [x] 支付同步和修复时, 对一些模糊状态进行处理, 例如支付宝返回的订单未查到
   - [x] 待支付支付单定时同步状态, 先用spring定时任务实现, 通过支付同步实现
   - [x] 退款功能联调
 - 2024-01-06:
+  - [x] 订单取消/修复/取消/同步等操作添加分布式锁, 防止出现重复操作
   - [ ] 增加支付修复记录
   - [ ] 退款状态同步逻辑
   - [ ] 退款回调的处理
   - [ ] 支付状态同步处理考虑退款情况
   - [ ] 增加消息通知机制(通知客户端)
 - **任务池**
-  - 支付同步时，有些状态无法区分处理, 导致无法修复
-    - 下面内容同: 支付网关/本地 
-    - 支付宝: 订单未找到/支付关闭 支付网关/支付超时 支付网关/支付成功  
-  - 订单取消/修复/取消/同步添加分布式锁, 防止操作订单时出现重复操作
   - 支付状态同步处理退款情况
   - 支付配置支持数据库配置和配置文件配置
   - 增加回调机制(通知客户端)

daxpay-single/daxpay-single-admin/pom.xml

@@ -25,7 +25,7 @@
             <version>${bootx-platform.version}</version>
         </dependency>
 
-        <!--若使用redisTemplate作为分布式锁底层，则需要引入-->
+        <!--分布式锁, 使用redisTemplate作为底层-->
         <dependency>
             <groupId>com.baomidou</groupId>
             <artifactId>lock4j-redis-template-spring-boot-starter</artifactId>

daxpay-single/daxpay-single-service/pom.xml

@@ -56,11 +56,6 @@
             <artifactId>common-super-query</artifactId>
         </dependency>
 
-        <dependency>
-            <groupId>org.redisson</groupId>
-            <artifactId>redisson-spring-boot-starter</artifactId>
-        </dependency>
-
         <!-- 支付核心包-->
         <dependency>
             <groupId>cn.bootx.platform</groupId>
@@ -104,5 +99,10 @@
             <artifactId>junit-jupiter</artifactId>
             <scope>test</scope>
         </dependency>
+        <dependency>
+            <groupId>com.baomidou</groupId>
+            <artifactId>lock4j-redis-template-spring-boot-starter</artifactId>
+            <version>${lock4j.version}</version>
+        </dependency>
     </dependencies>
 </project>

daxpay-single/daxpay-single-service/src/main/java/cn/bootx/platform/daxpay/service/core/payment/close/service/PayCloseService.java

@@ -1,5 +1,6 @@
 package cn.bootx.platform.daxpay.service.core.payment.close.service;
 
+import cn.bootx.platform.common.core.exception.RepetitiveOperationException;
 import cn.bootx.platform.daxpay.code.PayStatusEnum;
 import cn.bootx.platform.daxpay.exception.pay.PayFailureException;
 import cn.bootx.platform.daxpay.exception.pay.PayUnsupportedMethodException;
@@ -13,6 +14,8 @@ import cn.bootx.platform.daxpay.service.core.record.pay.entity.PayOrder;
 import cn.bootx.platform.daxpay.service.core.record.pay.service.PayOrderService;
 import cn.bootx.platform.daxpay.service.func.AbsPayCloseStrategy;
 import cn.hutool.core.collection.CollectionUtil;
+import com.baomidou.lock.LockInfo;
+import com.baomidou.lock.LockTemplate;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.stereotype.Service;
@@ -34,6 +37,8 @@ public class PayCloseService {
     private final PayOrderService payOrderService;
     private final PayCloseRecordService payCloseRecordService;
 
+    private final LockTemplate lockTemplate;
+
     /**
      * 关闭支付
      */
@@ -48,7 +53,15 @@ public class PayCloseService {
             payOrder = payOrderService.findByBusinessNo(param.getBusinessNo())
                     .orElseThrow(() -> new PayFailureException("未查询到支付订单"));
         }
-        this.close(payOrder);
+        LockInfo lock = lockTemplate.lock("payment:close:" + payOrder.getId());
+        if (Objects.isNull(lock)){
+            throw new RepetitiveOperationException("支付订单已在关闭中，请勿重复发起");
+        }
+        try {
+            this.close(payOrder);
+        } finally {
+            lockTemplate.releaseLock(lock);
+        }
     }
 
     /**

daxpay-single/daxpay-single-service/src/main/java/cn/bootx/platform/daxpay/service/core/payment/pay/service/PayService.java

@@ -1,21 +1,24 @@
 package cn.bootx.platform.daxpay.service.core.payment.pay.service;
 
+import cn.bootx.platform.common.core.exception.RepetitiveOperationException;
 import cn.bootx.platform.daxpay.code.PayStatusEnum;
-import cn.bootx.platform.daxpay.service.core.payment.pay.factory.PayStrategyFactory;
-import cn.bootx.platform.daxpay.service.core.record.pay.builder.PaymentBuilder;
-import cn.bootx.platform.daxpay.service.core.record.pay.entity.PayOrder;
-import cn.bootx.platform.daxpay.service.core.record.pay.service.PayOrderService;
 import cn.bootx.platform.daxpay.exception.pay.PayUnsupportedMethodException;
-import cn.bootx.platform.daxpay.service.func.AbsPayStrategy;
-import cn.bootx.platform.daxpay.service.func.PayStrategyConsumer;
 import cn.bootx.platform.daxpay.param.pay.PayParam;
 import cn.bootx.platform.daxpay.param.pay.PayWayParam;
 import cn.bootx.platform.daxpay.param.pay.SimplePayParam;
 import cn.bootx.platform.daxpay.result.pay.PayResult;
+import cn.bootx.platform.daxpay.service.core.payment.pay.factory.PayStrategyFactory;
+import cn.bootx.platform.daxpay.service.core.record.pay.builder.PaymentBuilder;
+import cn.bootx.platform.daxpay.service.core.record.pay.entity.PayOrder;
+import cn.bootx.platform.daxpay.service.core.record.pay.service.PayOrderService;
+import cn.bootx.platform.daxpay.service.func.AbsPayStrategy;
+import cn.bootx.platform.daxpay.service.func.PayStrategyConsumer;
 import cn.bootx.platform.daxpay.util.PayUtil;
 import cn.hutool.core.bean.BeanUtil;
 import cn.hutool.core.bean.copier.CopyOptions;
 import cn.hutool.core.collection.CollectionUtil;
+import com.baomidou.lock.LockInfo;
+import com.baomidou.lock.LockTemplate;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.stereotype.Service;
@@ -41,6 +44,8 @@ public class PayService {
 
     private final PayAssistService payAssistService;
 
+    private final LockTemplate lockTemplate;
+
     /**
      * 支付下单接口(同步/异步/组合支付)
      * 1. 同步支付：都只会在第一次执行中就完成支付，例如钱包、储值卡都是调用完就进行了扣减，完成了支付记录
@@ -55,18 +60,28 @@ public class PayService {
         // 异步支付方式检查
         PayUtil.validationAsyncPay(payParam);
 
-        // 获取并校验支付订单状态, 如果超时, 触发支付单同步和修复动作
+        String businessNo = payParam.getBusinessNo();
-        PayOrder payOrder = payAssistService.getOrderAndCheck(payParam.getBusinessNo());
+        // 加锁
+        LockInfo lock = lockTemplate.lock("payment:pay:" + businessNo);
+        if (Objects.isNull(lock)){
+            throw new RepetitiveOperationException("正在支付中，请勿重复支付");
+        }
+        try {
+            // 获取并校验支付订单状态, 如果超时, 触发支付单同步和修复动作
+            PayOrder payOrder = payAssistService.getOrderAndCheck(payParam.getBusinessNo());
 
-        // 初始化上下文
+            // 初始化上下文
-        payAssistService.initPayContext(payOrder, payParam);
+            payAssistService.initPayContext(payOrder, payParam);
 
-        // 异步支付且非第一次支付
+            // 异步支付且非第一次支付
-        if (Objects.nonNull(payOrder) && payOrder.isAsyncPay()) {
+            if (Objects.nonNull(payOrder) && payOrder.isAsyncPay()) {
-            return this.paySyncNotFirst(payParam, payOrder);
+                return this.paySyncNotFirst(payParam, payOrder);
-        } else {
+            } else {
-            // 第一次发起支付或同步支付
+                // 第一次发起支付或同步支付
-            return this.firstPay(payParam, payOrder);
+                return this.firstPay(payParam, payOrder);
+            }
+        } finally {
+            lockTemplate.releaseLock(lock);
         }
     }
 

daxpay-single/daxpay-single-service/src/main/java/cn/bootx/platform/daxpay/service/core/payment/refund/service/PayRefundService.java

@@ -1,5 +1,6 @@
 package cn.bootx.platform.daxpay.service.core.payment.refund.service;
 
+import cn.bootx.platform.common.core.exception.RepetitiveOperationException;
 import cn.bootx.platform.common.core.util.ValidationUtil;
 import cn.bootx.platform.daxpay.code.PayStatusEnum;
 import cn.bootx.platform.daxpay.service.core.payment.refund.factory.PayRefundStrategyFactory;
@@ -13,6 +14,8 @@ import cn.bootx.platform.daxpay.param.pay.SimpleRefundParam;
 import cn.bootx.platform.daxpay.result.pay.RefundResult;
 import cn.hutool.core.bean.BeanUtil;
 import cn.hutool.core.collection.CollectionUtil;
+import com.baomidou.lock.LockInfo;
+import com.baomidou.lock.LockTemplate;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.stereotype.Service;
@@ -20,6 +23,7 @@ import org.springframework.transaction.annotation.Transactional;
 
 import java.util.Collections;
 import java.util.List;
+import java.util.Objects;
 import java.util.stream.Collectors;
 
 /**
@@ -35,6 +39,9 @@ public class PayRefundService {
     private final PayRefundAssistService payRefundAssistService;;
 
     private final PayOrderService payOrderService;
+
+    private final LockTemplate lockTemplate;
+
     /**
      * 支付退款
      */
@@ -69,19 +76,30 @@ public class PayRefundService {
         PayOrder payOrder = payRefundAssistService.getPayOrderAndCheckByRefundParam(param, simple);
         // 参数校验
         ValidationUtil.validateParam(param);
-        // 退款上下文初始化
+
-        payRefundAssistService.initRefundContext(param);
+        // 加锁
-        // 是否全部退款
+        LockInfo lock = lockTemplate.lock("payment:refund:" + payOrder.getId());
-        if (param.isRefundAll()){
+        if (Objects.isNull(lock)){
-            // 全部退款根据支付订单的退款信息构造退款参数
+            throw new RepetitiveOperationException("退款处理中，请勿重复操作");
-            List<RefundChannelParam> channelParams = payOrder.getRefundableInfos()
+        }
-                    .stream()
+
-                    .map(o -> new RefundChannelParam().setChannel(o.getChannel())
+        try {
-                            .setAmount(o.getAmount()))
+            // 退款上下文初始化
-                    .collect(Collectors.toList());
+            payRefundAssistService.initRefundContext(param);
-            param.setRefundChannels(channelParams);
+            // 是否全部退款
+            if (param.isRefundAll()){
+                // 全部退款根据支付订单的退款信息构造退款参数
+                List<RefundChannelParam> channelParams = payOrder.getRefundableInfos()
+                        .stream()
+                        .map(o -> new RefundChannelParam().setChannel(o.getChannel())
+                                .setAmount(o.getAmount()))
+                        .collect(Collectors.toList());
+                param.setRefundChannels(channelParams);
+            }
+            return this.refundByChannel(param,payOrder);
+        } finally {
+            lockTemplate.releaseLock(lock);
         }
-        return this.refundByChannel(param,payOrder);
     }
 
     /**

daxpay-single/daxpay-single-service/src/main/java/cn/bootx/platform/daxpay/service/core/payment/sync/service/PaySyncService.java

@@ -1,6 +1,7 @@
 package cn.bootx.platform.daxpay.service.core.payment.sync.service;
 
 import cn.bootx.platform.common.core.exception.BizException;
+import cn.bootx.platform.common.core.exception.RepetitiveOperationException;
 import cn.bootx.platform.common.core.util.LocalDateTimeUtil;
 import cn.bootx.platform.daxpay.code.PayStatusEnum;
 import cn.bootx.platform.daxpay.code.PaySyncStatusEnum;
@@ -19,6 +20,8 @@ import cn.bootx.platform.daxpay.service.core.record.pay.service.PayOrderService;
 import cn.bootx.platform.daxpay.service.core.record.sync.entity.PaySyncRecord;
 import cn.bootx.platform.daxpay.service.core.record.sync.service.PaySyncRecordService;
 import cn.bootx.platform.daxpay.service.func.AbsPaySyncStrategy;
+import com.baomidou.lock.LockInfo;
+import com.baomidou.lock.LockTemplate;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.stereotype.Service;
@@ -48,6 +51,8 @@ public class PaySyncService {
 
     private final PayRepairService repairService;
 
+    private final LockTemplate lockTemplate;
+
     /**
      * 支付同步, 开启一个新的事务, 不受外部抛出异常的影响
      */
@@ -75,46 +80,56 @@ public class PaySyncService {
      * 2. 如果状态不一致, 调用修复逻辑进行修复
      */
     @Transactional(propagation = Propagation.REQUIRES_NEW)
-    public PaySyncResult syncPayOrder(PayOrder order) {
+    public PaySyncResult syncPayOrder(PayOrder payOrder) {
-        // 获取同步策略类
+        // 加锁
-        AbsPaySyncStrategy syncPayStrategy = PaySyncStrategyFactory.create(order.getAsyncChannel());
+        LockInfo lock = lockTemplate.lock("payment:refund:" + payOrder.getId());
-        syncPayStrategy.initPayParam(order);
+        if (Objects.isNull(lock)){
-        // 记录支付单同步前后的状态
+            throw new RepetitiveOperationException("支付同步处理中，请勿重复操作");
-        String oldStatus = order.getStatus();
-        String repairStatus = null;
-
-        // 执行同步操作, 获取支付网关同步的结果
-        GatewaySyncResult syncResult = syncPayStrategy.doSyncStatus();
-        // 判断是否同步成功
-        if (Objects.equals(syncResult.getSyncStatus(), PaySyncStatusEnum.FAIL)){
-            // 同步失败, 返回失败响应, 同时记录失败的日志
-            this.saveRecord(order, syncResult, true, oldStatus, null, syncResult.getErrorMsg());
-            return new PaySyncResult().setErrorMsg(syncResult.getErrorMsg());
         }
 
-        // 判断网关状态是否和支付单一致, 同时更新网关同步状态
-        boolean statusSync = this.checkAndAdjustSyncStatus(syncResult,order);
         try {
-            // 状态不一致，执行支付单修复逻辑
+            // 获取同步策略类
-            if (!statusSync){
+            AbsPaySyncStrategy syncPayStrategy = PaySyncStrategyFactory.create(payOrder.getAsyncChannel());
-                this.resultHandler(syncResult, order);
+            syncPayStrategy.initPayParam(payOrder);
-                repairStatus = order.getStatus();
+            // 记录支付单同步前后的状态
-            }
+            String oldStatus = payOrder.getStatus();
-        } catch (PayFailureException e) {
+            String repairStatus = null;
-            // 同步失败, 返回失败响应, 同时记录失败的日志
-            syncResult.setSyncStatus(PaySyncStatusEnum.FAIL);
-            this.saveRecord(order, syncResult, false, oldStatus, null, e.getMessage());
-            return new PaySyncResult().setErrorMsg(e.getMessage());
-        }
 
-        // 同步成功记录日志
+            // 执行同步操作, 获取支付网关同步的结果
-        this.saveRecord( order, syncResult, !statusSync, oldStatus, repairStatus, null);
+            GatewaySyncResult syncResult = syncPayStrategy.doSyncStatus();
-        return new PaySyncResult()
+            // 判断是否同步成功
-                .setGatewayStatus(syncResult.getSyncStatus().getCode())
+            if (Objects.equals(syncResult.getSyncStatus(), PaySyncStatusEnum.FAIL)){
-                .setSuccess(true)
+                // 同步失败, 返回失败响应, 同时记录失败的日志
-                .setRepair(!statusSync)
+                this.saveRecord(payOrder, syncResult, true, oldStatus, null, syncResult.getErrorMsg());
-                .setOldStatus(oldStatus)
+                return new PaySyncResult().setErrorMsg(syncResult.getErrorMsg());
-                .setRepairStatus(repairStatus);
+            }
+
+            // 判断网关状态是否和支付单一致, 同时更新网关同步状态
+            boolean statusSync = this.checkAndAdjustSyncStatus(syncResult,payOrder);
+            try {
+                // 状态不一致，执行支付单修复逻辑
+                if (!statusSync){
+                    this.resultHandler(syncResult, payOrder);
+                    repairStatus = payOrder.getStatus();
+                }
+            } catch (PayFailureException e) {
+                // 同步失败, 返回失败响应, 同时记录失败的日志
+                syncResult.setSyncStatus(PaySyncStatusEnum.FAIL);
+                this.saveRecord(payOrder, syncResult, false, oldStatus, null, e.getMessage());
+                return new PaySyncResult().setErrorMsg(e.getMessage());
+            }
+
+            // 同步成功记录日志
+            this.saveRecord( payOrder, syncResult, !statusSync, oldStatus, repairStatus, null);
+            return new PaySyncResult()
+                    .setGatewayStatus(syncResult.getSyncStatus().getCode())
+                    .setSuccess(true)
+                    .setRepair(!statusSync)
+                    .setOldStatus(oldStatus)
+                    .setRepairStatus(repairStatus);
+        } finally {
+            lockTemplate.releaseLock(lock);
+        }
     }
 
     /**

daxpay-single/daxpay-single-service/src/main/java/cn/bootx/platform/daxpay/service/core/payment/sync/task/PayOrderSyncTaskService.java

@@ -1,39 +0,0 @@
-package cn.bootx.platform.daxpay.service.core.payment.sync.task;
-
-import cn.bootx.platform.daxpay.service.core.payment.sync.service.PaySyncService;
-import cn.bootx.platform.daxpay.param.pay.PaySyncParam;
-import lombok.RequiredArgsConstructor;
-import lombok.extern.slf4j.Slf4j;
-import org.springframework.stereotype.Service;
-
-/**
- * 对未过期的支付中订单进行状态同步
- * @author xxm
- * @since 2024/1/1
- */
-@Slf4j
-@Service
-@RequiredArgsConstructor
-public class PayOrderSyncTaskService {
-    private final cn.bootx.platform.daxpay.service.core.timeout.dao.PayExpiredTimeRepository PayExpiredTimeRepository;
-
-    private final PaySyncService paySyncService;
-
-    /**
-     * 同步支付订单任务
-     */
-    public void syncTask() {
-        log.info("开始同步支付订单");
-        // 1. 从超时订单列表中获取到未超时的订单号
-        for (String s : PayExpiredTimeRepository.getNormalKeysBy30Day()) {
-            try {
-                Long paymentId = Long.parseLong(s);
-                PaySyncParam paySyncParam = new PaySyncParam();
-                paySyncParam.setPaymentId(paymentId);
-                paySyncService.sync(paySyncParam);
-            } catch (Exception e) {
-                log.error("同步支付订单异常", e);
-            }
-        }
-    }
-}

daxpay-single/daxpay-single-service/src/main/java/cn/bootx/platform/daxpay/service/core/timeout/task/PayExpiredTimeTask.java

@@ -1,13 +1,18 @@
 package cn.bootx.platform.daxpay.service.core.timeout.task;
 
+import cn.bootx.platform.common.core.exception.RepetitiveOperationException;
 import cn.bootx.platform.daxpay.param.pay.PaySyncParam;
 import cn.bootx.platform.daxpay.service.core.payment.sync.service.PaySyncService;
 import cn.bootx.platform.daxpay.service.core.timeout.dao.PayExpiredTimeRepository;
+import com.baomidou.lock.LockInfo;
+import com.baomidou.lock.LockTemplate;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
+import org.springframework.scheduling.annotation.Scheduled;
 import org.springframework.stereotype.Service;
 
 import java.time.LocalDateTime;
+import java.util.Objects;
 import java.util.Set;
 
 /**
@@ -20,15 +25,20 @@ import java.util.Set;
 @RequiredArgsConstructor
 public class PayExpiredTimeTask {
     private final PayExpiredTimeRepository repository;
+
     private final PaySyncService paySyncService;
 
+    private final LockTemplate lockTemplate;
 
-//    @Scheduled(cron = "*/5 * * * * ?")
+    @Scheduled(cron = "*/5 * * * * ?")
     public void task(){
         log.info("执行超时取消任务....");
         Set<String> expiredKeys = repository.getExpiredKeys(LocalDateTime.now());
         for (String expiredKey : expiredKeys) {
-            log.info("key:{}", expiredKey);
+            LockInfo lock = lockTemplate.lock("payment:expired:" + expiredKey,10000,0);
+            if (Objects.isNull(lock)){
+                throw new RepetitiveOperationException("支付同步处理中，请勿重复操作");
+            }
             try {
                 // 执行同步操作, 网关同步时会对支付的进行状态的处理
                 Long paymentId = Long.parseLong(expiredKey);
@@ -37,6 +47,8 @@ public class PayExpiredTimeTask {
                 paySyncService.sync(paySyncParam);
             } catch (Exception e) {
                 log.error("超时取消任务 异常", e);
+            } finally {
+                lockTemplate.releaseLock(lock);
             }
 
         }

daxpay-single/daxpay-single-service/src/main/java/cn/bootx/platform/daxpay/service/core/timeout/task/PayWaitOrderSyncTask.java

@@ -1,12 +1,16 @@
 package cn.bootx.platform.daxpay.service.core.timeout.task;
 
+import cn.bootx.platform.common.core.exception.RepetitiveOperationException;
 import cn.bootx.platform.daxpay.param.pay.PaySyncParam;
 import cn.bootx.platform.daxpay.service.core.payment.sync.service.PaySyncService;
 import cn.bootx.platform.daxpay.service.core.timeout.dao.PayExpiredTimeRepository;
+import com.baomidou.lock.LockInfo;
+import com.baomidou.lock.LockTemplate;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.stereotype.Service;
 
+import java.util.Objects;
 import java.util.Set;
 
 /**
@@ -22,12 +26,17 @@ public class PayWaitOrderSyncTask {
 
     private final PaySyncService paySyncService;
 
+    private final LockTemplate lockTemplate;
+
     public void task(){
         log.info("开始同步支付订单");
         // 从超时订单列表中获取到未超时的订单号
         Set<String> keys = repository.getNormalKeysBy30Day();
         for (String key : keys) {
-            log.info("key:{}", key);
+            LockInfo lock = lockTemplate.lock("payment:sync:" + key,10000,0);
+            if (Objects.isNull(lock)){
+                throw new RepetitiveOperationException("支付同步处理中，请勿重复操作");
+            }
             try {
                 Long paymentId = Long.parseLong(key);
                 PaySyncParam paySyncParam = new PaySyncParam();
@@ -36,6 +45,8 @@ public class PayWaitOrderSyncTask {
                 paySyncService.sync(paySyncParam);
             } catch (Exception e) {
                 log.error("同步支付订单异常", e);
+            } finally {
+                lockTemplate.releaseLock(lock);
             }
         }
 

pom.xml

@@ -43,7 +43,7 @@
         <table-modify.version>1.5.4</table-modify.version>
         <wxjava.version>4.5.2.B</wxjava.version>
         <rocketmq.version>2.2.3</rocketmq.version>
-        <lock4j.version>2.2.4</lock4j.version>
+        <lock4j.version>2.2.5</lock4j.version>
     </properties>
     <!-- 项目依赖版本管理 -->
     <dependencyManagement>