基于RabbitMQ的分布式事务最终一致性解决方案

1. 分布式事务

所谓事务，通俗一点讲就是一系列操作要么同时成功，要么同时失败。而分布式事务就是这一系列的操作在不同的节点上，那要如何保证事务的ACID特性呢。

  • 原子性（atomicity）。一个事务是一个不可分割的工作单位，事务中包括的操作要么都成功，要么都失败。

  • 一致性（consistency）。事务必须是使数据库从一个一致性状态变到另一个一致性状态。一致性与原子性是密切相关的。

  • 隔离性（isolation）。一个事务的执行不能被其他事务干扰。即一个事务内部的操作及使用的数据对并发的其他事务是隔离的，并发执行的各个事务之间不能互相干扰。

  • 持久性（durability）。持久性也称永久性（permanence），指一个事务一旦提交或回滚，数据库会对数据持久化的保存。

2. 最终一致性方案

首先，什么叫一致性？一致性指系统中的所有数据备份，在同一时刻具有同样的值，所有节点访问同一份最新的数据副本。那么，什么又叫最终一致性呢。在此之前，先给大家介绍一下BASE理论。

  1. BA（Basically Available）：基本可用。在分布式系统出现故障的时候，允许牺牲部分非核心功能的可用性，常用的手段是访问部分功能时进入降级页面，来保障核心业务的可用性。

  2. S（Soft state）：软状态。允许系统中的数据存在中间状态，并且认为该状态是不影响系统的整体可用性的，即允许系统在不同节点上的数据备份短暂性的不一致。

  3. E（Eventually consistent）：最终一致性。所谓最终一致性，就是数据不可能永久的处于软状态，在一定的时间期限内，所有节点的数据备份应当是一致的，即数据延时一段时间后达到一致性。至于这个时间期限，取决于各种因素，包括业务需求、网络延时、系统负载、存储选型，数据复制方案设计等因素。

3. 可靠消息

所谓的可靠消息，即发布端消息不丢失，可靠抵达队列，消费端可靠接收。以RabbitMQ为例，消息的投递消费过程如下：

1.发布端确认

  1. 如果使用标准的AMQP协议，保证消息不丢失的唯一方法就是使用事务，使通道具有事务性，对每一条消息的发布和提交都是事务性的。在这种情况下，事务是不必要的重量级，并将吞吐量降低了250倍，为了解决这个问题，就引入了确认机制。

    • confirmCallback确认模式

      开启发布者确认

      spring:
    rabbitmq: 
        publisher-confirm-type: correlated

      自定义RabbitTemplate

      @Autowired
RabbitTemplate rabbitTemplate;

@PostConstruct
public void initRabbitTemplate(){

    rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
        /**
             *
             * @param correlationData 当前消息的唯一关联数据
             * @param b 消息是否成功收到
             * @param s 失败的原因
             */
        @Override
        public void confirm(CorrelationData correlationData, boolean b, String s) {
            System.out.println("当前消息【"+correlationData+"】==》服务端是否收到："+b+"==》失败的原因【"+s+"】");
        }
    });
}

    • returnCallback未投递到队列退回模式

      开启消息抵达队列确认

      spring: 
    rabbitmq: 
        publisher-returns: true # 开启发送端消息抵达队列的确认
        template:   #只有抵达队列，以异步发送优先回调returnCallback
          mandatory: true

      设置消息抵达队列回调

      @PostConstruct
public void initRabbitTemplate(){

    rabbitTemplate.setReturnCallback(new RabbitTemplate.ReturnCallback() {
        /**
             * 只要消息没有投递给指定的队列，就触发这个失败回调
             * @param message 投递失败的消息详细信息
             * @param i 回复状态码
             * @param s 回复的文本内容
             * @param s1 当时这个消息发送给哪个交换机
             * @param s2 当时这个消息用哪个路由键
             */
        @Override
        public void returnedMessage(Message message, int i, String s, String s1, String s2) {
            System.out.println("失败信息【"+message+"】==》状态码【"+i+"】==》文本内容【"+s+"】==》交换机【"+s1+"】==》路由键【"+s2+"】");
        }
    });
}

2. 消费端确认

ack机制

默认是自动确认的，只要消息收到，客户端会自动确认，服务端就会移除这个问题

  1. 问题：假如收到很多消息，自动回复给服务器ack，如果一个消息处理成功，宕机了。发生消息丢失

  2. 消费者手动确认模式：只要没有明确告诉MQ，消息被接收，没有Ack，消息就一直是unacked状态，即使服务器宕机，消息也不会丢失，会重新变为Ready状态。

    1. 开启手动确认模式

      spring: 
  raabbitmq: 
        listener:
          simple:
            acknowledge-mode: manual

    2. 手动签收

      long deliveryTag = message.getMessageProperties().getDeliveryTag();
try {
    // 签收 long deliveryTag, boolean mulitiple(是否批量模式)
    channel.basicAck(deliveryTag,false);
} catch (IOException e) {
    e.printStackTrace();
}

// 拒签 long deliveryTag, boolean mulitiple(是否批量模式), boolean requeue(是否重新入队)
channel.basicNack(deliveryTag,false,false);

4. 分布式事务案例

在电商背景下，以订单和库存系统之间的分布式事务为例，来介绍分布式事务基于消息队列的最终一致性方案。下单和扣减库存操作要么同时成功，要么同时失败，是事务的。如果只是本地事务的话，操作同一数据库，依赖数据库本身的事务特性，就可以完成。但是，对于分布式系统而言，订单系统和库存系统是操作不同的数据库的，那要如何实现这样的分布式事务。

