编码技巧——事务提交后执行

柔光的暖阳◎ 2023-09-29 11:42 41阅读 0赞

日常开发中，一些诸如"先读后写"、"先写A再写B"、"先写A再执行B"的场景，一般都会用到事务；这里的事务指的是本地事务，如果涉及RPC，一般我们通过异步补偿来保证最终一致性；本篇例举2个使用事务"先写A再执行B"的场景；

### 1. 订单场景 ###

**（1）处理支付回调**

这里主要是收到支付系统的结果回调后执行的逻辑，包括参数校验，业务订单校验，幂等处理，订单（成功/失败）状态更新；注意这里涉及"先查再写"的场景，外层使用了@Transaction注解，让Spring的**TransactionSynchronizationManager**来管理事务；

/**
         * 处理支付回调
         *
         * @param payResultReqDTO
         * @return
         */
        @Transactional
        @Override
        public boolean handlePayResult(PayResultReqDTO payResultReqDTO) {
            log.warn("[paySystem callback]handlePayResult [payResultReqDTO={}]", JSON.toJSONString(payResultReqDTO));
            try {
                // 基础参数校验
                boolean checkPayResultReq = checkPayResultReq(payResultReqDTO);
                if (!checkPayResultReq) {
                    return false;
                }
                // 判断支付状态
                String tradeStatus = payResultReqDTO.getTradeStatus();
                // tradeStatus不是终态,返回处理失败
                if (!OrderConstants.TRADE_STATUS_SUCCESS.equals(tradeStatus) && !OrderConstants.TRADE_STATUS_FAIL.equals(tradeStatus)) {
                    log.warn("[paySystem callback]tradeStatus is not 0000/0002! payResultReqDTO={}", JSON.toJSONString(payResultReqDTO));
                    return false;
                }
                // 查询本地订单
                MemberOrderDO orderDO = memberOrderDAO.selectByOrderNo(payResultReqDTO.getCpOrderNumber());
                if (orderDO == null) {
                    log.warn("[paySystem callback]order not found in DB! payResultReqDTO={}", JSON.toJSONString(payResultReqDTO));
                    return false;
                }
                // (已经处理过的)支付成功或失败不处理
                if (MemberOrderStatusEnum.PAID.getStatus().equals(orderDO.getStatus()) || MemberOrderStatusEnum.PAY_FAILED.getStatus().equals(orderDO.getStatus())) {
                    log.warn("[paySystem callback]memberOrder has already been handled. [payResultReqDTO={}] status={}", JSON.toJSONString(payResultReqDTO), orderDO.getStatus());
                    return true;
                }
                try {
                    if (OrderConstants.TRADE_STATUS_SUCCESS.equals(tradeStatus)) {
                        // 处理支付金额回调，若实际支付金额未取到，则取订单金额
                        String payAmount = StringUtils.isBlank(payResultReqDTO.getTradeAmount()) ? payResultReqDTO.getOrderAmount() : payResultReqDTO.getTradeAmount();
                        return handleMemberOrderAfterPaySuc(orderDO, payResultReqDTO.getOrderNumber(), payAmount, payResultReqDTO.getPayTime(), payResultReqDTO.getDiscount(), buildPayExtraInfo(payResultReqDTO.getPayExtInfo()));
                    } else if (OrderConstants.TRADE_STATUS_FAIL.equals(tradeStatus)) {
                        // 支付失败
                        return handleMemberOrderAfterPayFail(orderDO);
                    }
                } catch (Exception e) {
                    TransactionAspectSupport.currentTransactionStatus().setRollbackOnly();
                    log.error("[SERIOUS_DB][paySystem callback]handlePayResult error rollback! [payResultReqDTO={}] e:{}", JSON.toJSONString(payResultReqDTO), e);
                }
            } catch (Exception e) {
                log.error("[SERIOUS_BUSINESS][paySystem callback]handlePayResult error! [payResultReqDTO={}] e:{}", JSON.toJSONString(payResultReqDTO), e);
            }
            return false;
        }

**（2）订单更新和消息投递（消息事务）**

事务默认传递，因此订单状态更新的两个方法也被事务包裹，这里关注方法handleMemberOrderAfterPaySuc()和handleMemberOrderAfterPayFail；方法内主要包含业务订单状态的更新，以及将订单结果入库本地消息表**（消息事务）**，然后再投递出去给下游系统；

