订单初版—4.取消订单链路中的技术问题说明文档

大纲
1.超时自动取消订单功能的业务背景
2.超时自动取消订单和支付的并发问题
3.分布式锁解决超时订单取消和支付的并发
4.定时任务解决延时消费的消息的丢失问题
5.超时自动取消订单代码流程
6.RocketMQ延迟消息实现原理
7.取消订单务场景和接口被调用的情况
8.取消订单的业务链路
9.为什么拦截履约要同步而释放资产是异步
10.支付回调分别互斥预支付和取消订单
11.支付回调和取消订单互斥下的两种场景分析
12.拦截履约的具体业务流程
13.拦截履约和取消订单的Seata事务原理分析
14.取消订单全链路Seata回滚原理与并发分析
15.支付退款时的双异步设计原因(提升性能 + 解耦)
16.释放资产消息的高扩展性设计(多路发送消息)
17.取消订单链路中数据库事务与MQ消息不一致问题
18.释放资产多路MQ故障重试与幂等方案
19.双异步支付退款不一致问题分析
 
1.超时自动取消订单功能的业务背景
用户提交订单，订单系统生成订单后，由于种种原因，用户并没有立即点击去支付或者完成支付。此时默认该订单会在生成30分钟后，自动进行支付检查。也就是如果生成订单超过30分钟还没进行支付，那么就会自动取消订单。

超时自动取消订单的技术方案：RocketMQ延迟消息 + Redisson分布式锁 + XXL-JOB分布式任务调度。
 
2.超时自动取消订单和支付的并发问题
当用户提交的订单在30分钟后才发起支付时，此时就有可能出现取消订单和支付出现并发的问题。

 
3.分布式锁解决超时订单取消和支付的并发
在如下位置加同一把分布式锁即可解决超时订单取消和支付的并发问题：
一.订单系统处理预支付请求时添加分布式锁
二.订单系统处理支付回调请求时添加分布式锁
三.订单系统消费延时30m的订单消息进行超时检查时添加分布式锁
 
当处理预支付和消费延时订单消息进行超时检查时产生并发：如果处理预支付先获取锁，则超时检查会被阻塞，之后发现已支付不取消。如果超时检查先获取锁，则预支付被阻塞，之后预支付失败。
 
当处理支付回调和消费延时订单消息进行超时检查时产生并发：如果支付回调先获取锁，则超时检查会被阻塞，之后发现已支付不取消。如果超时检查先获取锁，则支付回调被阻塞，之后支付回调获取锁，发现订单已被超时检查取消，于是需要进行退款处理通知用户。

 
4.定时任务解决延时消费的消息的丢失问题
如果发送到MQ的延时30m消费的消息丢失，此时会通过定时任务来处理。该定时任务会对MySQL进行扫描，把创建时间超30m + 没支付的订单找出，然后再对这些超时订单进行取消。定时任务一般使用分布式调度框架XXL-JOB来实现。

以上便是超时自动取消订单的技术方案：RocketMQ延迟消息 + Redisson分布式锁 + XXL-JOB分布式任务调度。
 
5.超时自动取消订单代码流程
(1)生成订单时发送延时30m才能被消费的消息
(2)订单超时检查监听器
(3)对订单进行正式取消的业务逻辑
(4)定时任务检查创建时间超30m但没取消的订单
 
(1)生成订单时发送延时30m才能被消费的消息

1. @Service
2. public class OrderServiceImpl implements OrderService {
3.     ...
4.     //提交订单/生成订单接口
5.     //@param createOrderRequest 提交订单请求入参
6.     @GlobalTransactional(rollbackFor = Exception.class)
7.     @Override
8.     public CreateOrderDTO createOrder(CreateOrderRequest createOrderRequest) {
9.         //1.入参检查
10.         checkCreateOrderRequestParam(createOrderRequest);
11.         //2.风控检查
12.         checkRisk(createOrderRequest);
13.         //3.获取商品信息
14.         List<ProductSkuDTO> productSkuList = listProductSkus(createOrderRequest);
15.         //4.计算订单价格
16.         CalculateOrderAmountDTO calculateOrderAmountDTO = calculateOrderAmount(createOrderRequest, productSkuList);
17.         //5.验证订单实付金额
18.         checkRealPayAmount(createOrderRequest, calculateOrderAmountDTO);
19.         //6.锁定优惠券
20.         lockUserCoupon(createOrderRequest);
21.         //7.锁定商品库存
22.         lockProductStock(createOrderRequest);
23.         //8.生成订单到数据库
24.         addNewOrder(createOrderRequest, productSkuList, calculateOrderAmountDTO);
25.         //9.发送订单延迟消息用于支付超时自动关单
26.         sendPayOrderTimeoutDelayMessage(createOrderRequest);
27.         //返回订单信息
28.         CreateOrderDTO createOrderDTO = new CreateOrderDTO();
29.         createOrderDTO.setOrderId(createOrderRequest.getOrderId());
30.         return createOrderDTO;
31.     }
33.     //发送支付订单超时延迟消息，用于支付超时自动关单
34.     private void sendPayOrderTimeoutDelayMessage(CreateOrderRequest createOrderRequest) {
35.         PayOrderTimeoutDelayMessage message = new PayOrderTimeoutDelayMessage();
36.         message.setOrderId(createOrderRequest.getOrderId());
37.         message.setBusinessIdentifier(createOrderRequest.getBusinessIdentifier());
38.         message.setCancelType(OrderCancelTypeEnum.TIMEOUT_CANCELED.getCode());
39.         message.setUserId(createOrderRequest.getUserId());
40.         message.setOrderType(createOrderRequest.getOrderType());
41.         message.setOrderStatus(OrderStatusEnum.CREATED.getCode());
42.         String msgJson = JsonUtil.object2Json(message);
43.         defaultProducer.sendMessage(
44.             RocketMqConstant.PAY_ORDER_TIMEOUT_DELAY_TOPIC,
45.             msgJson,
46.             RocketDelayedLevel.DELAYED_30m,
47.             "支付订单超时延迟消息"
48.         );
49.     }
50.     ...
51. }
53. @Component
54. public class DefaultProducer {
55.     private final DefaultMQProducer producer;
57.     @Autowired
58.     public DefaultProducer(RocketMQProperties rocketMQProperties) {
59.         producer = new DefaultMQProducer(RocketMqConstant.ORDER_DEFAULT_PRODUCER_GROUP);
60.         producer.setNamesrvAddr(rocketMQProperties.getNameServer());
61.         start();
62.     }
64.     //对象在使用之前必须要调用一次，只能初始化一次
65.     public void start() {
66.         try {
67.             this.producer.start();
68.         } catch (MQClientException e) {
69.             log.error("producer start error", e);
70.         }
71.     }
72.     ...
74.     //发送消息
75.     public void sendMessage(String topic, String message, Integer delayTimeLevel, String type) {
76.         Message msg = new Message(topic, message.getBytes(StandardCharsets.UTF_8));
77.         try {
78.             if (delayTimeLevel > 0) {
79.                 msg.setDelayTimeLevel(delayTimeLevel);
80.             }
81.             SendResult send = producer.send(msg);
82.             if (SendStatus.SEND_OK == send.getSendStatus()) {
83.                 log.info("发送MQ消息成功, type:{}, message:{}", type, message);
84.             } else {
85.                 throw new OrderBizException(send.getSendStatus().toString());
86.             }
87.         } catch (Exception e) {
88.             log.error("发送MQ消息失败：", e);
89.             throw new OrderBizException(OrderErrorCodeEnum.SEND_MQ_FAILED);
90.         }
91.     }
92.     ...
93. }

复制代码

(2)订单超时检查监听器

1. @Configuration
2. public class ConsumerConfig {
3.     @Autowired
4.     private RocketMQProperties rocketMQProperties;
5.     ...
7.     //支付订单超时延迟消息消费者
8.     @Bean("payOrderTimeoutConsumer")
9.     public DefaultMQPushConsumer payOrderTimeoutConsumer(PayOrderTimeoutListener payOrderTimeoutListener) throws MQClientException {
10.         DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(PAY_ORDER_TIMEOUT_DELAY_CONSUMER_GROUP);
11.         consumer.setNamesrvAddr(rocketMQProperties.getNameServer());
12.         consumer.subscribe(PAY_ORDER_TIMEOUT_DELAY_TOPIC, "*");
13.         consumer.registerMessageListener(payOrderTimeoutListener);
14.         consumer.start();
15.         return consumer;
16.     }
17.     ...
18. }
20. //监听支付订单超时延迟消息
21. @Component
22. public class PayOrderTimeoutListener implements MessageListenerConcurrently {
23.     @Autowired
24.     private OrderAfterSaleService orderAfterSaleService;
26.     @Autowired
27.     private OrderInfoDAO orderInfoDAO;
29.     @Override
30.     public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
31.         try {
32.             for (MessageExt messageExt : list) {
33.                 String message = new String(messageExt.getBody());
34.                 PayOrderTimeoutDelayMessage payOrderTimeoutDelayMessage = JSON.parseObject(message, PayOrderTimeoutDelayMessage.class);
35.                 //消费延迟消息，执行关单逻辑
36.                 CancelOrderRequest cancelOrderRequest = new CancelOrderRequest();
37.                 cancelOrderRequest.setOrderId(payOrderTimeoutDelayMessage.getOrderId());
38.                 cancelOrderRequest.setBusinessIdentifier(payOrderTimeoutDelayMessage.getBusinessIdentifier());
39.                 cancelOrderRequest.setCancelType(payOrderTimeoutDelayMessage.getOrderType());
40.                 cancelOrderRequest.setUserId(payOrderTimeoutDelayMessage.getUserId());
41.                 cancelOrderRequest.setOrderType(payOrderTimeoutDelayMessage.getOrderType());
42.                 cancelOrderRequest.setOrderStatus(payOrderTimeoutDelayMessage.getOrderStatus());
43.                 //查询当前数据库的订单实时状态
44.                 OrderInfoDO orderInfoDO = orderInfoDAO.getByOrderId(payOrderTimeoutDelayMessage.getOrderId());
45.                 Integer orderStatusDatabase = orderInfoDO.getOrderStatus();
46.                 if (!OrderStatusEnum.CREATED.getCode().equals(orderStatusDatabase)) {
47.                     //订单实时状态不等于已创建
48.                     return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
49.                 }
50.                 //当前时间 小于 订单实际支付截止时间
51.                 if (new Date().before(orderInfoDO.getExpireTime())) {
52.                     return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
53.                 }
54.                 orderAfterSaleService.cancelOrder(cancelOrderRequest);
55.                 log.info("关闭订单，orderId:{}", cancelOrderRequest.getOrderId());
56.             }
57.             return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
58.         } catch (Exception e) {
59.             log.error("consumer error", e);
60.             return ConsumeConcurrentlyStatus.RECONSUME_LATER;
61.         }
62.     }
63. }

复制代码

(3)对订单进行正式取消的业务逻辑

1. @Service
2. public class OrderAfterSaleServiceImpl implements OrderAfterSaleService {
3.     ...
4.     //取消订单/超时未支付取消
5.     @Override
6.     public JsonResult<Boolean> cancelOrder(CancelOrderRequest cancelOrderRequest) {
7.         //入参检查
8.         checkCancelOrderRequestParam(cancelOrderRequest);
9.         //分布式锁，这里的锁和预支付及支付回调使用的分布式锁一样
10.         String orderId = cancelOrderRequest.getOrderId();
11.         String key = RedisLockKeyConstants.ORDER_PAY_KEY + orderId;
12.         try {
13.             boolean lock = redisLock.lock(key);
14.             if (!lock) {
15.                 throw new OrderBizException(OrderErrorCodeEnum.CANCEL_ORDER_REPEAT);
16.             }
17.             //执行取消订单
18.             executeCancelOrder(cancelOrderRequest, orderId);
19.             return JsonResult.buildSuccess(true);
20.         } catch (Exception e) {
21.             throw new OrderBizException(e.getMessage());
22.         } finally {
23.             redisLock.unlock(key);
24.         }
25.     }
27.     @Override
28.     @GlobalTransactional(rollbackFor = Exception.class)
29.     public void executeCancelOrder(CancelOrderRequest cancelOrderRequest, String orderId) {
30.         //1.组装数据
31.         CancelOrderAssembleRequest cancelOrderAssembleRequest = buildAssembleRequest(orderId, cancelOrderRequest);
32.         //幂等校验：防止多个线程同时操作取消同一笔订单
33.         if (OrderStatusEnum.CANCELED.getCode().equals(cancelOrderAssembleRequest.getOrderInfoDTO().getOrderStatus())) {
34.             return;
35.         }
37.         //2.检验订单支付状态
38.         checkOrderPayStatus(cancelOrderAssembleRequest);
40.         //3.更新订单状态和记录订单操作日志
41.         updateOrderStatusAndSaveOperationLog(cancelOrderAssembleRequest);
42.         //超时未支付的订单不用继续再往下执行取消履约和释放资产
43.         if (OrderStatusEnum.PAID.getCode() > cancelOrderAssembleRequest.getOrderInfoDTO().getOrderStatus()) {
44.             return;
45.         }
47.         //4.履约取消
48.         cancelFulfill(cancelOrderAssembleRequest);
50.         //5.发送释放资产消息到MQ
51.         defaultProducer.sendMessage(RocketMqConstant.RELEASE_ASSETS_TOPIC, JSONObject.toJSONString(cancelOrderAssembleRequest), "释放资产");
52.     }
54.     //更新订单状态和记录订单操作日志
55.     private void updateOrderStatusAndSaveOperationLog(CancelOrderAssembleRequest cancelOrderAssembleRequest) {
56.         //更新订单表
57.         OrderInfoDO orderInfoDO = cancelOrderAssembleRequest.getOrderInfoDTO().clone(OrderInfoDO.class);
58.         orderInfoDO.setCancelType(cancelOrderAssembleRequest.getCancelType().toString());
59.         orderInfoDO.setOrderStatus(OrderStatusEnum.CANCELED.getCode());
60.         orderInfoDO.setCancelTime(new Date());
61.         orderInfoDAO.updateOrderInfo(orderInfoDO);
62.         log.info("更新订单信息OrderInfo状态: orderId:{},status:{}", orderInfoDO.getOrderId(), orderInfoDO.getOrderStatus());
64.         //新增订单操作操作日志表
65.         Integer cancelType = Integer.valueOf(orderInfoDO.getCancelType());
66.         String orderId = orderInfoDO.getOrderId();
67.         OrderOperateLogDO orderOperateLogDO = new OrderOperateLogDO();
68.         orderOperateLogDO.setOrderId(orderId);
69.         orderOperateLogDO.setPreStatus(cancelOrderAssembleRequest.getOrderInfoDTO().getOrderStatus());
70.         orderOperateLogDO.setCurrentStatus(OrderStatusEnum.CANCELED.getCode());
71.         orderOperateLogDO.setOperateType(OrderOperateTypeEnum.AUTO_CANCEL_ORDER.getCode());
72.         if (OrderCancelTypeEnum.USER_CANCELED.getCode().equals(cancelType)) {
73.             orderOperateLogDO.setOperateType(OrderOperateTypeEnum.MANUAL_CANCEL_ORDER.getCode());
74.             orderOperateLogDO.setRemark(OrderOperateTypeEnum.MANUAL_CANCEL_ORDER.getMsg()
75.                 + orderOperateLogDO.getPreStatus() + "-" + orderOperateLogDO.getCurrentStatus());
76.         }
77.         if (OrderCancelTypeEnum.TIMEOUT_CANCELED.getCode().equals(cancelType)) {
78.             orderOperateLogDO.setOperateType(OrderOperateTypeEnum.AUTO_CANCEL_ORDER.getCode());
79.             orderOperateLogDO.setRemark(OrderOperateTypeEnum.AUTO_CANCEL_ORDER.getMsg()
80.                 + orderOperateLogDO.getPreStatus() + "-" + orderOperateLogDO.getCurrentStatus());
81.         }
82.         orderOperateLogDAO.save(orderOperateLogDO);
83.         log.info("新增订单操作日志OrderOperateLog状态,orderId:{}, PreStatus:{},CurrentStatus:{}", orderInfoDO.getOrderId(),
84.             orderOperateLogDO.getPreStatus(), orderOperateLogDO.getCurrentStatus());
85.     }
86.     ...
87. }