1. 业务流程及问题

  1. 一般的业务流程是：下单成功后，远程调用库存服务，扣减库存。伪代码如下

    public void saveOrder(){
    // 创建订单
    Order order = createOrder();
    // 保存订单
    orderService.save(order);
    // 远程调用库存服务
    wareFeignService.sub(order);
}

  2. 订单系统和库存系统本地是满足事务的。即订单服务发生异常，订单回滚；库存服务发生异常，库存是会回滚的。

  3. 由于订单服务是以内嵌的方式远程调用库存服务的，也就是说，库存服务发生异常，订单服务感知到远程调用异常，从而订单会回滚的，对于业务来说，这是没有问题的。

  4. 如果订单服务在远程调用库存服务之前发生异常，订单会回滚，并且也不会调用库存服务来扣减库存，这也是没有问题的。

  5. 如果订单服务在远程调用库存服务之后，并且远程扣减库存操作成功后，发生异常，则订单会回滚，但是远程库存服务是无法回滚的。这就导致了数据的不一致性。

2. 基于消息队列的解决方案分析

我们使用RabbitMQ来实现分布式事务的最终一致性。

1. 业务分析

  1. 由于用户下单和支付并不是同时进行，一般都是下单成功后，30min内可以支付。那我们来思考这样一个问题，如果我们在下单成功就扣减库存的话，会不会有什么问题。

    • 恶意刷单。下单后不支付，导致其他人无法下单。

  2. 那如果支付成功后再扣减库存呢？

    • 在支付订单时，会出现库存不足，支付失败。

  3. 综合这两种情况考虑，我们在下单成功后先锁定库存，支付成功再去扣减库存，如果超时未支付，则解锁库存。

2. 业务流程

  1. 下单成功，锁定库存

  2. 订单支付超时，则需要自动关闭订单。

  3. 订单关闭，库存需要解锁。

3. 定时任务

如何确保下单后，30min内保留订单。最先想到的方法应该时定时任务。

  1. 定时任务使用的是系统时间，我们无法为每一个订单都生成一定时任务。

  2. 我想大家应该都发现了使用定时任务会带来的问题，那就是每一个订单的保留时间并不是一致的30min，订单保留的时间区间为（0min，60min）。即在定时任务即将到来前完成下单，和定时任务刚结束完成下单。

所以使用定时任务来完成这个操作是不可行。

4. RabbitMQ延时队列

利用消息的存活时间和死信来完成延时任务。

1. 消息的存活时间（TTL：Time To Live）

  1. RabbitMQ可以对队列和消息分别设置TTL。

  2. 对队列设置就是队列没有消费者连着的保留时间，也可以对每一个单独的消息做单独的设置。超过了这个时间，我们认为这个消息就死了，称之为死信。

  3. 如果队列和消息都设置了，那么会取小的。所以一个消息如果被路由到不同的队列中，这个消息的死亡时间有可能不一样（不同的队列设置）。

  4. 单个消息的TTL，才是实现延迟任务的关键。可以通过设置消息的expiration字段或者x-message-ttl属性来设置时间，两者是一样的效果。

2. 死信交换机（DLX：Dead Letter Exchanges）

  1. 一个消息如果满足如下条件，就会进入死信路由（不是队列，一个路由可以对应多个队列）

    • 一个消息被消费者拒收，并且reject方法的参数里requeue是false。也就是说不会被再次放在队列里，被其他消费者使用。

    • 消息的TTL到了，消息过期了。

    • 队列长度限制满了，排在前面的消息会被丢弃或者扔到死信路由上。

  2. 在某一个设置Dead Letter Exchange的队列中有消息过期了，会自动触发消息的转发，发送到Dead Letter Exchange中去

  3. 先控制消息在一段时间后变成死信，然后控制变成死信的消息被路由到某个指定的交换机。二者结合就可以实现一个延时队列。

在下单成功后，发送一条消息到死信队列，经过一段时间（TTL），死信路由到订单释放队列，订单服务监听到消息释放放订单。

在这里插入图片描述

5. 分布式解决方案

1. 对以下情况，库存需要解锁。

  1. 首先之前提到的，订单服务在远程调用成功后，发生异常，导致订单回滚，库存也需要解锁。

  2. 订单延时取消，或者主动取消订单，都需要解锁库存。

  3. 订单服务下单成功后，订单服务宕机，超过订单支付时间，仍然无法恢复，导致无法发送消息通知库存服务解锁库存，故需要自动解锁库存。

2. 解决方案

针对以上情形，库存解锁的方案。

  1. 第一种情况，可以利用库存自动解锁来解决。库存锁定时发送消息到延时队列，经过TTL后，成为死信路由到库存解锁队列，库存服务监听到消息后，解锁库存。

    • 不能在订单未支付时就解锁库存，所以库存自动解锁的延迟时间应该大于订单延时取消的时间。

  2. 对于第二种情况，主动取消或者延时取消，都可以通过库存的自动解锁来完成库存的解锁。

  3. 自动解锁时，需要判断订单的状态，只有为取消状态的订单才可以解锁库存。但是这样仍然会存在问题。

    • 订单服务卡顿，导致订单状态消息一直改不了，而库存消息先到期，查询订单状态为新建状态，不解锁库存，并删除消息，导致库永远无法解锁。

    • 解决：订单超时取消的同时，发送订单取消的消息到队列，库存服务监听该消息，则解锁库存。

    • 为了防止重复解锁，需要满足幂等性。