/**
         * 支付成功后执行操作
         */
        private boolean handleMemberOrderAfterPaySuc(MemberOrderDO memberOrderDO, String payOrderNo, String payAmount, String payTime, String deductAmount, PayExtraInfo payExtraInfo) {
            fillMemberOrderParam(memberOrderDO, payOrderNo, payAmount, payTime, deductAmount, payExtraInfo);
            memberOrderDAO.updatePayResult(memberOrderDO);
            Map<String, String> msgBody = buildPaySucMsgBody(memberOrderDO);
            MessageDeliverDO messageDeliverDO = buildMessageDeliverDO(memberOrderDO.getOrderNo(), MessageDeliverOrderTypeEnum.MEMBER_ORDER.getType(), msgBody);
            // 如果是代扣支付单则更新代扣单状态
            if (MemberOrderTypeEnum.WITHHOLD_ORDER.getType().equals(memberOrderDO.getOrderType())) {
                withholdDAO.updateOrderStatus(memberOrderDO.getOrderNo(), WithholdStatusEnum.PAY_SUC.getStatus());
            }
            messageDeliverDAO.insert(messageDeliverDO);
            // 优惠活动资格使用
            activityService.makeBenefitUsed(memberOrderDO.getOpenid(), memberOrderDO.getOrderNo());
            // 投递订单消息RMQ
            postProcessHandleMemberOrder(memberOrderDO);
            return true;
        }
    
        /**
         * 支付失败后执行操作
         */
        private boolean handleMemberOrderAfterPayFail(MemberOrderDO memberOrderDO) {
            memberOrderDO.setPayStatus(PayStatusEnum.PAY_FAILED.getStatus());
            memberOrderDO.setStatus(MemberOrderStatusEnum.PAY_FAILED.getStatus());
            memberOrderDAO.updatePayResult(memberOrderDO);
            // 如果是代扣支付单则更新代扣单状态
            if (MemberOrderTypeEnum.WITHHOLD_ORDER.getType().equals(memberOrderDO.getOrderType())) {
                withholdDAO.updateOrderStatus(memberOrderDO.getOrderNo(), WithholdStatusEnum.PAY_FAIL.getStatus());
                log.warn("withhold_pay_fail,[openid={} orderNo={}]", memberOrderDO.getOpenid(), memberOrderDO.getOrderNo());
            }
            // 订单状态是否为支付失败,若订单状态为支付失败，则处理微信纯签约，签约失败但支付成功的异常情况
            if (PayStatusEnum.PAY_FAILED.getStatus().equals(memberOrderDO.getPayStatus()) && StringUtils.isNotBlank(memberOrderDO.getAgreementNo())) {
                // 处理蓝卡微信纯签约逻辑
                processWeChatMiniAgreement(memberOrderDO);
            }
            // 优惠活动资格恢复
            activityService.makeBenefitAvailable(memberOrderDO.getOpenid(), memberOrderDO.getOrderNo());
            // 支付失败 不会向core发RMQ
            postProcessHandleMemberOrder(memberOrderDO);
            return true;
        }

**（3）后置处理-投递消息**

这里先入库本地消息表，然后再投递消息；注意这里使用**TransactionSynchronizationManager**这个类，并且使用了"提交后再执行"afterCommit'的；**原因是**：事务会传递到这里，如果不使用afterCommit，投递消息服务的调用（DUBBO调用消息中间件）也会被包裹在外层大事务中，所有的操作ready之后才会一起commit，而对于RPC来说，已经被执行了；

**也就是说，极端情况，消息先发出去，订单状态和本地消息表才入库，而在入库之前，下游可能已经收到了这条发出去的消息通知，这是不允许的！例如下游系统收到消息后调用消息回执接口，而这时消息还没有入库本地消息表，就会出现回执失败；**

