分布式事务的最终解决

通过MQ的延时消息来解决订单/库存服务直接调用的事务问题

一、如何解决下单系统这个高并发里边的分布式事务呢？

首先，我们肯定不会用 2PC 模式、 TCC-事务补偿性方案，我们也不考虑

最终我们选择可靠消息+最终一致性这种方式

为了保证高并发，订单这一块还是自己回滚，

因为性能不佳，我们弃用seata的方案

---

1、库存服务自己怎么回滚？

• 第一种

通过MQ消息队列，我们在提交订单那里，当捕捉到异常要回滚的时候，给库存服务发一个消息，让库存服务自己把库存解锁

这样不需要让库存事务回滚，只需要给它发一个消息，不会损失什么性能

• 第二种

通过MQ延迟消息

库存服务本身也可以使用自动解锁模式。

怎么自动解锁呢？

需要使用消息队列来完成。

如果你想让我这的哪些库存解锁，首先你需要给我发一个消息告诉我。

然后我们专门的库存解锁服务，去来订阅我们stock.release.stock.queue这个队列里的消息。

那你给我发消息的时候，比如：用路由键stock.release，我知道要库存解锁，

然后，你的消息发给我们这个交换机stock-event-exchange。

交换机把这个消息路由给stock.release.stock.queue这个队列。

然后，这个队列stock.release.stock.queue里边存的这些消息都是库存要解锁的消息，我们的库存解锁服务只要收到了，它就会在后台慢慢的解锁消息。

我们不用保证强一致，我们哪怕是二十分钟、三十分钟，乃至于一天以后把这个库存解锁了，最终一致了就行。

所以我们可以来使用消息队列来完成我们的这个最终一致性。

---

2、锁库存的增强版逻辑

我们想要锁库存的话，我们先来保存一个库存工作单和库存工作单详情

相当于只要我们想要锁库存，我们先给数据库里边保存记录，我要锁库存。

接下来我们就来进行锁，只要锁成功了，那一切ok。

如果锁失败了，数据库里边相当于没有这个锁库存记录。

因为锁失败呢，我们这个本身自己所失败会全部回滚。

但如果可能是这种失败，比如我们来到订单里边，我们库存其实自己锁成功了。但是我们订单下边的其他完了，然后库存要进行解锁。那怎么办呢？

我们可以使用定时任务

---

二、订单服务的完整消息队列

1、库存自动解锁

库存微服务，有一个它的库存交换机stock-event-exchange.

如果想要解锁库存，应该是这样的。

首先订单创建成功之后，库存锁定成功，然后发一个消息给交换机，

这个消息里面的内容有订单编号、仓库编号、哪个商品锁了几个库存，

这个交换机，绑定了两个队列，

一个是按照stock.release.#模糊匹配的路由键绑定的stock.release.stock.queue队列

一个是stock.delay.queue队列

第一次发的库存锁定成功的消息，先使用路由键叫stock.locked

交换机按照这个路由键，找到stock.delay.queue延时队列

延时队列50分钟以后，用stock.release这个路由键，将死信交给库存交换机stock-event-exchange，

交换机收到以后，按照这个路由键查找，发现stock.release.#这个模糊匹配的路由键跟它是一样的，然后被交换机路由到我们这个stock.release.stock.queue队列。

接下来的解锁库存服务，专门来处理stock.release.stock.queue里的消息。

• 最终实现

• com.achang.achangmall.order.conf.MQConfig

通过配置类来创建队列、绑定、交换机

@Configuration
public class MQConfig {

    @Bean
    public Queue orderDelayQueue(){//死信队列
        HashMap<String, Object> argument = new HashMap<>();
        argument.put("x-dead-letter-exchange","order-event-exchange");//死信路由
        argument.put("x-dead-letter-routing-key","order.release.order");//死信路由键
        argument.put("x-message-ttl",60000);//消息过期时间 1分钟

        return new Queue("order.delay.queue",true,false,false,argument);
    }

    @Bean
    public Queue orderReleaseOrderQueue(){//普通队列
        return new Queue("order.release.order.queue",true,false,false,null);
    }

