分布式事务解决方案之可靠消息最终一致性（四）

一 什么是可靠消息最终一致性事务

　　可靠消息最终一致性方案是指当事务发起方执行完成本地事务后并发出一条消息，事务参与方(消息消费者)一定能够接收消息并处理事务成功，此方案强调的是只要消息发给事务参与方最终事务要达到一致。

此方案是利用消息中间件完成，如下图：

　　

　　事务发起方（消息生产方）将消息发给消息中间件，事务参与方从消息中间件接收消息，事务发起方和消息中间件之间，事务参与方（消息消费方）和消息中间件之间都是通过网络通信，由于网络通信的不确定性会导致分布式事务问题。

 

　　理解：可靠消息最终一致性，消息从事务的发起方到事务的参与方，这个消息必须可靠，转账必须由张三发给李四，并且这个消息强调最终一致性。即张三本地事务执行扣钱之后，要保证这个消息一定要发给李四，李四接到这个消息之后就能加钱了，如果李四加钱失败，张三是不能回滚的，此时需要消息消费重试或者人工参与给李四无比要加钱成功；

　　可靠消息最终一致性分为两个部分理解，可靠消息指消息从张三传给李四，从事务的发起方传给事务的参与方，这个过程是可靠的；最终一致性是指张三执行本地事务，扣钱后消息一定会发给事务参与方李四，之后李四就要保证最终一致性，无论如何就要把钱加上。

因此可靠消息最终一致性方案要解决以下几个问题：

1）本地事务与消息发送的原子性问题

　　本地事务与消息发送的原子性问题即：事务发起方在本地事务执行成功后消息必须发出去，否则就丢弃消息。即实现本地事务和消息发送的原子性，要么都成功，要么都失败。本地事务与消息发送的原子性问题是实现可靠消息最终一致性方案的关键问题。

先来尝试下这种操作，先发送消息，再操作数据库：

begin transaction；
//1.发送MQ
//2.数据库操作
commit transation;

这种情况下无法保证数据库操作与发送消息的一致性，因为可能发送消息成功，数据库操作失败。

你立马想到第二种方案，先进行数据库操作，再发送消息：

begin transaction；
//1.数据库操作
//2.发送MQ
commit transation;

这种情况下貌似没有问题，如果发送 MQ 消息失败，就会抛出异常，导致数据库事务回滚。但如果是超时异常，数据库回滚，但MQ其实已经正常发送了，同样会导致不一致。

2）事务参与方接收消息的可靠性

事务参与方必须能够从消息队列接收到消息，如果接收消息失败可以重复接收消息；

3）消息重复消费的问题

由于网络2 的存在，若某一个消费节点超时但是消费成功，此时消息中间件会重复投递此消息，就导致了消息的重复消费。【事务参与方接收到消息，但是没有确认 ack 机制，此时消息中间件会认为事务参与方没有消费消息，就会进行重试，需要保证事务参与方的幂等性】

要解决消息重复消费的问题就要实现事务参与方的方法幂等性。

二 解决方案

上节讨论了可靠消息最终一致性事务方案需要解决的问题，本节讨论具体的解决方案。

2.1 本地消息表方案  阿里一面：如何保障消息100%投递成功、消息幂等性？

本地消息表这个方案最初是 eBay 提出的，此方案的核心是通过本地事务保证数据业务操作和消息的一致性，然后通过定时任务将消息发送至消息中间件，待确认消息发送给消费方成功再将消息删除。

下面以注册送积分为例来说明：

下例共有两个微服务交互，用户服务和积分服务，用户服务负责添加用户，积分服务负责增加积分。

　　

交互流程如下： 

1）用户注册

 用户服务在本地事务新增用户和增加 ”积分消息日志“。（用户表和消息表通过本地事务保证一致） 下边是伪代码

begin transaction；
//1.新增用户
//2.存储积分消息日志
commit transation;

这种情况下，本地数据库操作与存储积分消息日志处于同一个事务中，本地数据库操作与记录消息日志操作具备原子性。

2）定时任务扫描日志

如何保证将消息发送给消息队列呢？

经过第一步消息已经写到消息日志表中，可以启动独立的线程，定时对消息日志表中的消息进行扫描并发送至消息中间件，在消息中间件反馈发送成功后删除该消息日志，否则等待定时任务下一周期重试。

3）消费消息

如何保证消费者一定能消费到消息呢？

这里可以使用 MQ 的 ack（即消息确认）机制，消费者监听 MQ，如果消费者接收到消息并且业务处理完成后向 MQ 发送 ack（即消息确认），此时说明消费者正常消费消息完成，MQ 将不再向消费者推送消息，否则消费者会不断重试向消费者来发送消息。

积分服务接收到”增加积分“消息，开始增加积分，积分增加成功后向消息中间件回应 ack，否则消息中间件将重复投递此消息。