3. 代码实现

使用消息队列实现分布式事务的最终一致性方案的流程图如下：


在这里插入图片描述

1. 订单服务

  1. 创建交换机、队列和绑定关系
@Configuration
public class OrderRabbitMQConfig {

   /**
    * 使用JSON序列化机制，进行消息转换
    * @return
    */
   @Bean
   public MessageConverter messageConverter(){
       return new Jackson2JsonMessageConverter();
   }
   @Bean
   public Exchange orderEventExchange(){
       return new TopicExchange("order-event-exchange",true,false);
   }

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

   @Bean
   public Queue orderDelayQueue(){
       HashMap<String, Object> args = new HashMap<>();
       args.put("x-dead-letter-exchange","order-event-exchange");
       args.put("x-dead-letter-routing-key","order.release.order");
       args.put("x-message-ttl",60000);
       return new Queue("order.delay.queue",true,false,false,args);
   }

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

   @Bean
   public Binding stockLockedBinding(){
       return new Binding("order.delay.queue",
               Binding.DestinationType.QUEUE,
               "order-event-exchange",
               "order.create.order",
               null);
   }
   /**
    * 订单释放直接和库存释放进行绑定
    */
   @Bean
   public Binding orderReleaseOtherBinding(){
       return new Binding("stock.release.stock.queue",
               Binding.DestinationType.QUEUE,
               "order-event-exchange",
               "order.release.other.#",
               null);
   }

}
  1. 订单创建伪代码
@Transactional
public Order createOrder(){
  // 创建订单
  Order order = createOrder();
  // 远程调用库存服务，锁定库存
  R r = wareFeignService.orderLockStock(wareSkuLockVo);
  if (r.getCode() == 0) {
    // 锁定成功，发送订单创建消息到延时队列
    rabbitTemplate.convertAndSend("order-event-exchange","order.create.order",order);
  } else {
     // 远程调用失败，抛出异常
     throw new Exception();
  }
  reture order;
}
  1. 订单服务监听订单释放信息，关闭订单
@RabbitListener(queues = "order.release.order.queue")
@Service
public class OrderCloseListener {

   @Autowired
   OmsOrderService orderService;

   @RabbitHandler
   public void listener(OmsOrderEntity orderEntity, Channel channel, Message message) throws IOException {
       System.out.println("收到过期订单，准备关单："+orderEntity.getBizOrderId());
       try{
           orderService.closeOrder(orderEntity.getBizOrderId());
           channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
       }catch (Exception e){
           channel.basicReject(message.getMessageProperties().getDeliveryTag(), true);
       }
   }
}
  1. 关闭订单，发送解锁库存消息。伪代码如下
public void closeOrder(String bizOrderId) {
  // 查询当前订单是否付款
  OmsOrderEntity order = getOne(new QueryWrapper<OmsOrderEntity>().eq("biz_order_id", bizOrderId));
  if (order != null) {
    // 判断订单状态，为新建状态才取消
    if (order.getOrderStatus() == OrderStatusConstant.CREATE.getCode()){
      //过期未支付，取消订单，设置订单状态为取消状态
      updateOrder.setOrderStatus(OrderStatusConstant.CANCEL.getCode());
      // 发送MQ
      rabbitTemplate.convertAndSend("order-event-exchange","order.release.other",orderTo);
      }
    }
}

2. 库存服务

  1. 创建交换机、队列和绑定关系
@Configuration
public class MyRabbitConfig {

    /**
     * 使用JSON序列化机制，进行消息转换
     * @return
     */
    @Bean
    public MessageConverter messageConverter(){
        return new Jackson2JsonMessageConverter();
    }
    @Bean
    public Exchange stockEventExchange(){
        return new TopicExchange("stock-event-exchange",true,true);
    }

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

    @Bean
    public Queue stockDelayQueue(){
        HashMap<String, Object> args = new HashMap<>();
        args.put("x-dead-letter-exchange","stock-event-exchange");
        args.put("x-dead-letter-routing-key","stock.release");
        args.put("x-message-ttl",120000);
        return new Queue("stock.delay.queue",true,false,false,args);
    }

    @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);
    }
}
  1. 库存锁定，伪代码如下
@Transactional
public void orderLockStock() {
  // 保存工作单
  taskService.save(task);
  // 为每件商品锁定库存
  for (OrderItemVo orderItem : orderItems) {
    //判断库存是否足够
    Long count = skuWareDao.lockStock(orderItem);
    if (count == 1){
        // 锁定成功
        // 保存工作单详情
        taskDetailService.save(taskDetail);
        // 自动解锁，发送工作单详情，防止回滚后找不到数据
        // 锁定库存，发送消息到延时队列
        rabbitTemplate.convertAndSend("stock-event-exchange", "stock.locked", taskTo);
    } else {
        // 锁定失败，抛出异常
        throw new Exception();
    }
  }
}
  1. 监听库存解锁信息
@Service
@RabbitListener(queues = "stock.release.stock.queue")
public class StockReleaseListener {

  @Autowired
  WmsSkuWareService wareSkuService;