    @Bean
    public Exchange orderEventExchange(){//普通的主题交换机（可绑定多个队列，根据条件路由）
        return new TopicExchange("order-event-exchange",true,false);
    }

    @Bean
    public Binding orderCreateOrderBinding(){
        return new Binding("order.delay.queue",
                           Binding.DestinationType.QUEUE,
                           "order-event-exchange",
                           "order.create.order",null);
    }

    @Bean
    public Binding orderReleaseOrderBinding(){
        return new Binding("order.release.order.queue",
                           Binding.DestinationType.QUEUE,
                           "order-event-exchange",
                           "order.release.order",null);
    }
}

• 库存服务加mq依赖

achangmall-ware/pom.xml

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-amqp</artifactId>
</dependency>

• rabbitmq配置

achangmall-ware/src/main/resources/application.yaml

spring:
  rabbitmq:
    host: 192.168.109.101
    port: 5672
    virtual-host: /
    listener:
      simple:
        acknowledge-mode: manual

• 主启动类启动MQ

achangmall-ware/src/main/java/com/achang/achangmall/AchangmallWareApplication.java

@EnableRabbit

• 配置类

com.achang.achangmall.ware.conf.RabbitMQConfig

@Configuration
public class RabbitMQConfig {

    @Bean
    public MessageConverter messageConversionException(){
        return new Jackson2JsonMessageConverter();
    }

    @Bean
    public Exchange stockEventExchange(){
        return new TopicExchange("stock-event-exchange",true,false);
    }

    @Bean
    public Queue stockReleaseStockQueue(){
        return new Queue("stock.release.stock.queue",true,false,false);
    }

    @Bean
    public Queue stockDelayQueue(){
        HashMap<String, Object> argument = new HashMap<>();
        argument.put("x-dead-letter-exchange","stock-event-exchange");//死信路由
        argument.put("x-dead-letter-routing-key","stock.release");//死信路由键
        argument.put("x-message-ttl", 120000L);//消息过期时间 1分钟
        return new Queue("stock.delay.queue",true,false,false,argument);
    }

    @Bean
    public Binding stockReleaseBinding(){
        return new Binding("stock.release.stock.queue",
                           Binding.DestinationType.QUEUE,
                           "stock-event-exchange",
                           "stock.release.#",
                           null);
    }

    @Bean
    public Binding stockLockedBinding(){
        return new Binding("stock.delay.queue",
                           Binding.DestinationType.QUEUE,
                           "stock-event-exchange",
                           "stock.locked",
                           null);
    }
}

• com.achang.achangmall.ware.controller.WareSkuController

/**
     * 锁定库存
     * @param vo
     *
     * 库存解锁的场景
     *      1）、下订单成功，订单过期没有支付被系统自动取消或者被用户手动取消，都要解锁库存
     *      2）、下订单成功，库存锁定成功，接下来的业务调用失败，导致订单回滚。之前锁定的库存就要自动解锁
     *      3）、	
     */
@PostMapping(value = "/lock/order")
public R orderLockStock(@RequestBody WareSkuLockVo vo) {

    //        try {
    boolean lockStock = wareSkuService.orderLockStock(vo);
    return R.ok().setData(lockStock);
    //        } catch (NoStockException e) {
    //            return R.error(NO_STOCK_EXCEPTION.getCode(),NO_STOCK_EXCEPTION.getMessage());
    //        }
}

• 发送到mq消息队列的to

com.achang.common.to.mq.StockLockedTo

/**
 * @Description: 发送到mq消息队列的to
 **/
@Data
public class StockLockedTo {

    /** 库存工作单的id **/
    private Long id;

    /** 工作单详情的所有信息 **/
    private StockDetailTo detailTo;
}

• com.achang.common.to.mq.StockDetailTo

@Data
public class StockDetailTo {

    private Long id;
    /**
     * sku_id
     */
    private Long skuId;
    /**
     * sku_name
     */
    private String skuName;
    /**
     * 购买个数
     */
    private Integer skuNum;
    /**
     * 工作单id
     */
    private Long taskId;