复制代码

(4)定时任务检查创建时间超30m但没取消的订单

1. //自动取消超时订单任务
2. @Component
3. public class AutoCancelExpiredOrderTask {
4.     @Autowired
5.     private OrderInfoDAO orderInfoDAO;
6.     @Autowired
7.     private OrderAfterSaleService orderAfterSaleService;
8.     @Autowired
9.     private OrderProperties orderProperties;
10.     @Autowired
11.     private RedisLock redisLock;
13.     //执行任务逻辑
14.     @Scheduled(fixedRate = 30 * 60 * 1000)
15.     public void execute() {
16.         //扫描当前时间 - 订单超时时间 -> 前的一小段时间范围(时间范围用配置中心配置)
17.         //比如当前时间11:40，订单超时时间是30分钟，扫描11:09:00 -> 11:10:00这一分钟的未支付订单，
18.         //缺点：有一个订单超过了30 + 1 = 31分钟，都没有被处理(取消)，这笔订单就永久待支付
19.         for (OrderInfoDO order : orderInfoDAO.listAllUnPaid()) {
20.             if (new Date().getTime() - order.getExpireTime().getTime() >= orderProperties.getExpireTime()) {
21.                 //分布式锁
22.                 String orderId = order.getOrderId();
23.                 String key = RedisLockKeyConstants.ORDER_PAY_KEY + orderId;
24.                 try {
25.                     boolean lock = redisLock.lock(key);
26.                     if (!lock) {
27.                         throw new OrderBizException(OrderErrorCodeEnum.CANCEL_ORDER_REPEAT);
28.                     }
29.                     //超过30min未支付
30.                     CancelOrderRequest request = new CancelOrderRequest();
31.                     request.setOrderId(order.getOrderId());
32.                     request.setUserId(order.getUserId());
33.                     request.setBusinessIdentifier(order.getBusinessIdentifier());
34.                     request.setOrderType(order.getOrderType());
35.                     request.setCancelType(OrderCancelTypeEnum.TIMEOUT_CANCELED.getCode());
36.                     request.setOrderStatus(order.getOrderStatus());
37.                     try {
38.                         orderAfterSaleService.cancelOrder(request);
39.                     } catch (Exception e) {
40.                         log.error("AutoCancelExpiredOrderTask execute error:", e);
41.                     }
42.                 } catch (Exception e) {
43.                     throw new OrderBizException(e.getMessage());
44.                 } finally {
45.                     redisLock.unlock(key);
46.                 }
47.             }
48.         }
49.     }
50. }

复制代码

 
6.RocketMQ延迟消息实现原理
发送需要延迟消费的消息到RocketMQ的某Topic时，RocketMQ会将该消息进行改写，投递到重试Topic，而非指定的某Topic。RocketMQ将该消息转发到重试Topic对应的ConsumerQueue后，RocketMQ内部会有一个内部的延迟调度组件ScheduleService，该组件便会消费重试Topic对应的ConsumerQueue里的消息，并判断里面的消息是否到达延迟时间。如果到达延迟时间，则会重新改写消息Topic，投递到开始指定的某Topic。

 
7.取消订单的业务场景和接口被调用的情况
(1)发生取消订单的三种场景
(2)调用取消订单接口的三种情况
 
(1)发生取消订单的三种场景
场景一：用户生成订单后还没支付，就取消订单。取消订单可能是用户手动取消，也可能是超时检查自动取消，此时只需要释放锁定的优惠券和库存即可。
 
场景二：用户生成订单后完成支付，还没拣货出库，用户就取消订单。此时不仅需要释放锁定的优惠券和库存，还需取消履约以及退款。
 
场景三：用户生成订单后完成支付，已拣货出库，开始物流配送。此时用户想要取消订单，只能等收到货以后，发起售后退货申请。

(2)调用取消订单接口的三种情况
情况一：⼿动取消，订单出库状态前都可取消
情况二：正向⽣成订单后，会发送延迟30分钟消费的订单消息到MQ。订单系统在30分钟后消费到该MQ消息时，若发现仍未⽀付则取消该订单
情况三：定时扫描，超过30分钟未⽀付才取消

 
8.取消订单的业务链路和代码流程
(1)实现概述
(2)对取消订单的处理
(3)对取消履约+更新订单状态+新增订单日志的处理
(4)对释放资产的处理
(5)对退款的退款前处理
(6)对退款的实际退款处理
 
(1)实现概述
一.实现流程图

二.技术要点
要点一：更新订单状态、取消履约、发送释放资产是⼀个分布式事务，需要将Seata的AT模式替换成RocketMQ的事务消息。
 
要点二：超时订单的处理⽅案是⽣成订单后发送⼀个延迟30分钟后才被消费的消息到MQ。订单系统消费该延迟消息时，会验证订单是否已支付，否就调⽤取消订单的接⼝。
 
(2)对取消订单的处理

1. //订单中心-逆向售后业务接口
2. @DubboService(version = "1.0.0", interfaceClass = AfterSaleApi.class, retries = 0)
3. public class AfterSaleApiImpl implements AfterSaleApi {
4.     @Autowired
5.     private OrderLackService orderLackItemService;
7.     @Autowired
8.     private OrderAfterSaleService orderAfterSaleService;
10.     //取消订单/超时未支付取消
11.     @Override
12.     public JsonResult<Boolean> cancelOrder(CancelOrderRequest cancelOrderRequest) {
13.         try {
14.             return orderAfterSaleService.cancelOrder(cancelOrderRequest);
15.         } catch (OrderBizException e) {
16.             log.error("biz error", e);
17.             return JsonResult.buildError(e.getErrorCode(), e.getErrorMsg());
18.         } catch (Exception e) {
19.             log.error("system error", e);
20.             return JsonResult.buildError(e.getMessage());
21.         }
22.     }
23.     ...
24. }
26. //取消订单入参
27. @Data
28. public class CancelOrderRequest extends AbstractObject implements Serializable {
29.     private static final long serialVersionUID = -4035579903997700971L;
30.     private String orderId;//订单号
31.     private Integer businessIdentifier;//订单渠道来源
32.     private Integer cancelType;//订单取消类型 0-手动取消 1-超时未支付
33.     private String userId;//用户id
34.     private Integer orderType;//订单类型
35.     private Integer orderStatus;//订单状态
36.     private Integer oldOrderStatus;//原订单状态
37. }
39. @Service
40. public class OrderAfterSaleServiceImpl implements OrderAfterSaleService {
41.     ...
42.     //取消订单/超时未支付取消
43.     @Override
44.     public JsonResult<Boolean> cancelOrder(CancelOrderRequest cancelOrderRequest) {
45.         //入参检查
46.         checkCancelOrderRequestParam(cancelOrderRequest);
48.         //分布式锁
49.         String orderId = cancelOrderRequest.getOrderId();
50.         String key = RedisLockKeyConstants.CANCEL_KEY + orderId;
51.         try {
52.             boolean lock = redisLock.lock(key);
53.             if (!lock) {
54.                 throw new OrderBizException(OrderErrorCodeEnum.CANCEL_ORDER_REPEAT);
55.             }
56.             //执行取消订单
57.             return executeCancelOrder(cancelOrderRequest, orderId);
58.         } catch (Exception e) {
59.             throw new OrderBizException(e.getMessage());
60.         } finally {
61.             redisLock.unlock(key);
62.         }
63.     }
65.     @Override
66.     public JsonResult<Boolean> executeCancelOrder(CancelOrderRequest cancelOrderRequest, String orderId) {
67.         //1.组装数据
68.         OrderInfoDO orderInfoDO = findOrderInfo(orderId, cancelOrderRequest.getCancelType());
69.         CancelOrderAssembleRequest cancelOrderAssembleRequest = buildAssembleRequest(orderId, cancelOrderRequest, orderInfoDO);
70.         if (cancelOrderAssembleRequest.getOrderInfoDTO().getOrderStatus() >= OrderStatusEnum.OUT_STOCK.getCode()) {
71.             throw new OrderBizException(OrderErrorCodeEnum.CURRENT_ORDER_STATUS_CANNOT_CANCEL);
72.         }
74.         TransactionMQProducer producer = defaultProducer.getProducer();
75.         producer.setTransactionListener(new TransactionListener() {
76.             @Override
77.             public LocalTransactionState executeLocalTransaction(Message message, Object o) {
78.                 try {
79.                     //2.执行履约取消、更新订单状态、新增订单日志操作
80.                     afterSaleManager.cancelOrderFulfillmentAndUpdateOrderStatus(cancelOrderAssembleRequest);
81.                     return LocalTransactionState.COMMIT_MESSAGE;
82.                 } catch (Exception e) {
83.                     log.error("system error", e);
84.                     return LocalTransactionState.ROLLBACK_MESSAGE;
85.                 }
86.             }
88.             @Override
89.             public LocalTransactionState checkLocalTransaction(MessageExt messageExt) {
90.                 //查询订单状态是否已更新为"已取消"
91.                 OrderInfoDO orderInfoByDatabase = orderInfoDAO.getByOrderId(orderId);
92.                 if (OrderStatusEnum.CANCELED.getCode().equals(orderInfoByDatabase.getOrderStatus())) {
93.                     return LocalTransactionState.COMMIT_MESSAGE;
94.                 }
95.                 return LocalTransactionState.ROLLBACK_MESSAGE;
96.             }
97.         });
99.         try {
100.             Message message = new Message(RocketMqConstant.RELEASE_ASSETS_TOPIC, JSONObject.toJSONString(cancelOrderAssembleRequest).getBytes(StandardCharsets.UTF_8));
101.             //3.发送事务消息 释放权益资产
102.             TransactionSendResult result = producer.sendMessageInTransaction(message, cancelOrderAssembleRequest);
103.             if (!result.getLocalTransactionState().equals(LocalTransactionState.COMMIT_MESSAGE)) {
104.                 throw new OrderBizException(OrderErrorCodeEnum.CANCEL_ORDER_PROCESS_FAILED);
105.             }
106.             return JsonResult.buildSuccess(true);
107.         } catch (Exception e) {
108.             throw new OrderBizException(OrderErrorCodeEnum.SEND_TRANSACTION_MQ_FAILED);
109.         }
110.     }
111.     ...
112. }

复制代码

(3)对取消履约+更新订单状态+新增订单日志的处理