  /**
   * 库存自动解锁
   *只要解锁库存的消息的失败，需要告诉MQ解锁失败，消息不要删除，重新放回队列
   * @param taskTo
   * @param message
   *
   */
  @RabbitHandler
  public void handleStockLockedRelease(StockLockedTaskTo taskTo, Message message, Channel channel) throws IOException {
    System.out.println("收到解锁库存的消息");
    try{
      wareSkuService.unlockStock(taskTo);
      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 {
    System.out.println("订单关闭，准备解锁库存");
    try{
      wareSkuService.unlockStock(orderTo);
      channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
    }catch (Exception e){
      channel.basicReject(message.getMessageProperties().getDeliveryTag(), true);
    }
  }
}
  1. 自动解锁库存，伪代码如下
@Transactional
public void unlockStock(StockLockedTaskTo taskTo) {
  // 判断是否存在该任务
  Detail detail = detailService.getById(taskTo.detailId)
  if (detail != null){
    // 查询订单状态，订单状态为【取消】，则解锁库存
    // 远程调用订单服务，获取订单状态
    R r = orderFeignService.getOrderStatus(bizOrderId);
    if (r.getCode() == 0) {
      // 远程查询成功
      OrderVo orderVo = r.getData(new TypeReference<OrderVo>() {
      });
      if (orderVo == null || orderVo.getOrderStatus() == OrderStatusConstant.CANCEL.getCode()) {
        // 订单不存在，或者订单已取消，都需要解锁库存
        if (detail.getLockStatus() == WareTaskStatusConstant.LOCKED.getCode()){
          //锁定状态下才需要解锁，已解锁的不用在解锁
          unlockStock(detail.getId(),detail.getSkuId(),detail.getSkuNum());
          // 更新taskDetail状态为已解锁
          taskDetailService.updateById(taskDetailEntity);
        }
      }
    } else {
      throw new RuntimeException("远程调用订单服务失败");
    }
  }
}
  1. 订单取消，解锁库存，伪代码如下
@Transactional
public void unlockStock(OrderTo orderTo) {
  // 无需查询订单最新状态，能来到这，肯定更新了订单状态的
  // 判断任务是否存在
  WmsOrderTaskEntity task = taskService.getOne(new QueryWrapper<WmsOrderTaskEntity>().eq("biz_order_id", bizOrderId));
  if (task != null){
    // 任务存在，获取任务项状态为锁定状态的所有任务项
    List<WmsOrderTaskDetailEntity> taskDetails = taskDetailService.list(new QueryWrapper<WmsOrderTaskDetailEntity>()
            .eq("task_id", task.getId())
            .eq("lock_status", WareTaskStatusConstant.LOCKED.getCode()));
    if (taskDetails != null && taskDetails.size() > 0){
      for (WmsOrderTaskDetailEntity taskDetail : taskDetails) {
        // 解锁库存
        unlockStock(taskDetail.getId(),taskDetail.getSkuId(),taskDetail.getSkuNum());
        // 更新taskDetail状态为已解锁
        taskDetailService.updateById(taskDetailEntity);
      }
    }
  }
}

使用RabbitMQ实现分布式事务的最终一致性方案的大致流程解析完毕。如果对源码感兴趣的，欢迎到github仓库clone。

1. 分布式事务

所谓事务，通俗一点讲就是一系列操作要么同时成功，要么同时失败。而分布式事务就是这一系列的操作在不同的节点上，那要如何保证事务的ACID特性呢。

  • 原子性（atomicity）。一个事务是一个不可分割的工作单位，事务中包括的操作要么都成功，要么都失败。

  • 一致性（consistency）。事务必须是使数据库从一个一致性状态变到另一个一致性状态。一致性与原子性是密切相关的。

  • 隔离性（isolation）。一个事务的执行不能被其他事务干扰。即一个事务内部的操作及使用的数据对并发的其他事务是隔离的，并发执行的各个事务之间不能互相干扰。

  • 持久性（durability）。持久性也称永久性（permanence），指一个事务一旦提交或回滚，数据库会对数据持久化的保存。

2. 最终一致性方案

首先，什么叫一致性？一致性指系统中的所有数据备份，在同一时刻具有同样的值，所有节点访问同一份最新的数据副本。那么，什么又叫最终一致性呢。在此之前，先给大家介绍一下BASE理论。

  1. BA（Basically Available）：基本可用。在分布式系统出现故障的时候，允许牺牲部分非核心功能的可用性，常用的手段是访问部分功能时进入降级页面，来保障核心业务的可用性。

  2. S（Soft state）：软状态。允许系统中的数据存在中间状态，并且认为该状态是不影响系统的整体可用性的，即允许系统在不同节点上的数据备份短暂性的不一致。

  3. E（Eventually consistent）：最终一致性。所谓最终一致性，就是数据不可能永久的处于软状态，在一定的时间期限内，所有节点的数据备份应当是一致的，即数据延时一段时间后达到一致性。至于这个时间期限，取决于各种因素，包括业务需求、网络延时、系统负载、存储选型，数据复制方案设计等因素。

3. 可靠消息

所谓的可靠消息，即发布端消息不丢失，可靠抵达队列，消费端可靠接收。以RabbitMQ为例，消息的投递消费过程如下：

1.发布端确认

  1. 如果使用标准的AMQP协议，保证消息不丢失的唯一方法就是使用事务，使通道具有事务性，对每一条消息的发布和提交都是事务性的。在这种情况下，事务是不必要的重量级，并将吞吐量降低了250倍，为了解决这个问题，就引入了确认机制。

    • confirmCallback确认模式

      开启发布者确认

      spring:
    rabbitmq: 
        publisher-confirm-type: correlated

      自定义RabbitTemplate

      @Autowired
RabbitTemplate rabbitTemplate;

@PostConstruct
public void initRabbitTemplate(){

    rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
        /**
             *
             * @param correlationData 当前消息的唯一关联数据
             * @param b 消息是否成功收到
             * @param s 失败的原因
             */
        @Override
        public void confirm(CorrelationData correlationData, boolean b, String s) {
            System.out.println("当前消息【"+correlationData+"】==》服务端是否收到："+b+"==》失败的原因【"+s+"】");
        }
    });
}

    • returnCallback未投递到队列退回模式

      开启消息抵达队列确认

      spring: 
    rabbitmq: 
        publisher-returns: true # 开启发送端消息抵达队列的确认
        template:   #只有抵达队列，以异步发送优先回调returnCallback
          mandatory: true

      设置消息抵达队列回调

      @PostConstruct