    /**
     * 仓库id
     */
    private Long wareId;

    /**
     * 锁定状态
     */
    private Integer lockStatus;

}

• com.achang.achangmall.ware.service.impl.WareSkuServiceImpl

@Service("wareSkuService")
public class WareSkuServiceImpl extends ServiceImpl<WareSkuDao, WareSkuEntity> implements WareSkuService {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    @Autowired
    private WareSkuDao wareSkuDao;

    @Autowired
    private OrderFeignService orderFeignService;

    @Autowired
    private WareOrderTaskDetailService wareOrderTaskDetailService;

    @Override
    public PageUtils queryPage(Map<String, Object> params) {
        QueryWrapper<WareSkuEntity> wrapper = new QueryWrapper<>();

        String skuId = (String) params.get("skuId");
        if (!StringUtils.isEmpty(skuId)){
            wrapper.eq("sku_id",skuId);
        }
        String wareId = (String) params.get("wareId");
        if (!StringUtils.isEmpty(wareId)){

        }


        IPage<WareSkuEntity> page = this.page(
            new Query<WareSkuEntity>().getPage(params),
            wrapper
        );

        return new PageUtils(page);
    }

    @Override
    public List<SkuHasStockVo> getSkuHasStock(List<Long> skuIdList) {
        List<SkuHasStockVo> collect = skuIdList.stream().map(item -> {
            SkuHasStockVo vo = new SkuHasStockVo();
            Long count = baseMapper.getSkuStock(item);
            vo.setSkuId(item);
            vo.setHasStock(count == null?false:count>0);
            return vo;
        }).collect(Collectors.toList());
        return collect;
    }

    @Autowired
    private WareOrderTaskService wareOrderTaskService;

    @Override
    public boolean orderLockStock(WareSkuLockVo vo) {

        /**
         * 保存库存工作单详情信息
         * 追溯
         */
        WareOrderTaskEntity wareOrderTaskEntity = new WareOrderTaskEntity();
        wareOrderTaskEntity.setOrderSn(vo.getOrderSn());
        wareOrderTaskEntity.setCreateTime(new Date());
        wareOrderTaskService.save(wareOrderTaskEntity);


        //1、按照下单的收货地址，找到一个就近仓库，锁定库存
        //2、找到每个商品在哪个仓库都有库存
        List<OrderItemVo> locks = vo.getLocks();

        List<SkuWareHasStock> collect = locks.stream().map((item) -> {
            SkuWareHasStock stock = new SkuWareHasStock();
            Long skuId = item.getSkuId();
            stock.setSkuId(skuId);
            stock.setNum(item.getCount());
            //查询这个商品在哪个仓库有库存
            List<Long> wareIdList = wareSkuDao.listWareIdHasSkuStock(skuId);
            stock.setWareId(wareIdList);

            return stock;
        }).collect(Collectors.toList());

        //2、锁定库存
        for (SkuWareHasStock hasStock : collect) {
            boolean skuStocked = false;
            Long skuId = hasStock.getSkuId();
            List<Long> wareIds = hasStock.getWareId();

            if (org.springframework.util.StringUtils.isEmpty(wareIds)) {
                //没有任何仓库有这个商品的库存
                //                throw new NoStockException(skuId);
            }

            //1、如果每一个商品都锁定成功,将当前商品锁定了几件的工作单记录发给MQ
            //2、锁定失败。前面保存的工作单信息都回滚了。发送出去的消息，即使要解锁库存，由于在数据库查不到指定的id，所有就不用解锁
            for (Long wareId : wareIds) {
                //锁定成功就返回1，失败就返回0
                Long count = wareSkuDao.lockSkuStock(skuId,wareId,hasStock.getNum());
                if (count == 1) {
                    skuStocked = true;
                    WareOrderTaskDetailEntity taskDetailEntity = new WareOrderTaskDetailEntity();
                    taskDetailEntity.setSkuId(skuId);
                    taskDetailEntity.setSkuName("");
                    taskDetailEntity.setSkuNum(hasStock.getNum());
                    taskDetailEntity.setTaskId(wareOrderTaskEntity.getId());
                    taskDetailEntity.setWareId(wareId);
                    taskDetailEntity.setLockStatus(1);

                    wareOrderTaskDetailService.save(taskDetailEntity);

                    //TODO 告诉MQ库存锁定成功
                    StockLockedTo lockedTo = new StockLockedTo();
                    lockedTo.setId(wareOrderTaskEntity.getId());
                    StockDetailTo detailTo = new StockDetailTo();
                    BeanUtils.copyProperties(taskDetailEntity,detailTo);
                    lockedTo.setDetailTo(detailTo);
                    rabbitTemplate.convertAndSend("stock-event-exchange","stock.locked",lockedTo);
                    break;
                } else {
                    //当前仓库锁失败，重试下一个仓库
                }
            }

            if (skuStocked == false) {
                //当前商品所有仓库都没有锁住
                throw new NoStockException(skuId);
            }
        }

        //3、肯定全部都是锁定成功的
        return true;
    }