1. @Service
2. public class AfterSaleManagerImpl implements AfterSaleManager {
3.     ...
4.     @Override
5.     public void cancelOrderFulfillmentAndUpdateOrderStatus(CancelOrderAssembleRequest cancelOrderAssembleRequest) {
6.         //履约取消
7.         cancelFulfill(cancelOrderAssembleRequest);
9.         //更新订单状态和记录订单操作日志
10.         updateOrderStatusAndSaveOperationLog(cancelOrderAssembleRequest);
11.     }
13.     //调用履约拦截订单
14.     private void cancelFulfill(CancelOrderAssembleRequest cancelOrderAssembleRequest) {
15.         OrderInfoDTO orderInfoDTO = cancelOrderAssembleRequest.getOrderInfoDTO();
16.         if (OrderStatusEnum.CREATED.getCode().equals(orderInfoDTO.getOrderStatus())) {
17.             return;
18.         }
19.         CancelFulfillRequest cancelFulfillRequest = orderInfoDTO.clone(CancelFulfillRequest.class);
20.         JsonResult<Boolean> jsonResult = fulfillApi.cancelFulfill(cancelFulfillRequest);
21.         if (!jsonResult.getSuccess()) {
22.             throw new OrderBizException(OrderErrorCodeEnum.CANCEL_ORDER_FULFILL_ERROR);
23.         }
24.     }
26.     //更新订单状态和记录订单操作日志
27.     private void updateOrderStatusAndSaveOperationLog(CancelOrderAssembleRequest cancelOrderAssembleRequest) {
28.         //更新订单表
29.         OrderInfoDO orderInfoDO = cancelOrderAssembleRequest.getOrderInfoDTO().clone(OrderInfoDO.class);
30.         orderInfoDO.setCancelType(cancelOrderAssembleRequest.getCancelType().toString());
31.         orderInfoDO.setOrderStatus(OrderStatusEnum.CANCELED.getCode());
32.         orderInfoDO.setCancelTime(new Date());
33.         orderInfoDAO.updateOrderInfo(orderInfoDO);
34.         log.info("更新订单信息OrderInfo状态: orderId:{},status:{}", orderInfoDO.getOrderId(), orderInfoDO.getOrderStatus());
36.         //新增订单操作操作日志表
37.         Integer cancelType = Integer.valueOf(orderInfoDO.getCancelType());
38.         String orderId = orderInfoDO.getOrderId();
39.         OrderOperateLogDO orderOperateLogDO = new OrderOperateLogDO();
40.         orderOperateLogDO.setOrderId(orderId);
41.         orderOperateLogDO.setPreStatus(cancelOrderAssembleRequest.getOrderInfoDTO().getOrderStatus());
42.         orderOperateLogDO.setCurrentStatus(OrderStatusEnum.CANCELED.getCode());
43.         orderOperateLogDO.setOperateType(OrderOperateTypeEnum.AUTO_CANCEL_ORDER.getCode());
44.         if (OrderCancelTypeEnum.USER_CANCELED.getCode().equals(cancelType)) {
45.             orderOperateLogDO.setOperateType(OrderOperateTypeEnum.MANUAL_CANCEL_ORDER.getCode());
46.             orderOperateLogDO.setRemark(OrderOperateTypeEnum.MANUAL_CANCEL_ORDER.getMsg()
47.                 + orderOperateLogDO.getPreStatus() + "-" + orderOperateLogDO.getCurrentStatus());
48.         }
49.         if (OrderCancelTypeEnum.TIMEOUT_CANCELED.getCode().equals(cancelType)) {
50.             orderOperateLogDO.setOperateType(OrderOperateTypeEnum.AUTO_CANCEL_ORDER.getCode());
51.             orderOperateLogDO.setRemark(OrderOperateTypeEnum.AUTO_CANCEL_ORDER.getMsg()
52.                 + orderOperateLogDO.getPreStatus() + "-" + orderOperateLogDO.getCurrentStatus());
53.         }
54.         orderOperateLogDAO.save(orderOperateLogDO);
55.         log.info("新增订单操作日志OrderOperateLog状态,orderId:{}, PreStatus:{},CurrentStatus:{}", orderInfoDO.getOrderId(),
56.             orderOperateLogDO.getPreStatus(), orderOperateLogDO.getCurrentStatus());
57.     }
58.     ...
59. }

复制代码

(4)对释放资产的处理

1. @Configuration
2. public class ConsumerConfig {
3.     @Autowired
4.     private RocketMQProperties rocketMQProperties;
5.     ...
7.     //释放资产消息消费者
8.     @Bean("releaseAssetsConsumer")
9.     public DefaultMQPushConsumer releaseAssetsConsumer(ReleaseAssetsListener releaseAssetsListener) throws MQClientException {
10.         DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(RELEASE_ASSETS_CONSUMER_GROUP);
11.         consumer.setNamesrvAddr(rocketMQProperties.getNameServer());
12.         consumer.subscribe(RELEASE_ASSETS_TOPIC, "*");
13.         consumer.registerMessageListener(releaseAssetsListener);
14.         consumer.start();
15.         return consumer;
16.     }
17. }
19. //消费MQ的释放资产消息
20. @Component
21. public class ReleaseAssetsListener implements MessageListenerConcurrently {
22.     @Autowired
23.     private DefaultProducer defaultProducer;
25.     @Autowired
26.     private OrderItemDAO orderItemDAO;
28.     @Override
29.     public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
30.         try {
31.             for (MessageExt messageExt : list) {
32.                 //1.消费到释放资产message
33.                 String message = new String(messageExt.getBody());
34.                 log.info("ReleaseAssetsListener message:{}", message);
35.                 CancelOrderAssembleRequest cancelOrderAssembleRequest = JSONObject.parseObject(message, CancelOrderAssembleRequest.class);
36.                 OrderInfoDTO orderInfoDTO = cancelOrderAssembleRequest.getOrderInfoDTO();
38.                 //2.发送取消订单退款请求MQ
39.                 if (orderInfoDTO.getOrderStatus() > OrderStatusEnum.CREATED.getCode()) {
40.                     defaultProducer.sendMessage(RocketMqConstant.CANCEL_REFUND_REQUEST_TOPIC,
41.                         JSONObject.toJSONString(cancelOrderAssembleRequest), "取消订单退款");
42.                 }
44.                 //3.发送释放库存MQ
45.                 ReleaseProductStockRequest releaseProductStockRequest = buildReleaseProductStock(orderInfoDTO, orderItemDAO);
46.                 defaultProducer.sendMessage(RocketMqConstant.CANCEL_RELEASE_INVENTORY_TOPIC,
47.                     JSONObject.toJSONString(releaseProductStockRequest), "取消订单释放库存");
49.                 //4.发送释放优惠券MQ
50.                 if (!Strings.isNullOrEmpty(orderInfoDTO.getCouponId())) {
51.                     ReleaseUserCouponRequest releaseUserCouponRequest = buildReleaseUserCoupon(orderInfoDTO);
52.                     defaultProducer.sendMessage(RocketMqConstant.CANCEL_RELEASE_PROPERTY_TOPIC,
53.                         JSONObject.toJSONString(releaseUserCouponRequest), "取消订单释放优惠券");
54.                 }
55.             }
56.             return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
57.         } catch (Exception e) {
58.             log.error("consumer error", e);
59.             return ConsumeConcurrentlyStatus.RECONSUME_LATER;
60.         }
61.     }
63.     //组装释放优惠券数据
64.     private ReleaseUserCouponRequest buildReleaseUserCoupon(OrderInfoDTO orderInfoDTO) {
65.         ReleaseUserCouponRequest releaseUserCouponRequest = new ReleaseUserCouponRequest();
66.         releaseUserCouponRequest.setCouponId(orderInfoDTO.getCouponId());
67.         releaseUserCouponRequest.setUserId(orderInfoDTO.getUserId());
68.         releaseUserCouponRequest.setOrderId(orderInfoDTO.getOrderId());
69.         return releaseUserCouponRequest;
70.     }
72.     //组装释放库存数据
73.     private ReleaseProductStockRequest buildReleaseProductStock(OrderInfoDTO orderInfoDTO, OrderItemDAO orderItemDAO) {
74.         String orderId = orderInfoDTO.getOrderId();
75.         List<ReleaseProductStockRequest.OrderItemRequest> orderItemRequestList = new ArrayList<>();
77.         //查询订单条目
78.         ReleaseProductStockRequest.OrderItemRequest orderItemRequest;
79.         List<OrderItemDO> orderItemDOList = orderItemDAO.listByOrderId(orderId);
80.         for (OrderItemDO orderItemDO : orderItemDOList) {
81.             orderItemRequest = new ReleaseProductStockRequest.OrderItemRequest();
82.             orderItemRequest.setSkuCode(orderItemDO.getSkuCode());
83.             orderItemRequest.setSaleQuantity(orderItemDO.getSaleQuantity());
84.             orderItemRequestList.add(orderItemRequest);
85.         }
86.         ReleaseProductStockRequest releaseProductStockRequest = new ReleaseProductStockRequest();
87.         releaseProductStockRequest.setOrderId(orderInfoDTO.getOrderId());
88.         releaseProductStockRequest.setOrderItemRequestList(orderItemRequestList);
89.         return releaseProductStockRequest;
90.     }
91. }
93. @Configuration
94. public class ConsumerConfig {
95.     @Autowired
96.     private RocketMQProperties rocketMQProperties;
97.     ...
99.     //消费退款请求消息 消费者
100.     @Bean("cancelRefundConsumer")
101.     public DefaultMQPushConsumer cancelRefundConsumer(CancelRefundListener cancelRefundListener) throws MQClientException {
102.         DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(RocketMqConstant.REQUEST_CONSUMER_GROUP);
103.         consumer.setNamesrvAddr(rocketMQProperties.getNameServer());
104.         consumer.subscribe(RocketMqConstant.CANCEL_REFUND_REQUEST_TOPIC, "*");
105.         consumer.registerMessageListener(cancelRefundListener);
106.         consumer.start();
107.         return consumer;
108.     }
109. }
111. //消费取消订单退款请求的消息
112. @Component
113. public class CancelRefundListener implements MessageListenerConcurrently {
114.     @Autowired
115.     private OrderAfterSaleService orderAfterSaleService;
117.     @Override
118.     public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
119.         try {
120.             for (MessageExt messageExt : list) {
121.                 String message = new String(messageExt.getBody());
122.                 CancelOrderAssembleRequest cancelOrderAssembleRequest = JSONObject.parseObject(message, CancelOrderAssembleRequest.class);
123.                 log.info("CancelRefundConsumer message:{}", message);
124.                 //执行 取消订单/超时未支付取消 前的操作
125.                 JsonResult<Boolean> jsonResult = orderAfterSaleService.processCancelOrder(cancelOrderAssembleRequest);
126.                 if (!jsonResult.getSuccess()) {
127.                     throw new OrderBizException(OrderErrorCodeEnum.CONSUME_MQ_FAILED);
128.                 }
129.             }
130.             return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
131.         } catch (Exception e) {
132.             log.error("consumer error", e);
133.             return ConsumeConcurrentlyStatus.RECONSUME_LATER;
134.         }
135.     }
136. }
138. @Configuration
139. public class ConsumerConfig {
140.     @Autowired
141.     private RocketMQProperties rocketMQProperties;
143.     //释放库存消息消费者
144.     @Bean("releaseInventoryConsumer")
145.     public DefaultMQPushConsumer releaseInventoryConsumer(ReleaseInventoryListener releaseInventoryListener) throws MQClientException {
146.         DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(RocketMqConstant.RELEASE_INVENTORY_CONSUMER_GROUP);
147.         consumer.setNamesrvAddr(rocketMQProperties.getNameServer());
148.         consumer.subscribe(RocketMqConstant.CANCEL_RELEASE_INVENTORY_TOPIC, "*");
149.         consumer.registerMessageListener(releaseInventoryListener);
150.         consumer.start();
151.         return consumer;
152.     }
153.     ...
154. }
156. //消费释放库存的消息
157. @Component
158. public class ReleaseInventoryListener implements MessageListenerConcurrently {
159.     @DubboReference(version = "1.0.0")
160.     private InventoryApi inventoryApi;
162.     @Override
163.     public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
164.         try {
165.             for (MessageExt msg : list) {
166.                 String content = new String(msg.getBody(), StandardCharsets.UTF_8);
167.                 log.info("ReleaseInventoryConsumer message:{}", content);
168.                 CancelOrderReleaseProductStockRequest cancelOrderReleaseProductStockRequest = JSONObject.parseObject(content, CancelOrderReleaseProductStockRequest.class);
169.                 JsonResult<Boolean> jsonResult = inventoryApi.cancelOrderReleaseProductStock(cancelOrderReleaseProductStockRequest);
170.                 if (!jsonResult.getSuccess()) {
171.                     throw new InventoryBizException(InventoryErrorCodeEnum.CONSUME_MQ_FAILED);
172.                 }
173.             }
174.             return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
175.         } catch (Exception e) {
176.             log.error("consumer error", e);
177.             return ConsumeConcurrentlyStatus.RECONSUME_LATER;
178.         }
179.     }
180. }
182. @Configuration
183. public class ConsumerConfig {
184.     @Autowired
185.     private RocketMQProperties rocketMQProperties;
187.     //释放资产权益消息消费者
188.     @Bean("releaseInventoryConsumer")
189.     public DefaultMQPushConsumer releaseInventoryConsumer(ReleasePropertyListener releasePropertyListener) throws MQClientException {
190.         DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(RocketMqConstant.RELEASE_PROPERTY_CONSUMER_GROUP);
191.         consumer.setNamesrvAddr(rocketMQProperties.getNameServer());
192.         consumer.subscribe(RocketMqConstant.CANCEL_RELEASE_PROPERTY_TOPIC, "*");
193.         consumer.registerMessageListener(releasePropertyListener);
194.         consumer.start();
195.         return consumer;
196.     }
197.     ...
198. }
200. //消费释放优惠券的消息
201. @Component
202. public class ReleasePropertyListener implements MessageListenerConcurrently {
203.     @DubboReference(version = "1.0.0")
204.     private MarketApi marketApi;
206.     @Override
207.     public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
208.         try {
209.             for (MessageExt msg : list) {
210.                 String content = new String(msg.getBody(), StandardCharsets.UTF_8);
211.                 log.info("ReleasePropertyConsumer message:{}", content);
212.                 CancelOrderReleaseUserCouponRequest cancelOrderReleaseUserCouponRequest = JSONObject.parseObject(content, CancelOrderReleaseUserCouponRequest.class);
213.                 JsonResult<Boolean> jsonResult = marketApi.cancelOrderReleaseCoupon(cancelOrderReleaseUserCouponRequest);
214.                 if (!jsonResult.getSuccess()) {
215.                     throw new MarketBizException(MarketErrorCodeEnum.CONSUME_MQ_FAILED);
216.                 }
217.             }
218.             return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
219.         } catch (Exception e) {
220.             log.error("consumer error", e);
221.             return ConsumeConcurrentlyStatus.RECONSUME_LATER;
222.         }
223.     }
224. }