public void initRabbitTemplate(){

    rabbitTemplate.setReturnCallback(new RabbitTemplate.ReturnCallback() {
        /**
             * 只要消息没有投递给指定的队列，就触发这个失败回调
             * @param message 投递失败的消息详细信息
             * @param i 回复状态码
             * @param s 回复的文本内容
             * @param s1 当时这个消息发送给哪个交换机
             * @param s2 当时这个消息用哪个路由键
             */
        @Override
        public void returnedMessage(Message message, int i, String s, String s1, String s2) {
            System.out.println("失败信息【"+message+"】==》状态码【"+i+"】==》文本内容【"+s+"】==》交换机【"+s1+"】==》路由键【"+s2+"】");
        }
    });
}

2. 消费端确认

ack机制

默认是自动确认的，只要消息收到，客户端会自动确认，服务端就会移除这个问题

  1. 问题：假如收到很多消息，自动回复给服务器ack，如果一个消息处理成功，宕机了。发生消息丢失

  2. 消费者手动确认模式：只要没有明确告诉MQ，消息被接收，没有Ack，消息就一直是unacked状态，即使服务器宕机，消息也不会丢失，会重新变为Ready状态。

    1. 开启手动确认模式

      spring: 
  raabbitmq: 
        listener:
          simple:
            acknowledge-mode: manual

    2. 手动签收

      long deliveryTag = message.getMessageProperties().getDeliveryTag();
try {
    // 签收 long deliveryTag, boolean mulitiple(是否批量模式)
    channel.basicAck(deliveryTag,false);
} catch (IOException e) {
    e.printStackTrace();
}

// 拒签 long deliveryTag, boolean mulitiple(是否批量模式), boolean requeue(是否重新入队)
channel.basicNack(deliveryTag,false,false);

4. 分布式事务案例

在电商背景下，以订单和库存系统之间的分布式事务为例，来介绍分布式事务基于消息队列的最终一致性方案。下单和扣减库存操作要么同时成功，要么同时失败，是事务的。如果只是本地事务的话，操作同一数据库，依赖数据库本身的事务特性，就可以完成。但是，对于分布式系统而言，订单系统和库存系统是操作不同的数据库的，那要如何实现这样的分布式事务。

1. 业务流程及问题

  1. 一般的业务流程是：下单成功后，远程调用库存服务，扣减库存。伪代码如下

    public void saveOrder(){
    // 创建订单
    Order order = createOrder();
    // 保存订单
    orderService.save(order);
    // 远程调用库存服务
    wareFeignService.sub(order);
}

  2. 订单系统和库存系统本地是满足事务的。即订单服务发生异常，订单回滚；库存服务发生异常，库存是会回滚的。

  3. 由于订单服务是以内嵌的方式远程调用库存服务的，也就是说，库存服务发生异常，订单服务感知到远程调用异常，从而订单会回滚的，对于业务来说，这是没有问题的。

  4. 如果订单服务在远程调用库存服务之前发生异常，订单会回滚，并且也不会调用库存服务来扣减库存，这也是没有问题的。

  5. 如果订单服务在远程调用库存服务之后，并且远程扣减库存操作成功后，发生异常，则订单会回滚，但是远程库存服务是无法回滚的。这就导致了数据的不一致性。

2. 基于消息队列的解决方案分析

我们使用RabbitMQ来实现分布式事务的最终一致性。

1. 业务分析

  1. 由于用户下单和支付并不是同时进行，一般都是下单成功后，30min内可以支付。那我们来思考这样一个问题，如果我们在下单成功就扣减库存的话，会不会有什么问题。

    • 恶意刷单。下单后不支付，导致其他人无法下单。

  2. 那如果支付成功后再扣减库存呢？

    • 在支付订单时，会出现库存不足，支付失败。

  3. 综合这两种情况考虑，我们在下单成功后先锁定库存，支付成功再去扣减库存，如果超时未支付，则解锁库存。

2. 业务流程

  1. 下单成功，锁定库存

  2. 订单支付超时，则需要自动关闭订单。

  3. 订单关闭，库存需要解锁。

3. 定时任务

如何确保下单后，30min内保留订单。最先想到的方法应该时定时任务。

  1. 定时任务使用的是系统时间，我们无法为每一个订单都生成一定时任务。

  2. 我想大家应该都发现了使用定时任务会带来的问题，那就是每一个订单的保留时间并不是一致的30min，订单保留的时间区间为（0min，60min）。即在定时任务即将到来前完成下单，和定时任务刚结束完成下单。