由于消息会重复投递，积分服务的”增加积分“功能需要实现幂等性。【若事务参与方消费超时但是消费成功，此时消息中间件 MQ 会重复投递此消息，就导致了消息的重复消费】

【幂等性：同一个操作，携带参数一致，不管执行多少次，都会得到相同的答案！】

2.2 RocketMQ 事务消息方案

RocketMQ 是一个来自阿里巴巴的分布式消息中间件，于 2012 年开源，并在 2017 年正式成为 Apache 顶级项目。据了解，包括阿里云上的消息产品以及收购的子公司在内，阿里集团的消息产品全线都运行在 RocketMQ 之上，并且最近几年的双十一大促中，RocketMQ 都有抢眼表现。Apache RocketMQ 4.3之后的版本正式支持事务消息，为分布式事务实现提供了便利性支持。

RocketMQ 事务消息设计则主要是为了解决 Producer 端的消息发送与本地事务执行的原子性问题，RocketMQ 的设计中 broker 与 producer 端的双向通信能力，使得 broker 天生可以作为一个事务协调者存在；而 RocketMQ 本身提供的存储机制为事务消息提供了持久化能力；RocketMQ 的高可用机制以及可靠消息设计则为事务消息在系统发生异常时依然能够保证达成事务的最终一致性。

在 RocketMQ 4.3 后实现了完整的事务消息，实际上其实是对本地消息表的一个封装，将本地消息表移动到了 MQ  【建议一定要手动画一画这张流程图】

     　

执行流程如下：内部，解决 Producer 端的消息发送与本地事务执行的原子性问题。

为方便理解我们还以注册送积分的例子来描述整个流程。

Producer 即 MQ 发送方，本例中是用户服务，负责新增用户。MQ 订阅方即消息消费方，本例中是积分服务，负责新增积分。

1）Producer 发送事务消息

　　Producer （MQ发送方）发送事务消息至 MQ Server，MQ Server 将消息状态标记为 Prepared（预备状态），注意此时这条消息消费者（MQ订阅方）是无法消费到的【未消费状态】

　　本例中，Producer 发送 ”增加积分消息“ 到 MQ Server；

2）MQ Server 回应消息发送成功

　　MQ Server 接收到 Producer 发送给的消息则回应发送成功，表示 MQ 已接收到消息；

3）Producer 执行本地事务

　　Producer 端执行业务代码逻辑，通过本地数据库事务控制。本例中，Producer 执行添加用户操作；

4）消息投递

　　若 Producer 本地事务执行成功，则自动向 MQServer 发送 commit 消息，MQServer 接收到 commit 消息后将”增加积分消息“ 状态标记为可消费，此时 MQ 订阅方（积分服务）即正常消费消息；

　　若 Producer 本地事务执行失败，则自动向 MQServer 发送 rollback 消息，MQ  Server 接收到 rollback 消息后  将删除”增加积分消息“ ；

　　MQ 订阅方（积分服务）消费消息，消费成功则向 MQ 回应 ack，否则将重复接收消息【需要实现幂等性操作】。这里 ack 默认自动回应，即程序执行正常则自动回应 ack。

5）事务回查

　　如果执行 Producer 端本地事务过程中，执行端挂掉，或者超时，MQ Server 将会不停的询问同组的其他 Producer 来获取事务执行状态，这个过程叫 事务回查。MQ Server 会根据事务回查结果来决定是否投递消息【消息状态从未消费状态变为可消费状态】；

以上主干流程已由 RocketMQ 实现，对用户侧来说，用户需要分别实现本地事务执行以及本地事务回查方法，因此只需关注本地事务的执行状态即可。

　　RoacketMQ 提供 RocketMQLocalTransactionListener 接口：

public interface RocketMQLocalTransactionListener {
/**
‐发送prepare消息成功此方法被回调，该方法用于执行本地事务
‐@param msg 回传的消息，利用transactionId即可获取到该消息的唯一Id
‐@param arg 调用send方法时传递的参数，当send时候若有额外的参数可以传递到send方法中，这里能获取到
‐@return 返回事务状态，COMMIT：提交 ROLLBACK：回滚 UNKNOW：回调
*/
RocketMQLocalTransactionState executeLocalTransaction(Message msg, Object arg);

三 RocketMQ 实现可靠消息最终一致性事务

1）业务说明

本实例通过 RocketMQ 中间件实现可靠消息最终一致性分布式事务，模拟两个账户的转账交易过程；

两个账户在分别在不同的银行(张三在 bank1、李四在 bank2)，bank1、bank2 是两个微服务。交易过程是，张三给李四转账指定金额。

上述交易步骤，张三扣减金额与给 bank2 发转账消息，两个操作必须是一个整体性的事务。

　　

