订单系统中的数据一致性方案及RocketMQ事务消息详解

数据一致性是确保业务操作正确执行的基础，本文将以电商系统为例，详细分析其分布式系统中的一致性问题。订单核心流程：

订单服务 -> 创建订单 -> 库存服务 -> 扣减库存 -> 积分服务 -> 增加积分 -> 仓储服务 -> 通知发货

分布式一致性技术方案

生产中存在两种常用的解决方案：TCC和可靠消息最终一致性。前者要求强一致，后者要求最终一致。

强一致主要用于核心模块，例如交易/订单等。最终一致一般用于边缘模块例如库存，通过mq去通知，保证最终一致性，也可以业务解耦。

TCC:

订单服务、库存服务、积分服务 -> 绑定为一个TCC事务；

撤销订单时，回滚扣减库存和增加积分。

可靠消息最终一致性:

可以去发送一个请求给消息中间件，由中间件保证一定会把消息交给下游的库存服务去扣减库存，仓储服务去通知发货等；

如果这个过程中有消息发送失败，则可靠消息中间件应该保证不停的重试投递消息。

本文重点分析如何利用RocketMQ的事务消息实现最终一致性，TCC事务将在另外一篇文章分享。

事务消息

RocketMQ的事务消息有两个核心概念（流程）：

• Half Message，半消息

暂时不能被 Consumer消费的消息。Producer已经把消息发送到 Broker端，但是此消息的状态被标记为不能投递，处于这种状态下的消息称为半消息。事实上，该状态下的消息会被放在一个叫做 RMQ_SYS_TRANS_HALF_TOPIC的主题下。

当 Producer端对它二次确认后，也就是 Commit之后，Consumer端才可以消费到；那么如果是Rollback，该消息则会被删除，永远不会被消费到。

• 事务状态回查

可能会因为网络原因、应用问题等，导致Producer端一直没有对这个半消息进行确认，那么这时候 Broker服务器会定时扫描这些半消息，主动找Producer端查询该消息的状态。

简而言之，RocketMQ事务消息的实现原理就是基于两阶段提交和事务状态回查，来决定消息最终是提交还是回滚的。

核心流程

结合整个订单接口服务，分为两个支付链路，一个是核心链路（订单业务），一个是非核心链路（wms） 整个流程。

先向RocketMQ发送half msg，然后调用核心链路。核心链路要是返回失败，就会走失败的逻辑：退款，更改订单状态为取消，再给rocketmq发送callback废弃掉刚才的消息。

如果成功，就commit msg让消费者可以消费。如果在等待期间，一直没有callback/commit那么mq就会走回调查询具体的状态。

消费者接收到消息后，消费完成就回复mq一个ack， 如果消费失败了，mq就会重新投递或者换一个服务投递。使用rocketmq的half msg机制，可以实现这一套固定模式的最终一致性。

代码实现

【核心链路-订单、库存、积分】

核心业务流程

【步骤一】：发送事务消息（half msg）

springboot下，RocketMQ的集成还是很简单的，引入rocketmq-spring-boot-starter依赖、添加相关配置后，即可利用RocketMQTemplate的sendMessageInTransaction方法发送消息：

/**
 * 发送事务消息
 *
 * @param topic   topic
 * @param message 消息对象
 */
public void sendMessageInTransaction(String topic, Object message) {
    String transactionId = UUID.randomUUID().toString();
    TransactionSendResult result = this.rocketMQTemplate.sendMessageInTransaction(topic, MessageBuilder.withPayload(message)
            .setHeader(RocketMQHeaders.TRANSACTION_ID, transactionId)
            .build(), message);
}

【步骤二】：broker回调，执行本地事务

消息发送成功之后，系统需要知道RocketMQ的broker是否成功收到了消息，这里主要借助RocketMQTransactionListener注解实现。在成功收到回调后，会触发executeLocalTransaction来执行核心业务（订单、库存、积分等）。

@RocketMQTransactionListener
public class OrderTransactionListener implements RocketMQLocalTransactionListener {

    /**
     * 执行本地事务，即处理核心链路
     * @param msg
     * @param arg
     * @return
     */
    @Override
    public RocketMQLocalTransactionState executeLocalTransaction(Message msg, Object arg) {
        try {
          	// 事务transactionId
            String transactionId = msg.getHeaders().get("rocketmq_TRANSACTION_ID").toString();
            // 本地事务，执行核心链路业务
            String payload = new String((byte[]) msg.getPayload());
            OrderTranscationMesageDTO data = JSONObject.parseObject(payload, OrderTranscationMesageDTO.class);
            orderService.executeCoreBusiness(data.getPayMoney(),data.getOrderDO(), data.getTransactionNo(),data.getPayType(),transactionId);
        } catch (Exception e) {
            log.error("本地事务执行异常:{}事务消息回滚", e.getMessage());
            return RocketMQLocalTransactionState.ROLLBACK;
        }
      	log.info("提交事务消息");
        return RocketMQLocalTransactionState.COMMIT;
    }