    /**
     * 解锁库存的方法
     * @param skuId
     * @param wareId
     * @param num
     * @param taskDetailId
     */
    public void unLockStock(Long skuId,Long wareId,Integer num,Long taskDetailId) {

        //库存解锁
        wareSkuDao.unLockStock(skuId,wareId,num);

        //更新工作单的状态
        WareOrderTaskDetailEntity taskDetailEntity = new WareOrderTaskDetailEntity();
        taskDetailEntity.setId(taskDetailId);
        //变为已解锁
        taskDetailEntity.setLockStatus(2);
        wareOrderTaskDetailService.updateById(taskDetailEntity);

    }


    @Data
    class SkuWareHasStock {
        private Long skuId;
        private Integer num;
        private List<Long> wareId;
    }

    /**
     * 防止订单服务卡顿，导致订单状态消息一直改不了，库存优先到期，查订单状态新建，什么都不处理
     * 导致卡顿的订单，永远都不能解锁库存
     * @param orderTo
     */
    @Transactional(rollbackFor = Exception.class)
    @Override
    public void unlockStock(OrderTo orderTo) {

        String orderSn = orderTo.getOrderSn();
        //查一下最新的库存解锁状态，防止重复解锁库存
        WareOrderTaskEntity orderTaskEntity = wareOrderTaskService.getOrderTaskByOrderSn(orderSn);

        //按照工作单的id找到所有 没有解锁的库存，进行解锁
        Long id = orderTaskEntity.getId();
        List<WareOrderTaskDetailEntity> list = wareOrderTaskDetailService.list(new QueryWrapper<WareOrderTaskDetailEntity>()
                                                                               .eq("task_id", id).eq("lock_status", 1));

        for (WareOrderTaskDetailEntity taskDetailEntity : list) {
            unLockStock(taskDetailEntity.getSkuId(),
                        taskDetailEntity.getWareId(),
                        taskDetailEntity.getSkuNum(),
                        taskDetailEntity.getId());
        }

    }