复制代码

(5)对退款的退款前处理

1. @Service
2. public class OrderAfterSaleServiceImpl implements OrderAfterSaleService {
3.     ...
4.     @Override
5.     public JsonResult<Boolean> processCancelOrder(CancelOrderAssembleRequest cancelOrderAssembleRequest) {
6.         String orderId = cancelOrderAssembleRequest.getOrderId();
7.         //分布式锁
8.         String key = RedisLockKeyConstants.REFUND_KEY + orderId;
9.         try {
10.             boolean lock = redisLock.lock(key);
11.             if (!lock) {
12.                 throw new OrderBizException(OrderErrorCodeEnum.PROCESS_REFUND_REPEAT);
13.             }
15.             //执行退款前的准备工作
16.             //生成售后订单号
17.             OrderInfoDTO orderInfoDTO = cancelOrderAssembleRequest.getOrderInfoDTO();
18.             OrderInfoDO orderInfoDO = orderInfoDTO.clone(OrderInfoDO.class);
19.             String afterSaleId = orderNoManager.genOrderId(OrderNoTypeEnum.AFTER_SALE.getCode(), orderInfoDO.getUserId());
21.             //1.计算 取消订单 退款金额
22.             CancelOrderRefundAmountDTO cancelOrderRefundAmountDTO = calculatingCancelOrderRefundAmount(cancelOrderAssembleRequest);
23.             cancelOrderAssembleRequest.setCancelOrderRefundAmountDTO(cancelOrderRefundAmountDTO);
25.             TransactionMQProducer producer = defaultProducer.getProducer();
26.             producer.setTransactionListener(new TransactionListener() {
27.                 @Override
28.                 public LocalTransactionState executeLocalTransaction(Message message, Object o) {
29.                     try {
30.                         //2.取消订单操作 记录售后信息
31.                         afterSaleManager.insertCancelOrderAfterSale(cancelOrderAssembleRequest, AfterSaleStatusEnum.REVIEW_PASS.getCode(), orderInfoDO, afterSaleId);
32.                         return LocalTransactionState.COMMIT_MESSAGE;
33.                     } catch (Exception e) {
34.                         log.error("system error", e);
35.                         return LocalTransactionState.ROLLBACK_MESSAGE;
36.                     }
37.                 }
39.                 @Override
40.                 public LocalTransactionState checkLocalTransaction(MessageExt messageExt) {
41.                     // 查询售后数据是否插入成功
42.                     AfterSaleInfoDO afterSaleInfoDO = afterSaleInfoDAO.getOneByAfterSaleId(Long.valueOf(afterSaleId));
43.                     List afterSaleItemDOList = afterSaleItemDAO.listByAfterSaleId(Long.valueOf(afterSaleId));
44.                     List afterSaleLogDOList = afterSaleLogDAO.listByAfterSaleId(Long.valueOf(afterSaleId));
45.                     List afterSaleRefundDOList = afterSaleRefundDAO.listByAfterSaleId(Long.valueOf(afterSaleId));
46.                     if (afterSaleInfoDO != null && afterSaleItemDOList.isEmpty() && afterSaleLogDOList.isEmpty() && afterSaleRefundDOList.isEmpty()) {
47.                         return LocalTransactionState.COMMIT_MESSAGE;
48.                     }
49.                     return LocalTransactionState.ROLLBACK_MESSAGE;
50.                 }
51.             });
53.             try {
54.                 //3.组装事务MQ消息
55.                 ActualRefundMessage actualRefundMessage = new ActualRefundMessage();
56.                 actualRefundMessage.setOrderId(cancelOrderAssembleRequest.getOrderId());
57.                 actualRefundMessage.setLastReturnGoods(cancelOrderAssembleRequest.isLastReturnGoods());
58.                 actualRefundMessage.setAfterSaleId(Long.valueOf(afterSaleId));
59.                 Message message = new Message(RocketMqConstant.ACTUAL_REFUND_TOPIC, JSONObject.toJSONString(actualRefundMessage).getBytes(StandardCharsets.UTF_8));
61.                 //4.发送事务MQ消息
62.                 TransactionSendResult result = producer.sendMessageInTransaction(message, actualRefundMessage);
63.                 if (!result.getLocalTransactionState().equals(LocalTransactionState.COMMIT_MESSAGE)) {
64.                     throw new OrderBizException(OrderErrorCodeEnum.PROCESS_REFUND_FAILED);
65.                 }
66.                 return JsonResult.buildSuccess(true);
67.             } catch (Exception e) {
68.                 throw new OrderBizException(OrderErrorCodeEnum.SEND_TRANSACTION_MQ_FAILED);
69.             }
70.         } finally {
71.             redisLock.unlock(key);
72.         }
73.     }
74.     ...
75. }
77. @Service
78. public class AfterSaleManagerImpl implements AfterSaleManager {
79.     ...
80.     //取消订单操作 记录售后信息
81.     @Transactional(rollbackFor = Exception.class)
82.     @Override
83.     public void insertCancelOrderAfterSale(CancelOrderAssembleRequest cancelOrderAssembleRequest, Integer afterSaleStatus, OrderInfoDO orderInfoDO, String afterSaleId) {
84.         OrderInfoDTO orderInfoDTO = cancelOrderAssembleRequest.getOrderInfoDTO();
85.         //取消订单过程中的 申请退款金额 和 实际退款金额 都是实付退款金额 金额相同
86.         AfterSaleInfoDO afterSaleInfoDO = new AfterSaleInfoDO();
87.         afterSaleInfoDO.setApplyRefundAmount(orderInfoDO.getPayAmount());
88.         afterSaleInfoDO.setRealRefundAmount(orderInfoDO.getPayAmount());
90.         //1.新增售后订单表
91.         Integer cancelOrderAfterSaleStatus = AfterSaleStatusEnum.REVIEW_PASS.getCode();
92.         insertCancelOrderAfterSaleInfoTable(orderInfoDO, cancelOrderAfterSaleStatus, afterSaleInfoDO, afterSaleId);
93.         cancelOrderAssembleRequest.setAfterSaleId(afterSaleId);
95.         //2.新增售后条目表
96.         String orderId = cancelOrderAssembleRequest.getOrderId();
97.         List<OrderItemDTO> orderItemDTOList = cancelOrderAssembleRequest.getOrderItemDTOList();
98.         insertAfterSaleItemTable(orderId, orderItemDTOList, afterSaleId);
100.         //3.新增售后变更表
101.         insertCancelOrderAfterSaleLogTable(afterSaleId, orderInfoDTO, AfterSaleStatusEnum.UN_CREATED.getCode(), afterSaleStatus);
103.         //4.新增售后支付表
104.         AfterSaleRefundDO afterSaleRefundDO = insertAfterSaleRefundTable(orderInfoDTO, afterSaleId, afterSaleInfoDO);
105.         cancelOrderAssembleRequest.setAfterSaleRefundId(afterSaleRefundDO.getId());
106.     }
107.     ...
108. }

复制代码

(6)对退款的实际退款处理

1. @Configuration
2. public class ConsumerConfig {
3.     @Autowired
4.     private RocketMQProperties rocketMQProperties;
5.     ...
7.     @Bean("actualRefundConsumer")
8.     public DefaultMQPushConsumer actualRefundConsumer(ActualRefundListener actualRefundListener) throws MQClientException {
9.         DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(ACTUAL_REFUND_CONSUMER_GROUP);
10.         consumer.setNamesrvAddr(rocketMQProperties.getNameServer());
11.         consumer.subscribe(ACTUAL_REFUND_TOPIC, "*");
12.         consumer.registerMessageListener(actualRefundListener);
13.         consumer.start();
14.         return consumer;
15.     }
16. }
18. //实际退款消费者
19. @Component
20. public class ActualRefundListener implements MessageListenerConcurrently {
21.     @Autowired
22.     private OrderAfterSaleService orderAfterSaleService;
24.     @Override
25.     public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
26.         try {
27.             for (MessageExt messageExt : list) {
28.                 String message = new String(messageExt.getBody());
29.                 ActualRefundMessage actualRefundMessage = JSONObject.parseObject(message, ActualRefundMessage.class);
30.                 log.info("ActualRefundConsumer message:{}", message);
31.                 JsonResult<Boolean> jsonResult = orderAfterSaleService.refundMoney(actualRefundMessage);
32.                 if (!jsonResult.getSuccess()) {
33.                     throw new OrderBizException(jsonResult.getErrorCode(), jsonResult.getErrorMessage());
34.                 }
35.             }
36.             return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
37.         } catch (Exception e) {
38.             log.error("consumer error", e);
39.             return ConsumeConcurrentlyStatus.RECONSUME_LATER;
40.         }
41.     }
42. }
44. @Service
45. public class OrderAfterSaleServiceImpl implements OrderAfterSaleService {
46.     ...
47.     //执行退款
48.     @Override
49.     public JsonResult<Boolean> refundMoney(ActualRefundMessage actualRefundMessage) {
50.         Long afterSaleId = actualRefundMessage.getAfterSaleId();
51.         String key = RedisLockKeyConstants.REFUND_KEY + afterSaleId;
52.         try {
53.             boolean lock = redisLock.lock(key);
54.             if (!lock) {
55.                 throw new OrderBizException(OrderErrorCodeEnum.REFUND_MONEY_REPEAT);
56.             }
57.             AfterSaleInfoDO afterSaleInfoDO = afterSaleInfoDAO.getOneByAfterSaleId(actualRefundMessage.getAfterSaleId());
58.             AfterSaleRefundDO afterSaleRefundDO = afterSaleRefundDAO.findOrderAfterSaleStatus(String.valueOf(afterSaleId));
60.             //1.封装调用支付退款接口的数据
61.             PayRefundRequest payRefundRequest = buildPayRefundRequest(actualRefundMessage, afterSaleRefundDO);
63.             //2.执行退款
64.             if (!payApi.executeRefund(payRefundRequest)) {
65.                 throw new OrderBizException(OrderErrorCodeEnum.ORDER_REFUND_AMOUNT_FAILED);
66.             }
68.             //3.本次售后的订单条目是当前订单的最后一笔，发送事务MQ退优惠券，此时isLastReturnGoods标记是true
69.             if (actualRefundMessage.isLastReturnGoods()) {
70.                 TransactionMQProducer producer = defaultProducer.getProducer();
71.                 //组装事务MQ消息体
72.                 ReleaseUserCouponRequest releaseUserCouponRequest = buildLastOrderReleasesCouponMessage(producer, afterSaleInfoDO, afterSaleId, actualRefundMessage);
73.                 try {
74.                     //4.发送事务消息 释放优惠券
75.                     Message message = new Message(RocketMqConstant.CANCEL_RELEASE_PROPERTY_TOPIC, JSONObject.toJSONString(releaseUserCouponRequest).getBytes(StandardCharsets.UTF_8));
76.                     TransactionSendResult result = producer.sendMessageInTransaction(message, releaseUserCouponRequest);
77.                     if (!result.getLocalTransactionState().equals(LocalTransactionState.COMMIT_MESSAGE)) {
78.                         throw new OrderBizException(OrderErrorCodeEnum.REFUND_MONEY_RELEASE_COUPON_FAILED);
79.                     }
80.                     return JsonResult.buildSuccess(true);
81.                 } catch (Exception e) {
82.                     throw new OrderBizException(OrderErrorCodeEnum.SEND_TRANSACTION_MQ_FAILED);
83.                 }
84.             } else {
85.                 //当前售后条目非本订单的最后一笔 和 取消订单，在此更新售后状态后流程结束
86.                 //更新售后单状态
87.                 updateAfterSaleStatus(afterSaleInfoDO, AfterSaleStatusEnum.REVIEW_PASS.getCode(), AfterSaleStatusEnum.REFUNDING.getCode());
88.                 return JsonResult.buildSuccess(true);
89.             }
90.         } catch (OrderBizException e) {
91.             log.error("system error", e);
92.             return JsonResult.buildError(e.getMessage());
93.         } finally {
94.             redisLock.unlock(key);
95.         }
96.     }
98.     private ReleaseUserCouponRequest buildLastOrderReleasesCouponMessage(TransactionMQProducer producer, AfterSaleInfoDO afterSaleInfoDO, Long afterSaleId, ActualRefundMessage actualRefundMessage) {
99.         producer.setTransactionListener(new TransactionListener() {
100.             @Override
101.             public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
102.                 try {
103.                     //更新售后单状态
104.                     updateAfterSaleStatus(afterSaleInfoDO, AfterSaleStatusEnum.REVIEW_PASS.getCode(), AfterSaleStatusEnum.REFUNDING.getCode());
105.                     return LocalTransactionState.COMMIT_MESSAGE;
106.                 } catch (Exception e) {
107.                     log.error("system error", e);
108.                     return LocalTransactionState.ROLLBACK_MESSAGE;
109.                 }
110.             }
112.             @Override
113.             public LocalTransactionState checkLocalTransaction(MessageExt msg) {
114.                 //查询售后单状态是"退款中"
115.                 AfterSaleInfoDO afterSaleInfoDO = afterSaleInfoDAO.getOneByAfterSaleId(afterSaleId);
116.                 if (AfterSaleStatusEnum.REFUNDING.getCode().equals(afterSaleInfoDO.getAfterSaleStatus())) {
117.                     return LocalTransactionState.COMMIT_MESSAGE;
118.                 }
119.                 return LocalTransactionState.ROLLBACK_MESSAGE;
120.             }
121.         });
123.         //组装释放优惠券权益消息数据
124.         String orderId = actualRefundMessage.getOrderId();
125.         OrderInfoDO orderInfoDO = orderInfoDAO.getByOrderId(orderId);
126.         ReleaseUserCouponRequest releaseUserCouponRequest = new ReleaseUserCouponRequest();
127.         releaseUserCouponRequest.setCouponId(orderInfoDO.getCouponId());
128.         releaseUserCouponRequest.setUserId(orderInfoDO.getUserId());
129.         return releaseUserCouponRequest;
130.     }
131.     ...
132. }

