分布式事务常见的集中解决方案

一、前言

1. 事务

事务是由一组操作构成的可靠的独立的工作单元，事务具备ACID的特性，即原子性、一致性、隔离性和持久性。

2. 本地事务

单机环境下的一致性解决方案就是本地事务,
本地事务的优点就是支持严格的ACID特性,高效，可靠，状态可以只在资源管理器中维护，而且应用编程模型简单。

3. 全局事务

区别于本地事务,常用语分布式环境下的事务管理。可以理解成,对各个单机的事务进行协调管理。要提交一起提交,要回滚一起会馆

二、方案

1. 最大努力通知

可以利用mq的ack机制。使用场景插入流水信息时候,数据不需要立刻处理,放到MQ中慢慢处理

系统A本地事务执行完后，发送一个消息到MQ
有一专门消费MQ的最大努力通知服务，会消费MQ,接着调用系统B的接口。
如果成功就消息消费成功
如果失败消息消费失败

2. 本地消息表

使用场景: 菜鸟tob仓库占用库存时候,会将数据插入到表中,然后定时扫描失败的,重新去发送消息。
缺点是B如果处理成功了,但是没有把状态改成功,导致A重复去发消息

A系统在本地事务里面同时,插入一条数据到本地消息表,然后将消息发送出去
B系统受到消息后,进行处理自己的事务,如果处理成功了就将A消息表里面的状态改为成功。
A定时扫描消息表,把没有成功的重新发出消息。当然B要做好幂等操作,防止重复发消息

3. 消息最终一致性

最终一致性就不是说成功一起成功,失败一起失败。而是说一个成功,运行另外一个稍微慢一点。只要最后成功就行了。

利用RocketMQ事务消息。

A系统先发送一个 prepared 消息到MQ。MQ先做保存,不立马给消费方
A系统事务执行成功,然后发一个确认消息,此时MQ收到了确认消息,才会把 prepared 发给消费方
如果A执行失败,就可以直接回滚掉前面发的 prepared 消息,此时MQ就不会把消息发给消费方
B系统如果失败了,会自动重试。

可以看出来这个和本地消息表其实差不多,不过只是A系统去扫描消息表,而是都交给了消息队列来处理了。

import org.apache.rocketmq.client.producer.LocalTransactionState;
import org.apache.rocketmq.client.producer.TransactionListener;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;

import java.util.concurrent.ConcurrentHashMap;

public class OrderTransactionListenerImpl implements TransactionListener {
true
trueprivate ConcurrentHashMap<String, Integer> countHashMap = new ConcurrentHashMap<>();
true
trueprivate final static int MAX_COUNT = 5;


true@Override
    public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
        // 
    	String bizUniNo = msg.getUserProperty("bizUniNo"); // 从消息中获取业务唯一ID。
    	// 将bizUniNo入库，表名：t_message_transaction,表结构  bizUniNo(主键),业务类型。
        return LocalTransactionState.UNKNOW;
    }

    //问你是否要确认消息。可以从消息中拿到业务信息,进行二次查询,来确定是否确认还是回滚
    @Override
    public LocalTransactionState checkLocalTransaction(MessageExt msg) {
        Integer status = 0;
        // 从数据库查查询t_message_transaction表，如果该表中存在记录，则提交，
        String bizUniNo = msg.getUserProperty("bizUniNo"); // 从消息中获取业务唯一ID。
        // 然后t_message_transaction 表，是否存在bizUniNo，如果存在，则返回COMMIT_MESSAGE，
        // 不存在，则记录查询次数，未超过次数，返回UNKNOW，超过次数，返回ROLLBACK_MESSAGE
        if(query(bizUniNo) > 0 ) {
        	return LocalTransactionState.COMMIT_MESSAGE;
        }
        
        return rollBackOrUnown(bizUniNo);
    }
    
    public int query(String bizUniNo) {
    	return 1; //select count(1) from t_message_transaction a where a.biz_uni_no=#{bizUniNo}
    }
    
    public LocalTransactionState rollBackOrUnown(String bizUniNo) {
    	Integer num = countHashMap.get(bizUniNo);
    	if(num != null &&  ++num > MAX_COUNT) {
    		countHashMap.remove(bizUniNo);
    		return LocalTransactionState.ROLLBACK_MESSAGE;
    	}
    	return LocalTransactionState.UNKNOW;
    	
    }
 
}