    @Override
    public void unlockStock(StockLockedTo to) {
        //库存工作单的id
        StockDetailTo detail = to.getDetailTo();
        Long detailId = detail.getId();

        /**
         * 解锁
         * 1、查询数据库关于这个订单锁定库存信息
         *   有：证明库存锁定成功了
         *      解锁：订单状况
         *          1、没有这个订单，必须解锁库存
         *          2、有这个订单，不一定解锁库存
         *              订单状态：已取消：解锁库存
         *                      已支付：不能解锁库存
         */
        WareOrderTaskDetailEntity taskDetailInfo = wareOrderTaskDetailService.getById(detailId);
        if (taskDetailInfo != null) {
            //查出wms_ware_order_task工作单的信息
            Long id = to.getId();
            WareOrderTaskEntity orderTaskInfo = wareOrderTaskService.getById(id);
            //获取订单号查询订单状态
            String orderSn = orderTaskInfo.getOrderSn();
            //远程查询订单信息
            R orderData = orderFeignService.getOrderStatus(orderSn);
            if (orderData.getCode() == 0) {
                //订单数据返回成功
                OrderVo orderInfo = orderData.getData("data", new TypeReference<OrderVo>() {});

                //判断订单状态是否已取消或者支付或者订单不存在
                if (orderInfo == null || orderInfo.getStatus() == 4) {
                    //订单已被取消，才能解锁库存
                    if (taskDetailInfo.getLockStatus() == 1) {
                        //当前库存工作单详情状态1，已锁定，但是未解锁才可以解锁
                        unLockStock(detail.getSkuId(),detail.getWareId(),detail.getSkuNum(),detailId);
                    }
                }
            } else {
                //消息拒绝以后重新放在队列里面，让别人继续消费解锁
                //远程调用服务失败
                throw new RuntimeException("远程调用服务失败");
            }
        } else {
            //无需解锁
        }
    }

}

• achangmall-ware/src/main/resources/mapper/ware/WareSkuDao.xml

<update id="unLockStock">
    UPDATE wms_ware_sku
    SET stock_locked = stock_locked - #{num}
    WHERE
    sku_id = ${skuId}
    AND ware_id = #{wareId}
</update>

• com.achang.achangmall.ware.service.impl.WareOrderTaskServiceImpl

@Override
public WareOrderTaskEntity getOrderTaskByOrderSn(String orderSn) {
    WareOrderTaskEntity orderTaskEntity = this.baseMapper.selectOne(
        new QueryWrapper<WareOrderTaskEntity>().eq("order_sn", orderSn));
    return orderTaskEntity;
}

• com.achang.achangmall.ware.listener.StockReleaseListener

@Slf4j
@RabbitListener(queues = "stock.release.stock.queue")
@Service
public class StockReleaseListener {

    @Autowired
    private WareSkuService wareSkuService;

    /**
     * 1、库存自动解锁
     *  下订单成功，库存锁定成功，接下来的业务调用失败，导致订单回滚。之前锁定的库存就要自动解锁
     *
     *  2、订单失败
     *      库存锁定失败
     *
     *   只要解锁库存的消息失败，一定要告诉服务解锁失败
     */
    @RabbitHandler
    public void handleStockLockedRelease(StockLockedTo to, Message message, Channel channel) throws IOException {
        log.info("******收到解锁库存的信息******");
        try {

            //当前消息是否被第二次及以后（重新）派发过来了
            // Boolean redelivered = message.getMessageProperties().getRedelivered();

            //解锁库存
            wareSkuService.unlockStock(to);
            // 手动删除消息
            channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
        } catch (Exception e) {
            // 解锁失败 将消息重新放回队列，让别人消费
            channel.basicReject(message.getMessageProperties().getDeliveryTag(),true);
        }
    }

    @RabbitHandler
    public void handleOrderCloseRelease(OrderTo orderTo, Message message, Channel channel) throws IOException {

        log.info("******收到订单关闭，准备解锁库存的信息******");

        try {
            wareSkuService.unlockStock(orderTo);
            // 手动删除消息
            channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
        } catch (Exception e) {
            // 解锁失败 将消息重新放回队列，让别人消费
            channel.basicReject(message.getMessageProperties().getDeliveryTag(),true);
        }
    }

}

• com.achang.achangmall.order.service.impl.OrderServiceImpl

        /**
         * 按照订单号获取订单信息
         * @param orderSn
         * @return
         */
        @Override
        public OrderEntity getOrderByOrderSn(String orderSn) {
            OrderEntity orderEntity = this.baseMapper.selectOne(new QueryWrapper<OrderEntity>().eq("order_sn", orderSn));
            return orderEntity;
        }

• com.achang.achangmall.order.controller.OrderController

/**
* 根据订单编号查询订单状态
*/
@GetMapping(value = "/status/{orderSn}")
public R getOrderStatus(@PathVariable("orderSn") String orderSn) {
    OrderEntity orderEntity = orderService.getOrderByOrderSn(orderSn);
    return R.ok().setData(orderEntity);
}