复制代码

 
9.为什么拦截履约要同步而释放资产是异步
(1)在正向订单流程中触发履约采用异步柔性事务
(2)在逆向订单流程中拦截履约采用同步刚性事务
 
(1)在正向订单流程中触发履约采用异步柔性事务
生成订单写库、锁优惠券、锁库存这三个动作会绑定在一个刚性事务里，从而实现"生成订单写库 + 锁优惠券 + 锁库存"的强同步效果。但是当支付成功完成支付回调后，推送订单消息进行履约时，则通过MQ异步推送。也就是说："生单 + 锁优惠券 + 锁库存"采用了同步刚性事务，而"推送订单消息触发履约"则采用了异步柔性事务。
 
原因一：在正向订单流程中，生单会使用到优惠券及涉及扣减库存。如果不采用同步刚性事务，则可能会出现虽然订单已经生成了，但是用户却还是能看到其优惠券的状态还处于未使用，或库存还没扣减。这样就可能对用户造成困惑，数据产生了短暂的不一致问题。严重可能会导致用户重复使用优惠券，以及库存超卖问题。
 
所以"生单 + 锁优惠券 + 锁库存"采用同步刚性事务，当生单操作一旦完成，用户就可以看到优惠券已使用 + 库存已扣减。

原因二：推送订单消息触发履约可以采用异步柔性事务，是因为用户支付完一个订单后，并没有必要马上看到订单已在履约处理，所以"推送订单进行履约"是天然可以被用户接受延迟的。
 
(2)在逆向订单流程中拦截履约采用同步刚性事务
与正向订单流程刚好相反："修改订单状态 + 拦截订单不被履约"采用了同步刚性事务，而"释放优惠券 + 释放库存"则采用了异步柔性事务。因为取消订单的第一要务必然是拦截履约。
 
如果"修改订单状态 + 拦截订单不被履约"采用了异步柔性事务，那么可能出现订单状态已经改为已取消，但是由于异步拦截履约慢了，导致订单都已经被打包发货了，这样就可能发生纠纷和资损了。
 
如果订单还没被支付就进行了取消，此时订单状态被修改后，可以不用立刻关注库存和优惠券。在用户看来，其还没释放的优惠券最多不能马上恢复使用，影响不严重。最差也不会出现优惠券被用户重复使用的恶劣情况，以及库存超卖问题。
 
10.支付回调分别互斥预支付和取消订单

首先，预支付和支付回调会加同一把锁：ORDER_PAY_KEY。然后，支付回调和取消订单也会加同一把锁：CANCEL_KEY。所以支付回调会加两把锁：ORDER_PAY_KEY和CANCEL_KEY，这可以通过Redisson的MutiLock来实现。

1. @Service
2. public class OrderServiceImpl implements OrderService {
3.     ...
4.     //预支付订单
5.     @Override
6.     @Transactional(rollbackFor = Exception.class)
7.     public PrePayOrderDTO prePayOrder(PrePayOrderRequest prePayOrderRequest) {
8.         //入参检查
9.         checkPrePayOrderRequestParam(prePayOrderRequest);
10.         String orderId = prePayOrderRequest.getOrderId();
11.         Integer payAmount = prePayOrderRequest.getPayAmount();
13.         //加分布式锁(与订单支付回调时加的是同一把锁)
14.         String key = RedisLockKeyConstants.ORDER_PAY_KEY + orderId;
15.         boolean lock = redisLock.lock(key);
16.         if (!lock) {
17.             throw new OrderBizException(OrderErrorCodeEnum.ORDER_PRE_PAY_ERROR);
18.         }
19.         try {
20.             ...
21.         } finally {
22.             //释放分布式锁
23.             redisLock.unlock(key);
24.         }
25.     }
26.     ...
28.     //支付回调
29.     //支付回调有2把分布式锁的原因说明：同一笔订单在同一时间只能支付or取消
30.     //不可以同时对一笔订单，既发起支付，又发起取消
31.     @Override
32.     public void payCallback(PayCallbackRequest payCallbackRequest) {
33.         //入参检查
34.         checkPayCallbackRequestParam(payCallbackRequest);
35.         String orderId = payCallbackRequest.getOrderId();
36.         Integer payAmount = payCallbackRequest.getPayAmount();
37.         Integer payType = payCallbackRequest.getPayType();
38.         List<String> redisKeyList = Lists.newArrayList();
40.         //加支付分布式锁避免支付系统并发回调
41.         String orderPayKey = RedisLockKeyConstants.ORDER_PAY_KEY + orderId;
43.         //加取消订单分布式锁避免支付和取消订单同时操作同一笔订单
44.         String cancelOrderKey = RedisLockKeyConstants.CANCEL_KEY + orderId;
45.         redisKeyList.add(orderPayKey);
46.         redisKeyList.add(cancelOrderKey);
48.         boolean lock = redisLock.multiLock(redisKeyList);
49.         if (!lock) {
50.             throw new OrderBizException(OrderErrorCodeEnum.ORDER_PAY_CALLBACK_ERROR);
51.         }
52.         try {
53.             ...
54.         } catch (Exception e) {
55.             throw new OrderBizException(e.getMessage());
56.         } finally {
57.             //释放分布式锁
58.             redisLock.unMultiLock(redisKeyList);
59.         }
60.     }
61.     ...
62. }
64. @Service
65. public class OrderAfterSaleServiceImpl implements OrderAfterSaleService {
66.     ...
67.     //取消订单/超时未支付取消
68.     @Override
69.     @GlobalTransactional(rollbackFor = Exception.class)
70.     public JsonResult<Boolean> cancelOrder(CancelOrderRequest cancelOrderRequest) {
71.         //入参检查
72.         checkCancelOrderRequestParam(cancelOrderRequest);
74.         //分布式锁
75.         String orderId = cancelOrderRequest.getOrderId();
76.         String key = RedisLockKeyConstants.CANCEL_KEY + orderId;
77.         boolean lock = redisLock.lock(key);
78.         if (!lock) {
79.             throw new OrderBizException(OrderErrorCodeEnum.CANCEL_ORDER_REPEAT);
80.         }
81.         try {
82.             //执行取消订单
83.             return executeCancelOrder(cancelOrderRequest, orderId);
84.         } catch (Exception e) {
85.             log.error("biz error", e);
86.             throw new OrderBizException(e.getMessage());
87.         } finally {
88.             redisLock.unlock(key);
89.         }
90.     }
91.     ...
92. }

复制代码

 
11.支付回调和取消订单互斥下的两种场景分析
支付回调的处理过程中，会对异常场景进行判断：
 
一.如果订单状态是"已创建"
那么会更新订单状态为已支付，并发送事务消息。
 
二.如果订单状态不是"已创建"
那么判断订单状态是否是取消状态，如果是取消状态，则继续判断订单是否未支付。若是未支付，则进行退款，并抛出异常。若不是未支付，则按支付方式抛出不同的异常。如果不是取消状态，则判断支付回调是否是同种支付方式。如果是，则返回。如果不是，则进行退款，并抛出异常。

1. @Service
2. public class OrderServiceImpl implements OrderService {
3.     ...
4.     //支付回调
5.     //支付回调有2把分布式锁的原因说明：同一笔订单在同一时间只能支付or取消
6.     //不可以同时对一笔订单，既发起支付，又发起取消
7.     @Override
8.     public void payCallback(PayCallbackRequest payCallbackRequest) {
9.         //1.入参检查
10.         checkPayCallbackRequestParam(payCallbackRequest);
12.         String orderId = payCallbackRequest.getOrderId();
13.         Integer payAmount = payCallbackRequest.getPayAmount();
14.         Integer payType = payCallbackRequest.getPayType();
15.         List<String> redisKeyList = Lists.newArrayList();
17.         //2.加支付分布式锁避免支付系统并发回调
18.         String orderPayKey = RedisLockKeyConstants.ORDER_PAY_KEY + orderId;
19.         //加取消订单分布式锁避免支付和取消订单同时操作同一笔订单
20.         String cancelOrderKey = RedisLockKeyConstants.CANCEL_KEY + orderId;
21.         redisKeyList.add(orderPayKey);
22.         redisKeyList.add(cancelOrderKey);
23.         boolean lock = redisLock.multiLock(redisKeyList);
24.         if (!lock) {
25.             throw new OrderBizException(OrderErrorCodeEnum.ORDER_PAY_CALLBACK_ERROR);
26.         }
27.         try {
28.             //从数据库中查询出当前订单信息
29.             OrderInfoDO orderInfoDO = orderInfoDAO.getByOrderId(orderId);
30.             OrderPaymentDetailDO orderPaymentDetailDO = orderPaymentDetailDAO.getPaymentDetailByOrderId(orderId);
31.             //3.校验参数
32.             if (orderInfoDO == null || orderPaymentDetailDO == null) {
33.                 throw new OrderBizException(OrderErrorCodeEnum.ORDER_INFO_IS_NULL);
34.             }
35.             if (!payAmount.equals(orderInfoDO.getPayAmount())) {
36.                 throw new OrderBizException(OrderErrorCodeEnum.ORDER_CALLBACK_PAY_AMOUNT_ERROR);
37.             }
39.             //4.异常场景判断
40.             Integer orderStatus = orderInfoDO.getOrderStatus();
41.             if (OrderStatusEnum.CREATED.getCode().equals(orderStatus)) {
42.                 //如果订单状态是"已创建"，直接更新订单状态为已支付，并发送事务消息
43.                 TransactionMQProducer transactionMQProducer = defaultProducer.getProducer();
44.                 transactionMQProducer.setTransactionListener(new TransactionListener() {
45.                     @Override
46.                     public LocalTransactionState executeLocalTransaction(Message message, Object o) {
47.                         try {
48.                             orderManager.updateOrderStatusPaid(payCallbackRequest, orderInfoDO, orderPaymentDetailDO);
49.                             return LocalTransactionState.COMMIT_MESSAGE;
50.                         } catch (BaseBizException e) {
51.                             throw e;
52.                         } catch (Exception e) {
53.                             log.error("system error", e);
54.                             return LocalTransactionState.ROLLBACK_MESSAGE;
55.                         }
56.                     }
58.                     @Override
59.                     public LocalTransactionState checkLocalTransaction(MessageExt messageExt) {
60.                         //检查订单是否是已支付
61.                         OrderInfoDO orderInfoDO = orderInfoDAO.getByOrderId(orderId);
62.                         if (orderInfoDO != null && OrderStatusEnum.PAID.getCode().equals(orderInfoDO.getOrderStatus())) {
63.                             return LocalTransactionState.COMMIT_MESSAGE;
64.                         }
65.                         return LocalTransactionState.ROLLBACK_MESSAGE;
66.                     }
67.                 });
68.                 //发送 "订单已完成支付" 消息
69.                 sendPaidOrderSuccessMessage(transactionMQProducer, orderInfoDO);
70.             } else {
71.                 //如果订单状态不是 "已创建"
72.                 if (OrderStatusEnum.CANCELED.getCode().equals(orderStatus)) {
73.                     //如果订单状态是取消状态
74.                     Integer payStatus = orderPaymentDetailDO.getPayStatus();
75.                     if (PayStatusEnum.UNPAID.getCode().equals(payStatus)) {
76.                         //调用退款
77.                         executeOrderRefund(orderInfoDO, orderPaymentDetailDO);
78.                         throw new OrderBizException(OrderErrorCodeEnum.ORDER_CANCEL_PAY_CALLBACK_ERROR);
79.                     } else if (PayStatusEnum.PAID.getCode().equals(payStatus)) {
80.                         if (payType.equals(orderPaymentDetailDO.getPayType())) {
81.                             throw new OrderBizException(OrderErrorCodeEnum.ORDER_CANCEL_PAY_CALLBACK_PAY_TYPE_SAME_ERROR);
82.                         } else {
83.                             throw new OrderBizException(OrderErrorCodeEnum.ORDER_CANCEL_PAY_CALLBACK_PAY_TYPE_NO_SAME_ERROR);
84.                         }
85.                     }
86.                 } else {
87.                     //如果订单状态不是取消状态
88.                     if (PayStatusEnum.PAID.getCode().equals(orderPaymentDetailDO.getPayStatus())) {
89.                         if (payType.equals(orderPaymentDetailDO.getPayType())) {
90.                             return;
91.                         }
92.                         //调用退款
93.                         executeOrderRefund(orderInfoDO, orderPaymentDetailDO);
94.                         throw new OrderBizException(OrderErrorCodeEnum.ORDER_CANCEL_PAY_CALLBACK_REPEAT_ERROR);
95.                     }
96.                 }
97.             }
98.         } catch (Exception e) {
99.             throw new OrderBizException(e.getMessage());
100.         } finally {
101.             //释放分布式锁
102.             redisLock.unMultiLock(redisKeyList);
103.         }
104.     }
106.     //发送订单已完成支付消息，触发订单进行履约
107.     private void sendPaidOrderSuccessMessage(TransactionMQProducer transactionMQProducer, OrderInfoDO orderInfoDO) throws MQClientException {
108.         String orderId = orderInfoDO.getOrderId();
109.         PaidOrderSuccessMessage message = new PaidOrderSuccessMessage();
110.         message.setOrderId(orderId);
111.         String topic = RocketMqConstant.PAID_ORDER_SUCCESS_TOPIC;
112.         byte[] body = JSON.toJSONString(message).getBytes(StandardCharsets.UTF_8);
113.         Message mq = new Message(topic, body);
114.         TransactionSendResult result = transactionMQProducer.sendMessageInTransaction(mq, orderInfoDO);
115.         if (!result.getLocalTransactionState().equals(LocalTransactionState.COMMIT_MESSAGE)) {
116.             throw new OrderBizException(OrderErrorCodeEnum.ORDER_PAY_CALLBACK_SEND_MQ_ERROR);
117.         }
118.     }
119.     ...
120. }
122. //支付系统回调请求对象
123. @Data
124. public class PayCallbackRequest extends AbstractObject implements Serializable {
125.     private static final long serialVersionUID = 3685085492927992753L;
126.     private String orderId;//订单ID
127.     private String payAccount;//支付账户
128.     private Integer payAmount;//支付金额
129.     private String outTradeNo;//支付系统交易单号
130.     private Integer payType;//支付方式
131.     private String merchantId;//商户号
132.     private String payChannel;//支付渠道
133.     private String appid;//微信平台 appid
134. }