/**
         * 会员订单后置处理
         */
        private void postProcessHandleMemberOrder(MemberOrderDO memberOrderDO) {
            if (MemberOrderStatusEnum.PAID.getStatus().equals(memberOrderDO.getStatus())) {
                // fixme 要求一定要更新订单表和消息入库 才能发消息 极端情况RPC发消息调用瞬间 core回调 但是DB事务还没有提交完成 RPC应该剥离事务
                TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronizationAdapter() {
                    @Override
                    public void afterCommit() {
                        // 支付成功发消息
                        Map<String, String> msgBody = buildPaySucMsgBody(memberOrderDO);
                        boolean sendMsgResult = notifyService.sendPayMsg(msgBody);
                        if (sendMsgResult) {
                            log.warn("send paySucMsg to core sus.[orderNo={} msg={}]", memberOrderDO.getOrderNo(), JSON.toJSONString(msgBody));
                        } else {
                            log.warn("send paySucMsg to core failed.[orderNo={} msg={}]", memberOrderDO.getOrderNo(), JSON.toJSONString(msgBody));
                        }
                        if (StringUtils.isNotBlank(memberOrderDO.getAgreementNo())) {
                            // 支付成功后，查询是否有需要投递的签约信息，用于处理微信纯签约，新用户购买，在收到支付回调之后，需要进行处理
                            MessageDeliverDO messageDeliverDO = messageDeliverDAO.queryByOrderAndType(memberOrderDO.getAgreementNo(), MessageDeliverOrderTypeEnum.AGREEMENT.getType());
                            if (messageDeliverDO != null) {
                                // 发送签约信息
                                Map<String, String> signedMsgBody = JSON.parseObject(messageDeliverDO.getMsgBody(), Map.class);
                                boolean sendSignedMsgResult = notifyService.sendSignMsg(signedMsgBody);
                                log.warn("pay_suc_send_signSucMsg_to_core_sus.[orderNo={} res={} msg={}]", memberOrderDO.getOrderNo(), sendSignedMsgResult, JSON.toJSONString(sendSignedMsgResult));
                            }
                        }
                    }
                });
            }
        }

### **2. 物料更新并通知下游** ###

开发中遇到的一个滥用事务的问题，原本的起义是，希望在活动计划物料审核通过后（落库后），将物料同步给下游（RMQ），并且调用通知服务告知相关人员；因此，在代码中使用事务@Transaction包裹业务逻辑，包括：（1）按照计划id查询计划、（2）查询后校验、（3）校验通过后更新DB中计划的状态、（4）更新成功后调用消息投递服务同步下游、（5）同时调用通知服务发送短信邮件给相关人员；**乍一看好像没有问题，"先查后写"确实需要事务包裹，并且如果通知服务调用失败，我们也可以允许DB更新回滚，运营可通过再次操作完成审核**；

但是——

遇到的问题描述下：部分RPC发生异常（消息投递服务调用成功，但是通知服务发送短信邮件调用失败），导致事务回滚，因此DB未更新，但部分RPC（消息投递服务调用成功）已经执行出去了；

原因很明显，就是我们的RPC是无法会滚的，尤其是当涉及多个RPC操作，后面的RPC失败并不会导致前面的RPC回滚（除非显示的调用RPC的回滚方法，不过一般是没有的），只有本地的DB操作回滚了；并且，抛出的RPC异常，并不能确保本地RPC调用一定是失败了，也可能是超时了，但是远程的服务已经被执行成功了；

> **结论就是**：在事务@Transaction中，尽量避免RPC操作；应当将RPC剥离出事务，如执行完本地DB事务后再去执行RPC，如果RPC失败，我们可以通过重试模板甚至是异常日志告警+手动重试的方式补偿，而RPC接口一般都是幂等的，重复的调用也不会有问题；

因此使用了下方的代码，即提交事务后再执行RPC：

/**
         * 审核计划并通知下游
         *
         * @param reqDTO
         * @return
         */
        @Transactional(rollbackFor = {Exception.class, BusinessException.class})
        @Override
        public long verify(InternalTestingPlanVerifyDTO reqDTO) {
    			/* 前置查询DB并校验... */
    
    			// 更新DB
                gameInternalTestingPlanApplyDAO.updateByPrimaryKeySelective(operateDomain);
    
                // 这里是后置的RPC操作 尽量从事务中剥离出来或使用异步线程执行
                TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronizationAdapter() {
                    @Override
                    public void afterCommit() {
                        // 审核通过时-通知下游 物料通知
                        if (InternalTestingPlanStatusEnum.PASSED.getType().equals(verifyResultDTO.getVerifyResultStatus())) {
                            final Map<String, Object> body = BeanUtils.convertToMap(applyDO);
                            // 物料通知【RPC】
                            MessageDeliverTaskReqDTO materialModifyMessage = buildBaseMessageDeliverTaskReqDTO(planId, MessageSceneTypeEnum.INTESTINGPLAN_MATERIAL.getSceneType(), body);
                            messageDeliverService.deliver(materialModifyMessage);
                            log.warn("verify_pass_send_internalTestingPlan_material_message_suc.");
                        }
                        // 审核通过/不通过时-短信邮件提醒【RPC】
                        noticeService.notifyVerifyResult(reqDTO);
                    }
                });
                return planId;
            } else {
                throw new BusinessException(ResultCodeEnum.BAD_PARAM, "invalid_planId_cannot_operate");
            }
        }