• com.achang.achangmall.ware.feign.OrderFeignService

    @FeignClient("achangmall-order")
    public interface OrderFeignService {

        @GetMapping(value = "/order/order/status/{orderSn}")
        R getOrderStatus(@PathVariable("orderSn") String orderSn);

    }

• com.achang.achangmall.ware.vo.OrderVo

@Data
public class OrderVo {

    private Long id;
    /**
     * member_id
     */
    private Long memberId;
    /**
     * 订单号
     */
    private String orderSn;
    /**
     * 使用的优惠券
     */
    private Long couponId;
    /**
     * create_time
     */
    private Date createTime;
    /**
     * 用户名
     */
    private String memberUsername;
    /**
     * 订单总额
     */
    private BigDecimal totalAmount;
    /**
     * 应付总额
     */
    private BigDecimal payAmount;
    /**
     * 运费金额
     */
    private BigDecimal freightAmount;
    /**
     * 促销优化金额（促销价、满减、阶梯价）
     */
    private BigDecimal promotionAmount;
    /**
     * 积分抵扣金额
     */
    private BigDecimal integrationAmount;
    /**
     * 优惠券抵扣金额
     */
    private BigDecimal couponAmount;
    /**
     * 后台调整订单使用的折扣金额
     */
    private BigDecimal discountAmount;
    /**
     * 支付方式【1->支付宝；2->微信；3->银联； 4->货到付款；】
     */
    private Integer payType;
    /**
     * 订单来源[0->PC订单；1->app订单]
     */
    private Integer sourceType;
    /**
     * 订单状态【0->待付款；1->待发货；2->已发货；3->已完成；4->已关闭；5->无效订单】
     */
    private Integer status;
    /**
     * 物流公司(配送方式)
     */
    private String deliveryCompany;
    /**
     * 物流单号
     */
    private String deliverySn;
    /**
     * 自动确认时间（天）
     */
    private Integer autoConfirmDay;
    /**
     * 可以获得的积分
     */
    private Integer integration;
    /**
     * 可以获得的成长值
     */
    private Integer growth;
    /**
     * 发票类型[0->不开发票；1->电子发票；2->纸质发票]
     */
    private Integer billType;
    /**
     * 发票抬头
     */
    private String billHeader;
    /**
     * 发票内容
     */
    private String billContent;
    /**
     * 收票人电话
     */
    private String billReceiverPhone;
    /**
     * 收票人邮箱
     */
    private String billReceiverEmail;
    /**
     * 收货人姓名
     */
    private String receiverName;
    /**
     * 收货人电话
     */
    private String receiverPhone;
    /**
     * 收货人邮编
     */
    private String receiverPostCode;
    /**
     * 省份/直辖市
     */
    private String receiverProvince;
    /**
     * 城市
     */
    private String receiverCity;
    /**
     * 区
     */
    private String receiverRegion;
    /**
     * 详细地址
     */
    private String receiverDetailAddress;
    /**
     * 订单备注
     */
    private String note;
    /**
     * 确认收货状态[0->未确认；1->已确认]
     */
    private Integer confirmStatus;
    /**
     * 删除状态【0->未删除；1->已删除】
     */
    private Integer deleteStatus;
    /**
     * 下单时使用的积分
     */
    private Integer useIntegration;
    /**
     * 支付时间
     */
    private Date paymentTime;
    /**
     * 发货时间
     */
    private Date deliveryTime;
    /**
     * 确认收货时间
     */
    private Date receiveTime;
    /**
     * 评价时间
     */
    private Date commentTime;
    /**
     * 修改时间
     */
    private Date modifyTime;

}

• com.achang.achangmall.order.interceptor.LoginUserInterceptor

//用户登录拦截器
@Component
public class LoginUserInterceptor implements HandlerInterceptor {

    public static ThreadLocal<MemberResponseVo> loginUser = new ThreadLocal<>();