复制代码

 
12.拦截履约的具体业务流程
流程图如下：

代码如下：
一.取消订单入口

1. @Service
2. public class OrderAfterSaleServiceImpl implements OrderAfterSaleService {
3.     @Autowired
4.     private AfterSaleManager afterSaleManager;
5.     ...
7.     //取消订单/超时未支付取消
8.     @Override
9.     @GlobalTransactional(rollbackFor = Exception.class)
10.     public JsonResult<Boolean> cancelOrder(CancelOrderRequest cancelOrderRequest) {
11.         //入参检查
12.         checkCancelOrderRequestParam(cancelOrderRequest);
13.         //分布式锁
14.         String orderId = cancelOrderRequest.getOrderId();
15.         String key = RedisLockKeyConstants.CANCEL_KEY + orderId;
16.         boolean lock = redisLock.lock(key);
17.         if (!lock) {
18.             throw new OrderBizException(OrderErrorCodeEnum.CANCEL_ORDER_REPEAT);
19.         }
20.         try {
21.             //执行取消订单
22.             return executeCancelOrder(cancelOrderRequest, orderId);
23.         } catch (Exception e) {
24.             log.error("biz error", e);
25.             throw new OrderBizException(e.getMessage());
26.         } finally {
27.             redisLock.unlock(key);
28.         }
29.     }
31.     @Override
32.     public JsonResult<Boolean> executeCancelOrder(CancelOrderRequest cancelOrderRequest, String orderId) {
33.         //1.组装数据
34.         OrderInfoDO orderInfoDO = findOrderInfo(orderId, cancelOrderRequest.getCancelType());
35.         CancelOrderAssembleRequest cancelOrderAssembleRequest = buildAssembleRequest(orderId, cancelOrderRequest, orderInfoDO);
36.         if (cancelOrderAssembleRequest.getOrderInfoDTO().getOrderStatus() >= OrderStatusEnum.OUT_STOCK.getCode()) {
37.             throw new OrderBizException(OrderErrorCodeEnum.CURRENT_ORDER_STATUS_CANNOT_CANCEL);
38.         }
39.         TransactionMQProducer producer = defaultProducer.getProducer();
40.         producer.setTransactionListener(new TransactionListener() {
41.             @Override
42.             public LocalTransactionState executeLocalTransaction(Message message, Object o) {
43.                 try {
44.                     //2.执行履约取消、更新订单状态、新增订单日志操作
45.                     afterSaleManager.cancelOrderFulfillmentAndUpdateOrderStatus(cancelOrderAssembleRequest);
46.                     return LocalTransactionState.COMMIT_MESSAGE;
47.                 } catch (Exception e) {
48.                     log.error("system error", e);
49.                     return LocalTransactionState.ROLLBACK_MESSAGE;
50.                 }
51.             }
53.             @Override
54.             public LocalTransactionState checkLocalTransaction(MessageExt messageExt) {
55.                 //查询订单状态是否已更新为"已取消"
56.                 OrderInfoDO orderInfoByDatabase = orderInfoDAO.getByOrderId(orderId);
57.                 if (OrderStatusEnum.CANCELED.getCode().equals(orderInfoByDatabase.getOrderStatus())) {
58.                     return LocalTransactionState.COMMIT_MESSAGE;
59.                 }
60.                 return LocalTransactionState.ROLLBACK_MESSAGE;
61.             }
62.         });
64.         try {
65.             Message message = new Message(RocketMqConstant.RELEASE_ASSETS_TOPIC, JSONObject.toJSONString(cancelOrderAssembleRequest).getBytes(StandardCharsets.UTF_8));
66.             //3.发送事务消息 释放权益资产
67.             TransactionSendResult result = producer.sendMessageInTransaction(message, cancelOrderAssembleRequest);
68.             if (!result.getLocalTransactionState().equals(LocalTransactionState.COMMIT_MESSAGE)) {
69.                 throw new OrderBizException(OrderErrorCodeEnum.CANCEL_ORDER_PROCESS_FAILED);
70.             }
71.             return JsonResult.buildSuccess(true);
72.         } catch (Exception e) {
73.             throw new OrderBizException(OrderErrorCodeEnum.SEND_TRANSACTION_MQ_FAILED);
74.         }
75.     }
76.     ...
77. }

复制代码

二.拦截履约处理

1. @Service
2. public class AfterSaleManagerImpl implements AfterSaleManager {
3.     @DubboReference(version = "1.0.0")
4.     private FulfillApi fulfillApi;
5.     ...
7.     @Override
8.     public void cancelOrderFulfillmentAndUpdateOrderStatus(CancelOrderAssembleRequest cancelOrderAssembleRequest) {
9.         //履约取消
10.         cancelFulfill(cancelOrderAssembleRequest);
11.         //更新订单状态和记录订单操作日志
12.         updateOrderStatusAndSaveOperationLog(cancelOrderAssembleRequest);
13.     }
15.     //调用履约拦截订单
16.     private void cancelFulfill(CancelOrderAssembleRequest cancelOrderAssembleRequest) {
17.         OrderInfoDTO orderInfoDTO = cancelOrderAssembleRequest.getOrderInfoDTO();
18.         if (OrderStatusEnum.CREATED.getCode().equals(orderInfoDTO.getOrderStatus())) {
19.             return;
20.         }
21.         CancelFulfillRequest cancelFulfillRequest = orderInfoDTO.clone(CancelFulfillRequest.class);
22.         JsonResult<Boolean> jsonResult = fulfillApi.cancelFulfill(cancelFulfillRequest);
23.         if (!jsonResult.getSuccess()) {
24.             throw new OrderBizException(OrderErrorCodeEnum.CANCEL_ORDER_FULFILL_ERROR);
25.         }
26.     }
27.     ...
28. }
30. @DubboService(version = "1.0.0", interfaceClass = FulfillApi.class, retries = 0)
31. public class FulfillApiImpl implements FulfillApi {
32.     @Autowired
33.     private FulfillService fulfillService;
35.     @DubboReference(version = "1.0.0", retries = 0)
36.     private WmsApi wmsApi;
38.     @DubboReference(version = "1.0.0", retries = 0)
39.     private TmsApi tmsApi;
40.     ...
42.     @Override
43.     public JsonResult<Boolean> cancelFulfill(CancelFulfillRequest cancelFulfillRequest) {
44.         log.info("取消履约：request={}", JSONObject.toJSONString(cancelFulfillRequest));
45.         //1.取消履约单
46.         fulfillService.cancelFulfillOrder(request.getOrderId());
47.         //2.取消捡货
48.         wmsApi.cancelPickGoods(request.getOrderId());
49.         //3.取消发货
50.         tmsApi.cancelSendOut(request.getOrderId());
52.         return JsonResult.buildSuccess(true);
53.     }
54.     ...
55. }
57. @Service
58. public class FulfillServiceImpl implements FulfillService {
59.     @Autowired
60.     private OrderFulfillDAO orderFulfillDAO;
62.     @Autowired
63.     private OrderFulfillItemDAO orderFulfillItemDAO;
64.     ...
66.     @Override
67.     public void cancelFulfillOrder(String orderId) {
68.         //1.查询履约单
69.         OrderFulfillDO orderFulfill = orderFulfillDAO.getOne(orderId);
70.         //2.移除履约单
71.         if (null != orderFulfill) {
72.             orderFulfillDAO.removeById(orderFulfill.getId());
73.             //3.查询履约单条目
74.             List<OrderFulfillItemDO> fulfillItems = orderFulfillItemDAO.listByFulfillId(orderFulfill.getFulfillId());
75.             //4.移除履约单条目
76.             for (OrderFulfillItemDO item : fulfillItems) {
77.                 orderFulfillItemDAO.removeById(item.getId());
78.             }
79.         }
80.     }
81.     ...
82. }
84. @DubboService(version = "1.0.0", interfaceClass = WmsApi.class, retries = 0)
85. public class WmsApiImpl implements WmsApi {
86.     ...
87.     @Transactional(rollbackFor = Exception.class)
88.     @Override
89.     public JsonResult<Boolean> cancelPickGoods(String orderId) {
90.         log.info("取消捡货，orderId={}", orderId);
91.         //1.查询出库单
92.         List<DeliveryOrderDO> deliveryOrders = deliveryOrderDAO.listByOrderId(orderId);
93.         //2.移除出库单和条目
94.         if (CollectionUtils.isNotEmpty(deliveryOrders)) {
95.             for (DeliveryOrderDO order : deliveryOrders) {
96.                 List<DeliveryOrderItemDO> items = deliveryOrderItemDAO.listByDeliveryOrderId(order.getDeliveryOrderId());
97.                 for (DeliveryOrderItemDO item : items) {
98.                     deliveryOrderItemDAO.removeById(item.getId());
99.                 }
100.                 deliveryOrderDAO.removeById(order.getId());
101.             }
102.         }
103.         return JsonResult.buildSuccess(true);
104.     }
105.     ...
106. }
108. @DubboService(version = "1.0.0", interfaceClass = TmsApi.class, retries = 0)
109. public class TmsApiImpl implements TmsApi {
110.     ...
111.     @Transactional(rollbackFor = Exception.class)
112.     @Override
113.     public JsonResult<Boolean> cancelSendOut(String orderId) {
114.         log.info("取消发货，orderId={}", orderId);
115.         //1.查询物流单
116.         List<LogisticOrderDO> logisticOrders = logisticOrderDAO.listByOrderId(orderId);
117.         //2.移除物流单
118.         for (LogisticOrderDO order : logisticOrders) {
119.             logisticOrderDAO.removeById(order.getId());
120.         }
121.         return JsonResult.buildSuccess(true);
122.     }
123.     ...
124. }

复制代码

 
13.拦截履约和取消订单的Seata事务原理分析
(1)取消订单接口使用Seata事务的AT模式
(2)Seata事务AT模式的原理
 
(1)取消订单接口使用Seata事务的AT模式
如果拦截履约时发现，订单已经出库配送，那么就会拦截失败。当拦截失败时，就不能更新订单状态了，可使用Seata的刚性事务实现。具体就是，在分布式事务入口添加@GlobalTransactional注解，在各分支事务添加@Transactional注解。