 2）本示例程序技术架构如下：

　　

　　交互流程如下： 

　　1、Bank1 向 MQ   Server 发送转账消息 2、Bank1 执行本地事务，扣减金额；

　　3、Bank2 接收消息，执行本地事务，添加金额。

3）dtx-txmsg-demo-bank1

dtx-txmsg-demo-bank1实现如下功能：  

1、张三扣减金额，提交本地事务。

2、向MQ发送转账消息。

 a）Dao

@Mapper
@Component
public interface AccountInfoDao {
    
    @Update("update account_info set account_balance=account_balance+#{amount} where account_no=#{accountNo}")
    int updateAccountBalance(@Param("accountNo") String accountNo, @Param("amount") Double amount);

    @Select("select * from account_info where where account_no=#{accountNo}")
    AccountInfo findByIdAccountNo(@Param("accountNo") String accountNo);

    @Select("select count(1) from de_duplication where tx_no = #{txNo}")
    int isExistTx(String txNo);

    @Insert("insert into de_duplication values(#{txNo},now());")
    int addTx(String txNo);

}

b）Service 给 MQ Server 发送消息，扣减金额

@Service
@Slf4j
public class AccountInfoServiceImpl implements AccountInfoService {

    @Autowired
    private AccountInfoDao accountInfoDao;

    @Autowired
    private RocketMQTemplate rocketMQTemplate;

    @Override
    public void sendUpdateAccountBalance(AccountChangeEvent accountChangeEvent) {
        // 将 accountChangeEvent 转成 json
        JSONObject jsonObject = new JSONObject();
        String jsonString = jsonObject.put("accountChange", accountChangeEvent).toString();
        // 生成 Message 类型
        Message<String> message = MessageBuilder.withPayload(jsonString).build();
        /**
         * 发送一条事务消息
         * String txProducerGroup   生产组
         * String destination       topic 主题
         * Message<?> message       发送的信息
         * Object arg               参数
         */
        rocketMQTemplate.sendMessageInTransaction("producer_group_txmsg_bank1", "topic_txmsg", message, null);

    }

    @Transactional
    @Override
    public void doUpdateAccountBalance(AccountChangeEvent accountChangeEvent) {
        // 幂等性校验
        if (accountInfoDao.isExistTx(accountChangeEvent.getTxNo()) > 0) {
            log.info("扣减金额已经执行完毕，不可重复执行");
            return;
        }
        // 扣减金额
        accountInfoDao.updateAccountBalance(accountChangeEvent.getAccountNo(), accountChangeEvent.getAmount() * -1);
        // 添加事务日志
        accountInfoDao.addTx(accountChangeEvent.getTxNo());
    }
}

c）RocketMQLocalTransactionListener 编写 RocketMQLocalTransactionListener 接口实现类，实现执行本地事务和事务回查两个方法。

// 需要指定事务的生产组，当事务调用方发送给MQ消息时，设置了生产组；当 MQ 接收到消息时，会给事务发送方发送成功接收到的信号，
// 此时根据生产组来回调本地事务执行方法，然后事务发送方执行完本地事务，给MQ发送 Commit 或者 Rollback 消息，MQ会将转账的消息状态修改为可消费状态，供李四事务参与方进行消息
@Component
@Slf4j
@RocketMQTransactionListener(txProducerGroup = "producer_group_txmsg_bank1")
public class ProducerTxmsgListener implements RocketMQLocalTransactionListener {

    @Autowired
    private AccountInfoService accountInfoService;
    @Autowired
    private AccountInfoDao accountInfoDao;

    // 事务消息发送后的回调方法，当消息发送给 MQ 成功，此方法被回调
    @Override
    public RocketMQLocalTransactionState executeLocalTransaction(Message msg, Object arg) {
        try {
            // 解析Message，转换成 AccountChangeEvent
            String messageString = new String((byte[])msg.getPayload());
            JSONObject jsonObject = JSONObject.parseObject(messageString);
            String accountChange = jsonObject.getString("accountChange");
            // 将 accountChange（String 类型） 转换成 AccountChangeEvent 对象
            AccountChangeEvent accountChangeEvent = JSONObject.parseObject(accountChange, AccountChangeEvent.class);

            accountInfoService.doUpdateAccountBalance(accountChangeEvent);
            // 当返回 RocketMQLocalTransactionState.COMMIT 类型，自动向 MQ 发送 commit 消息，MQ 将消息的状态转为可消费
            return RocketMQLocalTransactionState.COMMIT;

        } catch (Exception e) {
            log.info("张三转账【事务发送方】执行本地事务失败, errorMsg = {}", e.getMessage());
            return RocketMQLocalTransactionState.ROLLBACK;
        }
    }

