记一次分布式事务可靠消息最终一致性方案

场景：单体应用加一个@Transactional 即可通过spring事务控制事务。
原理是什么呢？
1. 带注解的对象是spring AOP-在bean的生命周期，初始化阶段处理过的。是一个代理对象
2. 作为一个代理对象，可以在业务方法前边getcon begin ->反射invoke 到业务方法->业务方法后边 commit 。
3. 基于以上两点，完美的在一个连接内控制了事务。
那么问题来了，当一个事务操作出现在多个不同的服务中。无法再控制在一个连接中，事务该怎么控制呢

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

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

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

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

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

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

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

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

解决方案：

本地消息表方案

假如2个业务 1.新增用户 2.存储积分 12要控制原子性
begin transaction；
	//1.新增用户
	//2.存储积分消息日志
commit transation;  1和2控制在同一个事务中

定时扫描消息日志表 -> mq 可靠传输 —> 手动ack,消费成功，修改消息日志表状态为处理成功。

RocketMQ事务消息方案
RocketMQ 事务消息设计则主要是为了解决 Producer 端的消息发送与本地事务执行的原子性问题

代码设计可参考：https://gitee.com/islibin/rocketmq-jta-demo.git

RocketMQ 靠 executeLocalTransaction 和 checkLocalTransaction 事务回查保证了发mq的原子性。

@Component
@Slf4j
public class OrderTransactionListener implements TransactionListener {

    @Autowired
    private OrderService orderService;

    @Autowired
    private TransactionLogService transactionLogService;

    @Override
    public LocalTransactionState executeLocalTransaction(Message message, Object o) {
        log.info("开始执行本地事务....");
        LocalTransactionState state = null;
        try {
            String body = new String(message.getBody());
            TOrder tOrder = JSONObject.parseObject(body, TOrder.class);
            // 执行本地事务
            orderService.insertOrder(tOrder, message.getTransactionId());
            state = LocalTransactionState.COMMIT_MESSAGE;
            log.info("本地事务已提交。{}", message.getTransactionId());
        }catch (Exception e){
            log.error("执行本地事务失败。{}", e.getMessage());
            state = LocalTransactionState.ROLLBACK_MESSAGE;
        }
        return state;
    }

    @Override
    public LocalTransactionState checkLocalTransaction(MessageExt messageExt) {
        log.info("开始回查本地事务....");
        LocalTransactionState state = null;
        String transId = messageExt.getTransactionId();
        // 如果本地事务存在，则事务提交成功
        TransactionLog transactionLog = transactionLogService.getById(transId);
        if (null != transactionLog){
            state = LocalTransactionState.COMMIT_MESSAGE;
        }else{
            state = LocalTransactionState.ROLLBACK_MESSAGE;
        }
        return state;
    }
}

记一次 分布式事务可靠消息最终一致性方案

Java相关栏目本月热门文章

记一次分布式事务可靠消息最终一致性方案