1. @Service
2. public class OrderAfterSaleServiceImpl implements OrderAfterSaleService {
3.     ...
4.     //取消订单/超时未支付取消
5.     @Override
6.     @GlobalTransactional(rollbackFor = Exception.class)
7.     public JsonResult<Boolean> cancelOrder(CancelOrderRequest cancelOrderRequest) {
8.         //入参检查
9.         checkCancelOrderRequestParam(cancelOrderRequest);
10.         //分布式锁
11.         String orderId = cancelOrderRequest.getOrderId();
12.         String key = RedisLockKeyConstants.CANCEL_KEY + orderId;
13.         boolean lock = redisLock.lock(key);
14.         if (!lock) {
15.             throw new OrderBizException(OrderErrorCodeEnum.CANCEL_ORDER_REPEAT);
16.         }
18.         try {
19.             //执行取消订单
20.             return executeCancelOrder(cancelOrderRequest, orderId);
21.         } catch (Exception e) {
22.             log.error("biz error", e);
23.             throw new OrderBizException(e.getMessage());
24.         } finally {
25.             redisLock.unlock(key);
26.         }
27.     }
29.     @Override
30.     public JsonResult<Boolean> executeCancelOrder(CancelOrderRequest cancelOrderRequest, String orderId) {
31.         //1.组装数据
32.         OrderInfoDO orderInfoDO = findOrderInfo(orderId, cancelOrderRequest.getCancelType());
33.         CancelOrderAssembleRequest cancelOrderAssembleRequest = buildAssembleRequest(orderId, cancelOrderRequest, orderInfoDO);
34.         if (cancelOrderAssembleRequest.getOrderInfoDTO().getOrderStatus() >= OrderStatusEnum.OUT_STOCK.getCode()) {
35.             throw new OrderBizException(OrderErrorCodeEnum.CURRENT_ORDER_STATUS_CANNOT_CANCEL);
36.         }
38.         TransactionMQProducer producer = defaultProducer.getProducer();
39.         producer.setTransactionListener(new TransactionListener() {
40.             @Override
41.             public LocalTransactionState executeLocalTransaction(Message message, Object o) {
42.                 try {
43.                     //2.执行履约取消、更新订单状态、新增订单日志操作
44.                     afterSaleManager.cancelOrderFulfillmentAndUpdateOrderStatus(cancelOrderAssembleRequest);
45.                     return LocalTransactionState.COMMIT_MESSAGE;
46.                 } catch (Exception e) {
47.                     log.error("system error", e);
48.                     return LocalTransactionState.ROLLBACK_MESSAGE;
49.                 }
50.             }
52.             @Override
53.             public LocalTransactionState checkLocalTransaction(MessageExt messageExt) {
54.                 //查询订单状态是否已更新为"已取消"
55.                 OrderInfoDO orderInfoByDatabase = orderInfoDAO.getByOrderId(orderId);
56.                 if (OrderStatusEnum.CANCELED.getCode().equals(orderInfoByDatabase.getOrderStatus())) {
57.                     return LocalTransactionState.COMMIT_MESSAGE;
58.                 }
59.                 return LocalTransactionState.ROLLBACK_MESSAGE;
60.             }
61.         });
63.         try {
64.             Message message = new Message(RocketMqConstant.RELEASE_ASSETS_TOPIC, JSONObject.toJSONString(cancelOrderAssembleRequest).getBytes(StandardCharsets.UTF_8));
65.             //3.发送事务消息 释放权益资产
66.             TransactionSendResult result = producer.sendMessageInTransaction(message, cancelOrderAssembleRequest);
67.             if (!result.getLocalTransactionState().equals(LocalTransactionState.COMMIT_MESSAGE)) {
68.                 throw new OrderBizException(OrderErrorCodeEnum.CANCEL_ORDER_PROCESS_FAILED);
69.             }
70.             return JsonResult.buildSuccess(true);
71.         } catch (Exception e) {
72.             throw new OrderBizException(OrderErrorCodeEnum.SEND_TRANSACTION_MQ_FAILED);
73.         }
74.     }
75.     ...
76. }

复制代码

(2)Seata事务AT模式的原理

1. 一.取消订单接口的入口会向Seata Server注册一个全局事务XID
2. 二.调用的履约服务会向Seata Server注册一个分支事务Branch ID
3. 三.在履约服务更新数据前，要先获取本地锁
4. 四.在履约服务在获取本地锁成功后，会插入undo log表数据
5. 五.然后，履约服务会向Seata Server服务器获取全局锁
6. 六.接着，履约服务才提交本地事务并释放本地锁
7. 七.最后，履约服务会向Seata Server上报分支事务成功

复制代码

 
14.取消订单全链路Seata回滚原理与并发分析
(1)回滚原理
(2)并发分析
 
(1)回滚原理
如果分支事务出现异常，就会上报Seata Server需要回滚全局事务。如果某些分支事务还没提交，那么直接就不需要执行了。如果某些分支事务已经提交成功，那么就根据undo log进行数据恢复。
 
(2)并发分析
拦截履约的各分支事务并不会出现并发竞争全局锁的问题。因为取消履约、取消出库、取消配送都是针对一个订单操作，同一时刻不会出现大量对一个订单的操作。
 
15.支付退款时的双异步设计原因(提升性能 + 解耦)
支付退款的时候会有两个消费者(双异步)：一个是退款准备消费者，一个是实际退款消费者。如果两个消费者合并成一个消费者，可能会出现两个问题。
 
问题一：退款速度可能会比较慢，因为一个消费者需要处理的事情多了。
 
问题二：如果第三方退款接口出现异常或报错，耦合比较严重特别不合理。这时计算好的数据和插入的记录，可能需要回滚或者重新消费消息。在故障情况下，第三方一直报错，那么系统就一直回滚，而且还会导致用户不能及时通过售后记录查看售后信息。

 
16.释放资产消息的高扩展性设计(多路发送消息)
订单系统的取消订单接口被调用时，会先发送一个释放资产的消息到MQ，然后订单系统会消费释放资产的消息。在消费释放资产的消息时，才发送具体释放哪些资产的消息到MQ，接着各个系统才能对各自监听的释放资产消息进行消费。
 
如果取消订单的核心业务代码，直接把具体的释放资产消息到MQ。那么当资产种类增加时，比如增加虚拟币、积分、权益等各种资产，就需要修改订单系统取消订单的核心业务代码了，扩展性会很差。
 
如果取消订单的核心业务代码，只发一个释放资产的消息到MQ。那么后续增加资产种类时，只需在消费释放资产的消息处进行改动即可，这样就可以大大提高了代码的扩展性。

 
17.取消订单链路中数据库事务与MQ消息不一致问题
(1)数据库事务与MQ消息不一致情况
(2)取消订单全链路数据不丢失的方案设计
 
(1)数据库事务与MQ消息不一致情况
在v1版本的取消订单接口中，由于"更新订单状态 + 取消履约 + 发送释放资产消息到MQ"属于刚性事务，所以其中任何一个操作出现Exception，都会进行全局回滚保证数据一致。但是更新数据库和发送消息到MQ，还是可能会出现数据不一致的情况。
 
比如代码顺序是：先发送释放资产消息，再更新订单状态，再取消履约。先发送释放资产消息到MQ成功，更新订单状态也成功，但取消履约失败。那么此时回滚只能回滚更新订单状态，发送到MQ的消息却不能回滚了。所以会存在这种不一致的风险，因此需要一种强一致的方案。
 
如下是v1版本的取消订单实现：

1. @Service
2. public class OrderAfterSaleServiceImpl implements OrderAfterSaleService {
3.     ...
4.     //取消订单/超时未支付取消
5.     @Override
6.     public JsonResult<Boolean> cancelOrder(CancelOrderRequest cancelOrderRequest) {
7.         //入参检查
8.         checkCancelOrderRequestParam(cancelOrderRequest);
9.         //分布式锁
10.         String orderId = cancelOrderRequest.getOrderId();
11.         String key = RedisLockKeyConstants.CANCEL_KEY + orderId;
13.         try {
14.             boolean lock = redisLock.lock(key);
15.             if (!lock) {
16.                 throw new OrderBizException(OrderErrorCodeEnum.CANCEL_ORDER_REPEAT);
17.             }
18.             //执行取消订单
19.             executeCancelOrder(cancelOrderRequest, orderId);
20.             return JsonResult.buildSuccess(true);
21.         } catch (Exception e) {
22.             throw new OrderBizException(e.getMessage());
23.         } finally {
24.             redisLock.unlock(key);
25.         }
26.     }
28.     @Override
29.     @GlobalTransactional(rollbackFor = Exception.class)
30.     public void executeCancelOrder(CancelOrderRequest cancelOrderRequest, String orderId) {
31.         //1.组装数据
32.         CancelOrderAssembleRequest cancelOrderAssembleRequest = buildAssembleRequest(orderId, cancelOrderRequest);
33.         //幂等校验：防止多个线程同时操作取消同一笔订单
34.         if (OrderStatusEnum.CANCELED.getCode().equals(cancelOrderAssembleRequest.getOrderInfoDTO().getOrderStatus())) {
35.             return;
36.         }
37.         //2.检验订单支付状态
38.         checkOrderPayStatus(cancelOrderAssembleRequest);
39.         //3.更新订单状态和记录订单操作日志
40.         updateOrderStatusAndSaveOperationLog(cancelOrderAssembleRequest);
41.         //超时未支付的订单不用继续再往下执行取消履约和释放资产
42.         if (OrderStatusEnum.PAID.getCode() > cancelOrderAssembleRequest.getOrderInfoDTO().getOrderStatus()) {
43.             return;
44.         }
45.         //4.履约取消
46.         cancelFulfill(cancelOrderAssembleRequest);
47.         //5.发送释放资产消息到MQ
48.         defaultProducer.sendMessage(RocketMqConstant.RELEASE_ASSETS_TOPIC, JSONObject.toJSONString(cancelOrderAssembleRequest), "释放资产");
49.     }
50.     ...
51. }

复制代码

(2)取消订单全链路数据不丢失的方案设计
为保证取消订单的链路中，数据库的更新事务与推送给MQ的消息强一致，需要将两者包裹在一个事务中，保证它们要么一起成功，要么一起失败。于是，就需要通过RocketMQ的事务机制来实现了。
 
如下是v2版本的取消订单实现，保证了数据库事务与MQ消息强一致。

1. @Service
2. public class OrderAfterSaleServiceImpl implements OrderAfterSaleService {
3.     ...
4.     //取消订单/超时未支付取消
5.     @Override
6.     @GlobalTransactional(rollbackFor = Exception.class)
7.     public JsonResult<Boolean> cancelOrder(CancelOrderRequest cancelOrderRequest) {
8.         //入参检查
9.         checkCancelOrderRequestParam(cancelOrderRequest);
10.         //分布式锁
11.         String orderId = cancelOrderRequest.getOrderId();
12.         String key = RedisLockKeyConstants.CANCEL_KEY + orderId;
13.         boolean lock = redisLock.lock(key);
14.         if (!lock) {
15.             throw new OrderBizException(OrderErrorCodeEnum.CANCEL_ORDER_REPEAT);
16.         }
17.         try {
18.             //执行取消订单
19.             return executeCancelOrder(cancelOrderRequest, orderId);
20.         } catch (Exception e) {
21.             log.error("biz error", e);
22.             throw new OrderBizException(e.getMessage());
23.         } finally {
24.             redisLock.unlock(key);
25.         }
26.     }
28.     @Override
29.     public JsonResult<Boolean> executeCancelOrder(CancelOrderRequest cancelOrderRequest, String orderId) {
30.         //1.组装数据
31.         OrderInfoDO orderInfoDO = findOrderInfo(orderId, cancelOrderRequest.getCancelType());
32.         CancelOrderAssembleRequest cancelOrderAssembleRequest = buildAssembleRequest(orderId, cancelOrderRequest, orderInfoDO);
33.         if (cancelOrderAssembleRequest.getOrderInfoDTO().getOrderStatus() >= OrderStatusEnum.OUT_STOCK.getCode()) {
34.             throw new OrderBizException(OrderErrorCodeEnum.CURRENT_ORDER_STATUS_CANNOT_CANCEL);
35.         }
37.         TransactionMQProducer producer = defaultProducer.getProducer();
38.         producer.setTransactionListener(new TransactionListener() {
39.             @Override
40.             public LocalTransactionState executeLocalTransaction(Message message, Object o) {
41.                 try {
42.                     //2.执行履约取消、更新订单状态、新增订单日志操作
43.                     afterSaleManager.cancelOrderFulfillmentAndUpdateOrderStatus(cancelOrderAssembleRequest);
44.                     return LocalTransactionState.COMMIT_MESSAGE;
45.                 } catch (Exception e) {
46.                     log.error("system error", e);
47.                     return LocalTransactionState.ROLLBACK_MESSAGE;
48.                 }
49.             }
51.             @Override
52.             public LocalTransactionState checkLocalTransaction(MessageExt messageExt) {
53.                 //查询订单状态是否已更新为"已取消"
54.                 OrderInfoDO orderInfoByDatabase = orderInfoDAO.getByOrderId(orderId);
55.                 if (OrderStatusEnum.CANCELED.getCode().equals(orderInfoByDatabase.getOrderStatus())) {
56.                     return LocalTransactionState.COMMIT_MESSAGE;
57.                 }
58.                 return LocalTransactionState.ROLLBACK_MESSAGE;
59.             }
60.         });
62.         try {
63.             Message message = new Message(RocketMqConstant.RELEASE_ASSETS_TOPIC, JSONObject.toJSONString(cancelOrderAssembleRequest).getBytes(StandardCharsets.UTF_8));
64.             //3.发送事务消息 释放权益资产
65.             TransactionSendResult result = producer.sendMessageInTransaction(message, cancelOrderAssembleRequest);
66.             if (!result.getLocalTransactionState().equals(LocalTransactionState.COMMIT_MESSAGE)) {
67.                 throw new OrderBizException(OrderErrorCodeEnum.CANCEL_ORDER_PROCESS_FAILED);
68.             }
69.             return JsonResult.buildSuccess(true);
70.         } catch (Exception e) {
71.             throw new OrderBizException(OrderErrorCodeEnum.SEND_TRANSACTION_MQ_FAILED);
72.         }
73.     }
74.     ...
75. }

复制代码

 
18.释放资产多路MQ故障重试与幂等方案
如果发送消息到MQ过程中出现故障，那么就会通过返回RECONSUME_LATER来进行重试。
 
此外，消费消息时调用的接口可能会出现同一请求多次被调用。因此，必须对调用的接口，使用分布式锁 + 状态前置校验进行幂等处理。加分布式锁是为了防止并非请求进来后可能会避开状态的前置校验。