    // 事务状态回查，查询是否扣减金额
    @Override
    public RocketMQLocalTransactionState checkLocalTransaction(Message msg) {
        // 解析Message，转换成 AccountChangeEvent
        String messageString = new String((byte[])msg.getPayload());
        JSONObject jsonObject = JSONObject.parseObject(messageString);
        String accountChange = jsonObject.getString("accountChange");
        // 将 accountChange（String 类型） 转换成 AccountChangeEvent 对象
        AccountChangeEvent accountChangeEvent = JSONObject.parseObject(accountChange, AccountChangeEvent.class);

        if (accountInfoDao.isExistTx(accountChangeEvent.getTxNo()) > 0) {   // 幂等性校验
            return RocketMQLocalTransactionState.COMMIT;
        } else {
            return RocketMQLocalTransactionState.UNKNOWN;
        }
    }
}

d）Controller 发送转账请求

@RestController
@Slf4j
public class AccountInfoController {

    @Autowired
    private AccountInfoService accountInfoService;

    @GetMapping(value = "/transfer")
    public String transfer(@RequestParam("accountNo") String accountNo, @RequestParam("amount") Double amount) {
        // 创建一个本地事务 ID，作为消息内容发送到 MQ
        String txNo = UUID.randomUUID().toString();
        AccountChangeEvent accountChangeEvent = new AccountChangeEvent(accountNo, amount, txNo);

        // 发送消息
        accountInfoService.sendUpdateAccountBalance(accountChangeEvent);
        return "转账成功";
    }

}

4）dtx-txmsg-demo-bank2

dtx-txmsg-demo-bank2 需要实现如下功能：   

1、监听MQ，接收消息；

2、接收到消息增加账户金额。

a）Dao

@Mapper
@Component
public interface AccountInfoDao {
    @Update("update account_info set account_balance=account_balance+#{amount} where account_no=#{accountNo}")
    int updateAccountBalance(@Param("accountNo") String accountNo, @Param("amount") Double amount);

    @Select("select count(1) from de_duplication where tx_no = #{txNo}")
    int isExistTx(String txNo);

    @Insert("insert into de_duplication values(#{txNo},now());")
    int addTx(String txNo);

}

b）MQ 监听类，消费消息

@Component
@Slf4j
@RocketMQMessageListener(consumerGroup = "consumer_group_txmsg_bank2", topic = "topic_txmsg")   // 需要监听指定的 topic，和MQ发送方的 topic 需要指定一样
public class TxmsgConsumer implements RocketMQListener<String> {

    @Autowired
    private AccountInfoService accountInfoService;

    // 接收消息
    @Transactional
    @Override
    public void onMessage(String message) {
        log.info("开始消费消息: {}", message);
        // 解析消息
        JSONObject jsonObject = JSONObject.parseObject(message);
        String accountChange = jsonObject.getString("accountChange");
        // 转成 AccountChangeEvent 对象，需要设置账户为李四账户
        AccountChangeEvent accountChangeEvent = JSONObject.parseObject(accountChange, AccountChangeEvent.class);
        accountChangeEvent.setAccountNo("2");

        // 更新本地账户，增加金额
        accountInfoService.addAccountInfoBalance(accountChangeEvent);
    }
}

c）Service 增加李四账户金额

@Service
@Slf4j
public class AccountInfoServiceImpl implements AccountInfoService {

    @Autowired
    private AccountInfoDao accountInfoDao;

    @Transactional
    @Override
    public void addAccountInfoBalance(AccountChangeEvent accountChangeEvent) {
        log.info("bank2更新本地账号，账号：{},金额：{}",accountChangeEvent.getAccountNo(),accountChangeEvent.getAmount());
        if (accountInfoDao.isExistTx(accountChangeEvent.getTxNo()) > 0) {       // 幂等校验
            log.info("李四账户金额添加执行完毕，无需重复执行, txNo = {}", accountChangeEvent.getTxNo());
            return;
        }
        // 增加金额
        accountInfoDao.updateAccountBalance(accountChangeEvent.getAccountNo(), accountChangeEvent.getAmount());

        // 添加事务记录，用户幂等性操作
        accountInfoDao.addTx(accountChangeEvent.getTxNo());
    }

}

四 小结

可靠消息最终一致性就是保证消息从生产方经过消息中间件传递到消费方的一致性，本案例使用了 RocketMQ 作为消息中间件，RocketMQ 主要解决了两个功能：

　　1、本地事务与消息发送的原子性问题；

　　2、事务参与方接收消息的可靠性。

可靠消息最终一致性事务适合执行周期长且实时性要求不高的场景。引入消息机制后，同步的事务操作变为基于消息执行的异步操作, 避免了分布式事务中的同步阻塞操作的影响，并实现了两个服务的解耦，数据达到最终一致性。

　　　　

　　每天进步一点点......