### 3. **TransactionSynchronizationManager** ###

TransactionSynchronizationManager是一个事务管理的核心类，通过TransactionSynchronizationManager我们可以管理当前线程的事务。而很多时候我们使用这个类是为了方便我们在事务结束或者开始之前实现一些自己的逻辑；如下：

@Service
    @Transactional
    @Slf4j
    public class UserService {
    
        @Autowired
        private UserRepository userRepository;
    
        @Autowired
        private OrderService orderService;
    
        public User addUser(User user) {
            log.info("任务开始！");
            userRepository.save(user);
            TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronization() {
    
                @Override
                public void beforeCommit(boolean readOnly) {
                    log.info("事务开始提交");
                }
    
                @Override
                public void afterCommit() {
                    log.info("事务提交结束");
                    orderService.addOrder();
                }
            });
            log.info("任务已经结束！");
            return user;
        }
    
    }

但是注意，由于事务默认是传递的，因此不要在afterCommit里面再去写一个事务包裹的调用，如下：

@Service
    @Transactional
    @Slf4j
    public class OrderService {
    
        @Autowired
        private OrderRepository orderRepository;
    
        @Transactional
        public Order addOrder() {
            log.info("开始插入order数据");
            Order order = new Order();
            order.setMoney(100D);
            orderRepository.save(order);
            log.info("插入order数据结束");
            return order;
        }
    
    }

因为原事务会传递到afterCommit里面的addOrder方法上，因此addOrder等不到原事务的提交，永远无法执行；可以打印一下原事务名和addOrder执行时的事务名验证：

String currentTransactionName = TransactionSynchronizationManager.getCurrentTransactionName();
    log.info("currentTransactionName is {}",currentTransactionName);

// 控制台输出：
    2022-01-05 23:28:28.669  INFO 12908 --- [main] dai.samples.jpa2.service.UserService     : currentTransactionName is dai.samples.jpa2.service.UserService.addUser
    2022-01-05 23:28:28.673  INFO 12908 --- [main] dai.samples.jpa2.service.OrderService    : currentTransactionName is dai.samples.jpa2.service.UserService.addUser

可见，两个方法使用了相同的事务，但是需要注意的是addOrder方法是在afterCommit事务提交之后执行的，此时会导致addOrder中的JPA数据保存最终无法提交。所以我们需要使addOrder进入一个新的事务中，如**在@Transactional注解中propagation参数用来控制事务的传播**；

**@Transactional注解的propagation参数默认被设置为Propagation.REQUIRED传递**，其逻辑是，如果当前没有事务，就新建一个事务，如果已经存在一个事务中，加入到这个事务中。而上面的业务中我们并不希望其加入已有的事务中，所以单介绍上面的逻辑，假如希望JPA的数据保存到数据库中，需要在事务注解修改为@Transactional(propagation = Propagation.REQUIRES\_NEW)参数；

然而在很多时候我们希望新加入的方法能够被同一个事务所管理，而使用Propagation.REQUIRES\_NEW会导致当前操作脱离上一级事务的控制；所以**在使用@Transactional(propagation = Propagation.REQUIRES\_NEW)的时候一定要慎重**，并且严格控制其被滥用；

参考：

[TransactionSynchronizationManager.registerSynchronization使用中事务传播产生的问题][TransactionSynchronizationManager.registerSynchronization]

[Spring 事务事件控制 解决业务异步操作解耦 TransactionSynchronizationManager][Spring _ _ TransactionSynchronizationManager]

[TransactionSynchronizationManager.registerSynchronization]: https://blog.csdn.net/qq330983778/article/details/112255441
[Spring _ _ TransactionSynchronizationManager]: https://www.jyoryo.com/archives/155.html