1. @Configuration
2. public class ConsumerConfig {
3.     @Autowired
4.     private RocketMQProperties rocketMQProperties;
5.     ...
7.     //释放资产消息消费者
8.     @Bean("releaseAssetsConsumer")
9.     public DefaultMQPushConsumer releaseAssetsConsumer(ReleaseAssetsListener releaseAssetsListener) throws MQClientException {
10.         DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(RELEASE_ASSETS_CONSUMER_GROUP);
11.         consumer.setNamesrvAddr(rocketMQProperties.getNameServer());
12.         consumer.subscribe(RELEASE_ASSETS_TOPIC, "*");
13.         consumer.registerMessageListener(releaseAssetsListener);
14.         consumer.start();
15.         return consumer;
16.     }
17. }
19. //监听并消费释放资产消息
20. @Component
21. public class ReleaseAssetsListener implements MessageListenerConcurrently {
22.     @Autowired
23.     private DefaultProducer defaultProducer;
25.     @Autowired
26.     private OrderItemDAO orderItemDAO;
28.     @Override
29.     public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
30.         try {
31.             for (MessageExt messageExt : list) {
32.                 //1.消费到释放资产message
33.                 String message = new String(messageExt.getBody());
34.                 log.info("ReleaseAssetsListener message:{}", message);
35.                 CancelOrderAssembleRequest cancelOrderAssembleRequest = JSONObject.parseObject(message, CancelOrderAssembleRequest.class);
36.                 OrderInfoDTO orderInfoDTO = cancelOrderAssembleRequest.getOrderInfoDTO();
37.                 //2.发送取消订单退款请求MQ
38.                 if (orderInfoDTO.getOrderStatus() > OrderStatusEnum.CREATED.getCode()) {
39.                     defaultProducer.sendMessage(RocketMqConstant.CANCEL_REFUND_REQUEST_TOPIC,
40.                         JSONObject.toJSONString(cancelOrderAssembleRequest), "取消订单退款");
41.                 }
42.                 //3.发送释放库存MQ
43.                 ReleaseProductStockRequest releaseProductStockRequest = buildReleaseProductStock(orderInfoDTO, orderItemDAO);
44.                 defaultProducer.sendMessage(RocketMqConstant.CANCEL_RELEASE_INVENTORY_TOPIC,
45.                     JSONObject.toJSONString(releaseProductStockRequest), "取消订单释放库存");
46.                 //4.发送释放优惠券MQ
47.                 if (!Strings.isNullOrEmpty(orderInfoDTO.getCouponId())) {
48.                     ReleaseUserCouponRequest releaseUserCouponRequest = buildReleaseUserCoupon(orderInfoDTO);
49.                     defaultProducer.sendMessage(RocketMqConstant.CANCEL_RELEASE_PROPERTY_TOPIC,
50.                         JSONObject.toJSONString(releaseUserCouponRequest), "取消订单释放优惠券");
51.                 }
52.             }
53.             return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
54.         } catch (Exception e) {
55.             log.error("consumer error", e);
56.             return ConsumeConcurrentlyStatus.RECONSUME_LATER;
57.         }
58.     }
59.     ...
60. }
62. @Configuration
63. public class ConsumerConfig {
64.     @Autowired
65.     private RocketMQProperties rocketMQProperties;
67.     //释放优惠券消息消费者
68.     @Bean("releaseInventoryConsumer")
69.     public DefaultMQPushConsumer releaseInventoryConsumer(ReleasePropertyListener releasePropertyListener) throws MQClientException {
70.         DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(RocketMqConstant.RELEASE_PROPERTY_CONSUMER_GROUP);
71.         consumer.setNamesrvAddr(rocketMQProperties.getNameServer());
72.         consumer.subscribe(RocketMqConstant.CANCEL_RELEASE_PROPERTY_TOPIC, "*");
73.         consumer.registerMessageListener(releasePropertyListener);
74.         consumer.start();
75.         return consumer;
76.     }
77. }
79. @Component
80. public class ReleasePropertyListener implements MessageListenerConcurrently {
81.     @DubboReference(version = "1.0.0")
82.     private MarketApi marketApi;
84.     @Override
85.     public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
86.         try {
87.             for (MessageExt msg : list) {
88.                 String content = new String(msg.getBody(), StandardCharsets.UTF_8);
89.                 log.info("ReleasePropertyConsumer message:{}", content);
90.                 ReleaseUserCouponRequest releaseUserCouponRequest = JSONObject.parseObject(content, ReleaseUserCouponRequest.class);
91.                 //释放优惠券
92.                 JsonResult<Boolean> jsonResult = marketApi.releaseUserCoupon(releaseUserCouponRequest);
93.                 if (!jsonResult.getSuccess()) {
94.                     throw new MarketBizException(MarketErrorCodeEnum.CONSUME_MQ_FAILED);
95.                 }
96.             }
97.             return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
98.         } catch (Exception e) {
99.             log.error("consumer error", e);
100.             return ConsumeConcurrentlyStatus.RECONSUME_LATER;
101.         }
102.     }
103. }
105. @DubboService(version = "1.0.0", interfaceClass = MarketApi.class, retries = 0)
106. public class MarketApiImpl implements MarketApi {
107.     ...
108.     //回退用户使用的优惠券
109.     @Override
110.     public JsonResult<Boolean> releaseUserCoupon(ReleaseUserCouponRequest releaseUserCouponRequest) {
111.         log.info("开始执行回滚优惠券,couponId:{}", releaseUserCouponRequest.getCouponId());
112.         //分布式锁
113.         String couponId = releaseUserCouponRequest.getCouponId();
114.         String key = RedisLockKeyConstants.RELEASE_COUPON_KEY + couponId;
115.         boolean lock = redisLock.lock(key);
116.         if (!lock) {
117.             throw new MarketBizException(MarketErrorCodeEnum.RELEASE_COUPON_FAILED);
118.         }
119.         try {
120.             //执行释放优惠券
121.             Boolean result = couponService.releaseUserCoupon(releaseUserCouponRequest);
122.             return JsonResult.buildSuccess(result);
123.         } catch (MarketBizException e) {
124.             log.error("biz error", e);
125.             return JsonResult.buildError(e.getErrorCode(), e.getErrorMsg());
126.         } catch (Exception e) {
127.             log.error("system error", e);
128.             return JsonResult.buildError(e.getMessage());
129.         } finally {
130.             redisLock.unlock(key);
131.         }
132.     }
133.     ...
134. }
136. @Service
137. public class CouponServiceImpl implements CouponService {
138.     ...
139.     //释放用户优惠券
140.     @Override
141.     public Boolean releaseUserCoupon(ReleaseUserCouponRequest releaseUserCouponRequest) {
142.         String userId = releaseUserCouponRequest.getUserId();
143.         String couponId = releaseUserCouponRequest.getCouponId();
144.         CouponDO couponAchieve = couponDAO.getUserCoupon(userId, couponId);
145.         if (CouponUsedStatusEnum.UN_USED.getCode().equals(couponAchieve.getUsed())) {
146.             log.info("当前用户未使用优惠券,不用回退,userId:{},couponId:{}", userId, couponId);
147.             return true;
148.         }
149.         couponAchieve.setUsed(CouponUsedStatusEnum.UN_USED.getCode());
150.         couponAchieve.setUsedTime(null);
151.         couponDAO.updateById(couponAchieve);
152.         return true;
153.     }
154.     ...
155. }

复制代码

 
19.双异步支付退款不一致问题分析
在退款准备消费者中，如果先写售后记录到数据库，再发退款消息到MQ。如果写入售后记录到数据库成功了，但发送实际退款消息到MQ却失败了，那么此时就会产生数据库与MQ的数据不一致问题。
 
在退款准备消费者中，如果先发退款消息到MQ，再写售后记录到数据库。如果发送退款消息到MQ成功了，但写入售后记录到数据库却失败了，那么同样会产生数据库与MQ的数据不一致问题。

此时就可以使用RocketMQ的事务消息机制；

1. @Configuration
2. public class ConsumerConfig {
3.     @Autowired
4.     private RocketMQProperties rocketMQProperties;
5.     ...
7.     //消费退款准备请求消息消费者
8.     @Bean("cancelRefundConsumer")
9.     public DefaultMQPushConsumer cancelRefundConsumer(CancelRefundListener cancelRefundListener) throws MQClientException {
10.         DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(RocketMqConstant.REQUEST_CONSUMER_GROUP);
11.         consumer.setNamesrvAddr(rocketMQProperties.getNameServer());
12.         consumer.subscribe(RocketMqConstant.CANCEL_REFUND_REQUEST_TOPIC, "*");
13.         consumer.registerMessageListener(cancelRefundListener);
14.         consumer.start();
15.         return consumer;
16.     }
17.     ...
18. }
20. //消费退款准备请求消息
21. @Component
22. public class CancelRefundListener implements MessageListenerConcurrently {
23.     @Autowired
24.     private OrderAfterSaleService orderAfterSaleService;
26.     @Override
27.     public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
28.         try {
29.             for (MessageExt messageExt : list) {
30.                 String message = new String(messageExt.getBody());
31.                 CancelOrderAssembleRequest cancelOrderAssembleRequest = JSONObject.parseObject(message, CancelOrderAssembleRequest.class);
32.                 log.info("CancelRefundConsumer message:{}", message);
33.                 //执行 取消订单/超时未支付取消 前的操作
34.                 JsonResult<Boolean> jsonResult = orderAfterSaleService.processCancelOrder(cancelOrderAssembleRequest);
35.                 if (!jsonResult.getSuccess()) {
36.                     throw new OrderBizException(OrderErrorCodeEnum.CONSUME_MQ_FAILED);
37.                 }
38.             }
39.             return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
40.         } catch (Exception e) {
41.             log.error("consumer error", e);
42.             return ConsumeConcurrentlyStatus.RECONSUME_LATER;
43.         }
44.     }
45. }
47. @Service
48. public class OrderAfterSaleServiceImpl implements OrderAfterSaleService {
49.     ...
50.     @Override
51.     public JsonResult<Boolean> processCancelOrder(CancelOrderAssembleRequest cancelOrderAssembleRequest) {
52.         String orderId = cancelOrderAssembleRequest.getOrderId();
53.         //分布式锁
54.         String key = RedisLockKeyConstants.REFUND_KEY + orderId;
55.         try {
56.             boolean lock = redisLock.lock(key);
57.             if (!lock) {
58.                 throw new OrderBizException(OrderErrorCodeEnum.PROCESS_REFUND_REPEAT);
59.             }
61.             //执行退款前的准备工作
62.             //生成售后订单号
63.             OrderInfoDTO orderInfoDTO = cancelOrderAssembleRequest.getOrderInfoDTO();
64.             OrderInfoDO orderInfoDO = orderInfoDTO.clone(OrderInfoDO.class);
65.             String afterSaleId = orderNoManager.genOrderId(OrderNoTypeEnum.AFTER_SALE.getCode(), orderInfoDO.getUserId());
67.             //1.计算 取消订单 退款金额
68.             CancelOrderRefundAmountDTO cancelOrderRefundAmountDTO = calculatingCancelOrderRefundAmount(cancelOrderAssembleRequest);
69.             cancelOrderAssembleRequest.setCancelOrderRefundAmountDTO(cancelOrderRefundAmountDTO);
70.             TransactionMQProducer producer = defaultProducer.getProducer();
71.             producer.setTransactionListener(new TransactionListener() {
72.                 @Override
73.                 public LocalTransactionState executeLocalTransaction(Message message, Object o) {
74.                     try {
75.                         //2.取消订单操作 记录售后信息
76.                         afterSaleManager.insertCancelOrderAfterSale(cancelOrderAssembleRequest, AfterSaleStatusEnum.REVIEW_PASS.getCode(), orderInfoDO, afterSaleId);
77.                         return LocalTransactionState.COMMIT_MESSAGE;
78.                     } catch (Exception e) {
79.                         log.error("system error", e);
80.                         return LocalTransactionState.ROLLBACK_MESSAGE;
81.                     }
82.                 }
84.                 @Override
85.                 public LocalTransactionState checkLocalTransaction(MessageExt messageExt) {
86.                     //查询售后数据是否插入成功
87.                     AfterSaleInfoDO afterSaleInfoDO = afterSaleInfoDAO.getOneByAfterSaleId(Long.valueOf(afterSaleId));
88.                     List afterSaleItemDOList = afterSaleItemDAO.listByAfterSaleId(Long.valueOf(afterSaleId));
89.                     List afterSaleLogDOList = afterSaleLogDAO.listByAfterSaleId(Long.valueOf(afterSaleId));
90.                     List afterSaleRefundDOList = afterSaleRefundDAO.listByAfterSaleId(Long.valueOf(afterSaleId));
91.                     if (afterSaleInfoDO != null && afterSaleItemDOList.isEmpty() && afterSaleLogDOList.isEmpty() && afterSaleRefundDOList.isEmpty()) {
92.                         return LocalTransactionState.COMMIT_MESSAGE;
93.                     }
94.                     return LocalTransactionState.ROLLBACK_MESSAGE;
95.                 }
96.             });
98.             try {
99.                 //3.组装事务MQ消息
100.                 ActualRefundMessage actualRefundMessage = new ActualRefundMessage();
101.                 actualRefundMessage.setOrderId(cancelOrderAssembleRequest.getOrderId());
102.                 actualRefundMessage.setLastReturnGoods(cancelOrderAssembleRequest.isLastReturnGoods());
103.                 actualRefundMessage.setAfterSaleId(Long.valueOf(afterSaleId));
104.                 Message message = new Message(RocketMqConstant.ACTUAL_REFUND_TOPIC, JSONObject.toJSONString(actualRefundMessage).getBytes(StandardCharsets.UTF_8));
105.                 //4.发送事务MQ消息--实际退款消息
106.                 TransactionSendResult result = producer.sendMessageInTransaction(message, actualRefundMessage);
107.                 if (!result.getLocalTransactionState().equals(LocalTransactionState.COMMIT_MESSAGE)) {
108.                     throw new OrderBizException(OrderErrorCodeEnum.PROCESS_REFUND_FAILED);
109.                 }
110.                 return JsonResult.buildSuccess(true);
111.             } catch (Exception e) {
112.                 throw new OrderBizException(OrderErrorCodeEnum.SEND_TRANSACTION_MQ_FAILED);
113.             }
114.         } finally {
115.             redisLock.unlock(key);
116.         }
117.     }
118.     ...
119. }

复制代码

 

来源：程序园用户自行投稿发布，如果侵权，请联系站长删除
免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！