    @Override
    public boolean preHandle(HttpServletRequest request, HttpServletResponse response, Object handler) throws Exception {

        boolean match = new AntPathMatcher().match("/order/order/status/**", request.getRequestURI());
        if (match){
            return true;
        }

        HttpSession session = request.getSession();
        MemberResponseVo user = (MemberResponseVo) session.getAttribute(AuthServerConstant.LOGIN_USER);
        if (user!=null){
            //登录了
            loginUser.set(user);
            return true;
        }else {
            //没登录
            session.setAttribute("msg","请先登录");
            response.sendRedirect("http://auth.achangmall.com/login.html");
            return false;
        }
    }

}

---

2、定时关闭订单

首先订单创建成功之后，使用order.create.order路由键将消息路由到order-event-exchange交换机

交换机发现order.create.order这个路由键绑定的是order.delay.queue这个延时队列，然后就把它放到order.delay.queue队列里

过了30分钟，这个延时队列里面的消息，也就是死信，通过order.release.order又路由回order-event-exchange交换机

然后交换机发现这个路由键对应的是order.release.order.queue这个队列，然后就放到order.release.order.queue这个队列里

最终监听order.release.order.queue这个队列的释放订单服务，发现有消息进来了，就会针对里面的数据对其进行关闭订单

• 问题

这种关闭订单方式会有一些问题

假设订单创建成功之后，订单服务的机器由于卡顿、消息延迟等原因，导致订单未及时取消

此时库存服务的逻辑是订单创建成功之后，它自己会发一个消息，等 40分钟 以后检查之前下单的订单是否已取消，如果是已取消，则解锁库存

结果，库存服务过来查询时，订单服务由于上述原因没有将订单修改为已取消，所以库存就不会解锁，此时的库存消息就算是消费了

等库存服务都检查完了，此时的订单服务才反应过来，然后把订单状态改为已取消了，但是此时库存服务会再有任何的操作了，因为检查订单的消息已经被消费了，库存永远得不到解锁。

• 解决

为了解决这个问题，我们在监听取消订单的消息时，再发一个消息，主动解锁库存。

• com.achang.achangmall.order.listener.OrderCloseListener

/**
 * @Description: 定时关闭订单
 **/

@RabbitListener(queues = "order.release.order.queue")
@Service
public class OrderCloseListener {

    @Autowired
    private OrderService orderService;

    @RabbitHandler
    public void listener(OrderEntity orderEntity, Channel channel, Message message) throws IOException {
        System.out.println("收到过期的订单信息，准备关闭订单" + orderEntity.getOrderSn());
        try {
            orderService.closeOrder(orderEntity);
            channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
        } catch (Exception e) {
            channel.basicReject(message.getMessageProperties().getDeliveryTag(),true);
        }

    }

}

• com.achang.achangmall.order.service.impl.OrderServiceImpl

    /**
     * 关闭订单
     * @param orderEntity
     */
@Override
public void closeOrder(OrderEntity orderEntity) {

    //关闭订单之前先查询一下数据库，判断此订单状态是否已支付
    OrderEntity orderInfo = this.getOne(new QueryWrapper<OrderEntity>().
                                        eq("order_sn",orderEntity.getOrderSn()));

    if (orderInfo.getStatus().equals(OrderStatusEnum.CREATE_NEW.getCode())) {
        //代付款状态进行关单
        OrderEntity orderUpdate = new OrderEntity();
        orderUpdate.setId(orderInfo.getId());
        orderUpdate.setStatus(OrderStatusEnum.CANCLED.getCode());
        this.updateById(orderUpdate);

        // 发送消息给MQ
        OrderTo orderTo = new OrderTo();
        BeanUtils.copyProperties(orderInfo, orderTo);

        try {
            //TODO 确保每个消息发送成功，给每个消息做好日志记录，(给数据库保存每一个详细信息)保存每个消息的详细信息
            rabbitTemplate.convertAndSend("order-event-exchange", "order.release.other", orderTo);
        } catch (Exception e) {
            //TODO 定期扫描数据库，重新发送失败的消息
        }
    }
}