所以使用定时任务来完成这个操作是不可行。

4. RabbitMQ延时队列

利用消息的存活时间和死信来完成延时任务。

1. 消息的存活时间（TTL：Time To Live）

  1. RabbitMQ可以对队列和消息分别设置TTL。

  2. 对队列设置就是队列没有消费者连着的保留时间，也可以对每一个单独的消息做单独的设置。超过了这个时间，我们认为这个消息就死了，称之为死信。

  3. 如果队列和消息都设置了，那么会取小的。所以一个消息如果被路由到不同的队列中，这个消息的死亡时间有可能不一样（不同的队列设置）。

  4. 单个消息的TTL，才是实现延迟任务的关键。可以通过设置消息的expiration字段或者x-message-ttl属性来设置时间，两者是一样的效果。

2. 死信交换机（DLX：Dead Letter Exchanges）

  1. 一个消息如果满足如下条件，就会进入死信路由（不是队列，一个路由可以对应多个队列）

    • 一个消息被消费者拒收，并且reject方法的参数里requeue是false。也就是说不会被再次放在队列里，被其他消费者使用。

    • 消息的TTL到了，消息过期了。

    • 队列长度限制满了，排在前面的消息会被丢弃或者扔到死信路由上。

  2. 在某一个设置Dead Letter Exchange的队列中有消息过期了，会自动触发消息的转发，发送到Dead Letter Exchange中去

  3. 先控制消息在一段时间后变成死信，然后控制变成死信的消息被路由到某个指定的交换机。二者结合就可以实现一个延时队列。

在下单成功后，发送一条消息到死信队列，经过一段时间（TTL），死信路由到订单释放队列，订单服务监听到消息释放放订单。

在这里插入图片描述

5. 分布式解决方案

1. 对以下情况，库存需要解锁。

  1. 首先之前提到的，订单服务在远程调用成功后，发生异常，导致订单回滚，库存也需要解锁。

  2. 订单延时取消，或者主动取消订单，都需要解锁库存。

  3. 订单服务下单成功后，订单服务宕机，超过订单支付时间，仍然无法恢复，导致无法发送消息通知库存服务解锁库存，故需要自动解锁库存。

2. 解决方案

针对以上情形，库存解锁的方案。

  1. 第一种情况，可以利用库存自动解锁来解决。库存锁定时发送消息到延时队列，经过TTL后，成为死信路由到库存解锁队列，库存服务监听到消息后，解锁库存。

    • 不能在订单未支付时就解锁库存，所以库存自动解锁的延迟时间应该大于订单延时取消的时间。

  2. 对于第二种情况，主动取消或者延时取消，都可以通过库存的自动解锁来完成库存的解锁。

  3. 自动解锁时，需要判断订单的状态，只有为取消状态的订单才可以解锁库存。但是这样仍然会存在问题。

    • 订单服务卡顿，导致订单状态消息一直改不了，而库存消息先到期，查询订单状态为新建状态，不解锁库存，并删除消息，导致库永远无法解锁。

    • 解决：订单超时取消的同时，发送订单取消的消息到队列，库存服务监听该消息，则解锁库存。

    • 为了防止重复解锁，需要满足幂等性。

3. 代码实现

使用消息队列实现分布式事务的最终一致性方案的流程图如下：


在这里插入图片描述

1. 订单服务

  1. 创建交换机、队列和绑定关系
@Configuration
public class OrderRabbitMQConfig {

   /**
    * 使用JSON序列化机制，进行消息转换
    * @return
    */
   @Bean
   public MessageConverter messageConverter(){
       return new Jackson2JsonMessageConverter();
   }
   @Bean
   public Exchange orderEventExchange(){
       return new TopicExchange("order-event-exchange",true,false);
   }

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

   @Bean
   public Queue orderDelayQueue(){
       HashMap<String, Object> args = new HashMap<>();
       args.put("x-dead-letter-exchange","order-event-exchange");
       args.put("x-dead-letter-routing-key","order.release.order");
       args.put("x-message-ttl",60000);
       return new Queue("order.delay.queue",true,false,false,args);
   }

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

   @Bean
   public Binding stockLockedBinding(){
       return new Binding("order.delay.queue",
               Binding.DestinationType.QUEUE,
               "order-event-exchange",
               "order.create.order",
               null);
   }
   /**
    * 订单释放直接和库存释放进行绑定
    */
   @Bean
   public Binding orderReleaseOtherBinding(){
       return new Binding("stock.release.stock.queue",
               Binding.DestinationType.QUEUE,
               "order-event-exchange",
               "order.release.other.#",
               null);
   }

}
  1. 订单创建伪代码
@Transactional
public Order createOrder(){
  // 创建订单
  Order order = createOrder();
  // 远程调用库存服务，锁定库存
  R r = wareFeignService.orderLockStock(wareSkuLockVo);
  if (r.getCode() == 0) {
    // 锁定成功，发送订单创建消息到延时队列
    rabbitTemplate.convertAndSend("order-event-exchange","order.create.order",order);
  } else {
     // 远程调用失败，抛出异常
     throw new Exception();
  }
  reture order;
}
  1. 订单服务监听订单释放信息，关闭订单
@RabbitListener(queues = "order.release.order.queue")
@Service
public class OrderCloseListener {

   @Autowired
   OmsOrderService orderService;