    /**
     * 校验本地事务（broker未收到提交或回滚事务消息时主动回查）
     * @param msg
     * @return
     */
    @Override
    public RocketMQLocalTransactionState checkLocalTransaction(Message msg) {
        // 事务transactionId
        String transactionId = msg.getHeaders().get("rocketmq_TRANSACTION_ID").toString();
        // 数据库能查到，说明本地事务执行失败，需要回滚
        if (Objects.isNull(transcationLogDao.getById(transactionId))){
            return RocketMQLocalTransactionState.ROLLBACK;
        }
        return RocketMQLocalTransactionState.COMMIT;
    }

}

执行核心业务的同时，还有一个很重要的环节，即记录事务ID。为什么要记录事务ID呢？假想这样一种情况：我们本地事务即核心的业务都成功执行后，需要提交RocketMQ的事务消息，只有提交后，消息才能被消费者（即非核心业务系统，如仓储）消费，但是如果提交时，网络出现异常，broker一直未收到怎么办呢，这时利用transactionId，也是RocketMA的回查机制了。

/**
  * 核心业务，并记录RocketMQ事务ID
*/
@GlobalTransactional  // seata全局事务
public void executeCoreBusiness(BigDecimal payMoney, AppDerivativeGoodsOrderDO orderDO, String transactionNo, String payType,String transactionId) {
		// 核心业务伪代码
    orderService.execute();
    storeFeign.execute();
    scoreFeign.execute();
    
    // 数据库记录rocket事务消息ID 用于异常情况下的回查
    if (Objects.nonNull(transactionId)){
        //写入事务日志
        TransactionLogDO log = new TransactionLogDO();
        log.setId(transactionId);
        log.setBusiness("order");
        log.setForeignKey(String.valueOf(orderDO.getId()));
        transcationLogDao.save(log);
    }
}

本地事务执行成功之后，记录事务ID，即便提交时，出现网络异常，broker迟迟未收到，也可以利用回查机制，即checkLocalTransaction方法，得知本地事务是否执行成功。

用于记录事务的表结构：

CREATE TABLE `transaction_log` (
  `id` varchar(32) COLLATE utf8mb4_bin NOT NULL DEFAULT '' COMMENT '事务ID',
  `business` varchar(32) COLLATE utf8mb4_bin NOT NULL COMMENT '业务标识',
  `foreign_key` varchar(32) COLLATE utf8mb4_bin NOT NULL COMMENT '对应业务表中的主键',
  PRIMARY KEY (`id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin;

【非核心链路-wms仓储】

【步骤三】：消费消息，处理其他业务

上述的步骤保证了核心业务与RocketMQ消息的一致性，即核心业务成功，消息就一定会被发送到broker。接下来就是非核心业务（如仓储物流）监听消息，通过@RocketMQMessageListener实现：

@RocketMQMessageListener(topic = "order_topic",consumerGroup = "order_group")
public class TestListener implements RocketMQListener {
    @Override
    public void onMessage(MessageExt messageExt) {
        if(messageExt.getReconsumeTimes() >= 3){
            log.error("消息重试已达最大次数，将通知业务人员排查问题。{}",messageExt.getMsgId());
            //消息处理，第3次处理失败后，发送邮件通知人工介入
            sendMail(messageExt.getMsgId());
        }
        // 仓储物流相关业务 
        wmsService.execute(messageExt.getBody());
    }
}

非核心业务，接受不了消息后，再处理相关业务，其实，此时已经与核心业务脱离了关联，因此，不管它成功与否，核心业务都已经完成了，这也为何是最终一致性，而非强一致性。

最终一致性主要依赖的是RocketMQ的重试机制以及补偿处理（比人工干预）。如上述代码中，假若wmsService执行业务过程抛出了异常，即消息消费失败，RocketMQ则会自动重发。默认16次，可以通过配置修改。另外，可以在重试一定次数后，做补偿处理，例如，将执行失败的任务记录在数据库，后续定时任务补偿处理，抑或是像上述代码，发送邮件通知相关人员。

幂等性消费

消息的重发，有可能带来另外一个问题，重复消费。不做处理，就可能导致数据重复插入，仓储系统就可能重复发货。

幂等性：就是用户对于同一操作发起的一次请求或者多次请求的结果是一致的，不会因为多次点击而产生了副作用。

实现幂等性消费的方式有很多种，具体怎么做，根据自己的情况来看。一种常用的方式就是利用redis缓存，在执行操作之前，先到缓存中查询，该操作是否已执行过。

总结

本文重点阐述了基于RocketMQ来实现最终一致性的分布式事务案例。