• com.achang.achangmall.order.service.impl.OrderServiceImpl

/**
     * 提交订单
     * @param vo
     * @return
     */
// @Transactional(isolation = Isolation.READ_COMMITTED) 设置事务的隔离级别
// @Transactional(propagation = Propagation.REQUIRED)   设置事务的传播级别
@Transactional(rollbackFor = Exception.class)
@GlobalTransactional(rollbackFor = Exception.class)
@Override
public SubmitOrderResponseVo submitOrder(OrderSubmitVo vo) {

    confirmVoThreadLocal.set(vo);

    SubmitOrderResponseVo responseVo = new SubmitOrderResponseVo();
    //去创建、下订单、验令牌、验价格、锁定库存...

    //获取当前用户登录的信息
    MemberResponseVo memberResponseVo = LoginUserInterceptor.loginUser.get();
    responseVo.setCode(0);

    //1、验证令牌是否合法【令牌的对比和删除必须保证原子性】
    String script = "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end";
    String orderToken = vo.getOrderToken();

    //通过lure脚本原子验证令牌和删除令牌
    Long result = redisTemplate.execute(new DefaultRedisScript<Long>(script, Long.class),
                                        Arrays.asList("order:token" + memberResponseVo.getId()),
                                        orderToken);

    if (result == 0L) {
        //令牌验证失败
        responseVo.setCode(1);
        return responseVo;
    } else {
        //令牌验证成功
        //1、创建订单、订单项等信息
        OrderCreateTo order = createOrder();

        //2、验证价格
        BigDecimal payAmount = order.getOrder().getPayAmount();
        BigDecimal payPrice = vo.getPayPrice();

        if (Math.abs(payAmount.subtract(payPrice).doubleValue()) < 0.01) {
            //金额对比
            //TODO 3、保存订单
            saveOrder(order);

            //4、库存锁定,只要有异常，回滚订单数据
            //订单号、所有订单项信息(skuId,skuNum,skuName)
            WareSkuLockVo lockVo = new WareSkuLockVo();
            lockVo.setOrderSn(order.getOrder().getOrderSn());

            //获取出要锁定的商品数据信息
            List<OrderItemVo> orderItemVos = order.getOrderItems().stream().map((item) -> {
                OrderItemVo orderItemVo = new OrderItemVo();
                orderItemVo.setSkuId(item.getSkuId());
                orderItemVo.setCount(item.getSkuQuantity());
                orderItemVo.setTitle(item.getSkuName());
                return orderItemVo;
            }).collect(Collectors.toList());
            lockVo.setLocks(orderItemVos);

            //TODO 调用远程锁定库存的方法
            //出现的问题：扣减库存成功了，但是由于网络原因超时，出现异常，导致订单事务回滚，库存事务不回滚(解决方案：seata)
            //为了保证高并发，不推荐使用seata，因为是加锁，并行化，提升不了效率,可以发消息给库存服务
            R r = wmsFeignService.orderLockStock(lockVo);
            if (r.getCode() == 0) {
                //锁定成功
                responseVo.setOrder(order.getOrder());
                // int i = 10/0;

                //TODO 订单创建成功，发送消息给MQ
                rabbitTemplate.convertAndSend("order-event-exchange","order.create.order",order.getOrder());

                //删除购物车里的数据
                redisTemplate.delete(CART_PREFIX+memberResponseVo.getId());
                return responseVo;
            } else {
                //锁定失败
                String msg = (String) r.get("msg");
                throw new NoStockException(msg);
                // responseVo.setCode(3);
                // return responseVo;
            }

        } else {
            responseVo.setCode(2);
            return responseVo;
        }
    }
}

---

3、主动解锁库存

具体是这样的，在释放订单之后，我们主动发一个消息解锁库存，

使用order.release.other将消息路由到交换机，

交换机根据order.release.other.#匹配到stock.release.stock.queue这个队列，并将消息发了过去，

库存服务有对这个队列进行监听，所有一旦有数据来了，就会对其进行解锁库存服务