   @RabbitHandler
   public void listener(OmsOrderEntity orderEntity, Channel channel, Message message) throws IOException {
       System.out.println("收到过期订单，准备关单："+orderEntity.getBizOrderId());
       try{
           orderService.closeOrder(orderEntity.getBizOrderId());
           channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
       }catch (Exception e){
           channel.basicReject(message.getMessageProperties().getDeliveryTag(), true);
       }
   }
}
  1. 关闭订单，发送解锁库存消息。伪代码如下
public void closeOrder(String bizOrderId) {
  // 查询当前订单是否付款
  OmsOrderEntity order = getOne(new QueryWrapper<OmsOrderEntity>().eq("biz_order_id", bizOrderId));
  if (order != null) {
    // 判断订单状态，为新建状态才取消
    if (order.getOrderStatus() == OrderStatusConstant.CREATE.getCode()){
      //过期未支付，取消订单，设置订单状态为取消状态
      updateOrder.setOrderStatus(OrderStatusConstant.CANCEL.getCode());
      // 发送MQ
      rabbitTemplate.convertAndSend("order-event-exchange","order.release.other",orderTo);
      }
    }
}

2. 库存服务

  1. 创建交换机、队列和绑定关系
@Configuration
public class MyRabbitConfig {

    /**
     * 使用JSON序列化机制，进行消息转换
     * @return
     */
    @Bean
    public MessageConverter messageConverter(){
        return new Jackson2JsonMessageConverter();
    }
    @Bean
    public Exchange stockEventExchange(){
        return new TopicExchange("stock-event-exchange",true,true);
    }

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

    @Bean
    public Queue stockDelayQueue(){
        HashMap<String, Object> args = new HashMap<>();
        args.put("x-dead-letter-exchange","stock-event-exchange");
        args.put("x-dead-letter-routing-key","stock.release");
        args.put("x-message-ttl",120000);
        return new Queue("stock.delay.queue",true,false,false,args);
    }

    @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);
    }
}
  1. 库存锁定，伪代码如下
@Transactional
public void orderLockStock() {
  // 保存工作单
  taskService.save(task);
  // 为每件商品锁定库存
  for (OrderItemVo orderItem : orderItems) {
    //判断库存是否足够
    Long count = skuWareDao.lockStock(orderItem);
    if (count == 1){
        // 锁定成功
        // 保存工作单详情
        taskDetailService.save(taskDetail);
        // 自动解锁，发送工作单详情，防止回滚后找不到数据
        // 锁定库存，发送消息到延时队列
        rabbitTemplate.convertAndSend("stock-event-exchange", "stock.locked", taskTo);
    } else {
        // 锁定失败，抛出异常
        throw new Exception();
    }
  }
}
  1. 监听库存解锁信息
@Service
@RabbitListener(queues = "stock.release.stock.queue")
public class StockReleaseListener {

  @Autowired
  WmsSkuWareService wareSkuService;

  /**
   * 库存自动解锁
   *只要解锁库存的消息的失败，需要告诉MQ解锁失败，消息不要删除，重新放回队列
   * @param taskTo
   * @param message
   *
   */
  @RabbitHandler
  public void handleStockLockedRelease(StockLockedTaskTo taskTo, Message message, Channel channel) throws IOException {
    System.out.println("收到解锁库存的消息");
    try{
      wareSkuService.unlockStock(taskTo);
      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 {
    System.out.println("订单关闭，准备解锁库存");
    try{
      wareSkuService.unlockStock(orderTo);
      channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
    }catch (Exception e){
      channel.basicReject(message.getMessageProperties().getDeliveryTag(), true);
    }
  }
}
  1. 自动解锁库存，伪代码如下
@Transactional
public void unlockStock(StockLockedTaskTo taskTo) {
  // 判断是否存在该任务
  Detail detail = detailService.getById(taskTo.detailId)
  if (detail != null){
    // 查询订单状态，订单状态为【取消】，则解锁库存
    // 远程调用订单服务，获取订单状态
    R r = orderFeignService.getOrderStatus(bizOrderId);
    if (r.getCode() == 0) {
      // 远程查询成功
      OrderVo orderVo = r.getData(new TypeReference<OrderVo>() {
      });
      if (orderVo == null || orderVo.getOrderStatus() == OrderStatusConstant.CANCEL.getCode()) {
        // 订单不存在，或者订单已取消，都需要解锁库存
        if (detail.getLockStatus() == WareTaskStatusConstant.LOCKED.getCode()){
          //锁定状态下才需要解锁，已解锁的不用在解锁
          unlockStock(detail.getId(),detail.getSkuId(),detail.getSkuNum());
          // 更新taskDetail状态为已解锁
          taskDetailService.updateById(taskDetailEntity);
        }
      }
    } else {
      throw new RuntimeException("远程调用订单服务失败");
    }
  }
}
  1. 订单取消，解锁库存，伪代码如下
@Transactional
public void unlockStock(OrderTo orderTo) {
  // 无需查询订单最新状态，能来到这，肯定更新了订单状态的
  // 判断任务是否存在
  WmsOrderTaskEntity task = taskService.getOne(new QueryWrapper<WmsOrderTaskEntity>().eq("biz_order_id", bizOrderId));
  if (task != null){
    // 任务存在，获取任务项状态为锁定状态的所有任务项
    List<WmsOrderTaskDetailEntity> taskDetails = taskDetailService.list(new QueryWrapper<WmsOrderTaskDetailEntity>()
            .eq("task_id", task.getId())
            .eq("lock_status", WareTaskStatusConstant.LOCKED.getCode()));
    if (taskDetails != null && taskDetails.size() > 0){
      for (WmsOrderTaskDetailEntity taskDetail : taskDetails) {
        // 解锁库存
        unlockStock(taskDetail.getId(),taskDetail.getSkuId(),taskDetail.getSkuNum());
        // 更新taskDetail状态为已解锁
        taskDetailService.updateById(taskDetailEntity);
      }
    }
  }
}

使用RabbitMQ实现分布式事务的最终一致性方案的大致流程解析完毕。如果对源码感兴趣的，欢迎到github仓库clone。

©著作权归作者所有,转载或内容合作请联系作者
  • 人面猴
    序言：七十年代末，一起剥皮案震惊了整个滨河市，随后出现的几起案子，更是在滨河造成了极大的恐慌，老刑警刘岩，带你破解...
    沈念sama阅读 203,324评论 5赞 476
  • 死咒
    序言：滨河连续发生了三起死亡事件，死亡现场离奇诡异，居然都是意外死亡，警方通过查阅死者的电脑和手机，发现死者居然都...
    沈念sama阅读 85,303评论 2赞 381
  • 救了他两次的神仙让他今天三更去死
    文/潘晓璐 我一进店门，熙熙楼的掌柜王于贵愁眉苦脸地迎上来，“玉大人，你说我怎么就摊上这事。” “怎么了？”我有些...
    开封第一讲书人阅读 150,192评论 0赞 337
  • 道士缉凶录：失踪的卖姜人
    文/不坏的土叔 我叫张陵，是天一观的道长。 经常有香客问我，道长，这世上最难降的妖魔是什么？ 我笑而不...
    开封第一讲书人阅读 54,555评论 1赞 273
  • 港岛之恋（遗憾婚礼）
    正文 为了忘掉前任，我火速办了婚礼，结果婚礼上，老公的妹妹穿的比我还像新娘。我一直安慰自己，他们只是感情好，可当我...
    茶点故事阅读 63,569评论 5赞 365
  • 恶毒庶女顶嫁案：这布局不是一般人想出来的
    文/花漫 我一把揭开白布。 她就那样静静地躺着，像睡着了一般。 火红的嫁衣衬着肌肤如雪。 梳的纹丝不乱的头发上，一...
    开封第一讲书人阅读 48,566评论 1赞 281
  • 城市分裂传说
    那天，我揣着相机与录音，去河边找鬼。 笑死，一个胖子当着我的面吹牛，可吹牛的内容都是我干的。 我是一名探鬼主播，决...
    沈念sama阅读 37,927评论 3赞 395
  • 双鸳鸯连环套：你想象不到人心有多黑
    文/苍兰香墨 我猛地睁开眼，长吁一口气：“原来是场噩梦啊……” “哼！你这毒妇竟也来了？” 一声冷哼从身侧响起，我...
    开封第一讲书人阅读 36,583评论 0赞 257
  • 万荣杀人案实录
    序言：老挝万荣一对情侣失踪，失踪者是张志新（化名）和其女友刘颖，没想到半个月后，有当地人在树林里发现了一具尸体，经...
    沈念sama阅读 40,827评论 1赞 297
  • 护林员之死
    正文 独居荒郊野岭守林人离奇死亡，尸身上长有42处带血的脓包…… 初始之章·张勋 以下内容为张勋视角 年9月15日...
    茶点故事阅读 35,590评论 2赞 320
  • 白月光启示录
    正文 我和宋清朗相恋三年，在试婚纱的时候发现自己被绿了。 大学时的朋友给我发了我未婚夫和他白月光在一起吃饭的照片。...
    茶点故事阅读 37,669评论 1赞 329
  • 活死人
    序言：一个原本活蹦乱跳的男人离奇死亡，死状恐怖，灵堂内的尸体忽然破棺而出，到底是诈尸还是另有隐情，我是刑警宁泽，带...
    沈念sama阅读 33,365评论 4赞 318
  • 日本核电站爆炸内幕
    正文 年R本政府宣布，位于F岛的核电站，受9级特大地震影响，放射性物质发生泄漏。R本人自食恶果不足惜，却给世界环境...
    茶点故事阅读 38,941评论 3赞 307
  • 男人毒药：我在死后第九天来索命
    文/蒙蒙 一、第九天 我趴在偏房一处隐蔽的房顶上张望。 院中可真热闹，春花似锦、人声如沸。这庄子的主人今日做“春日...
    开封第一讲书人阅读 29,928评论 0赞 19
  • 一桩弑父案，背后竟有这般阴谋
    文/苍兰香墨 我抬头看了看天上的太阳。三九已至，却和暖如春，着一层夹袄步出监牢的瞬间，已是汗流浃背。 一阵脚步声响...
    开封第一讲书人阅读 31,159评论 1赞 259
  • 情欲美人皮
    我被黑心中介骗来泰国打工， 没想到刚下飞机就差点儿被人妖公主榨干…… 1. 我叫王不留，地道东北人。 一个月前我还...
    沈念sama阅读 42,880评论 2赞 349
  • 代替公主和亲
    正文 我出身青楼，却偏偏与公主长得像，于是被迫代替她去往敌国和亲。 传闻我的和亲对象是个残疾皇子，可洞房花烛夜当晚...
    茶点故事阅读 42,399评论 2赞 342

推荐阅读更多精彩内容