分布式事务解决方案

作者: 麦大大吃不胖 | 来源:发表于2020-12-27 10:53 被阅读0次

by shihang.mai

1. 事件表+定时任务+MQ

分布式事务-定时任务+MQ+事件表

事件表核心表结构

字段	备注
Id	主键
order_type	事件类型，例如派单
state	状态
content	发送到mq的数据，包括本表事件id

加入定时任务，是为了做补偿机制。过了零点，扫面前一天的数据，重新投放到mq即可。

当工单模块消费mq时，创建一样的一条记录到工单系统的事件表。即两个事件表一样的。这样为了做幂等处理。

需要做幂等原因:

当工单系统完成后返回ack，mq挂了，那么就会出现状态已完成，但是mq还是未消费成功。mq还会重试发送到工单系统

2. TXLCN-LCN

用在本身带事务的场景,如mysql

l-lock
c:confirm
n:notify

2.1 原理

lcn组件架构图

lcn组件

lcn原理图如下，摘自tx-lcn官网

lcn官方原理

图中第1 2 3步是第1阶段，第4步是第2阶段(lcn中的n,执行提交或者回滚)

lcn步骤

2.2 原理

核心类DataSourceAspect

@Aspect
@Component
@Slf4j
public class DataSourceAspect implements Ordered {

   private final TxClientConfig txClientConfig;

   private final DTXResourceWeaver dtxResourceWeaver;

   public DataSourceAspect(TxClientConfig txClientConfig, DTXResourceWeaver dtxResourceWeaver) {
       this.txClientConfig = txClientConfig;
       this.dtxResourceWeaver = dtxResourceWeaver;
   }


   @Around("execution(* javax.sql.DataSource.getConnection(..))")
   public Object around(ProceedingJoinPoint point) throws Throwable {
       return dtxResourceWeaver.getConnection(() -> (Connection) point.proceed());
   }


   @Override
   public int getOrder() {
       return txClientConfig.getResourceOrder();
   }


}

核心类LcnConnectionProxy

public class LcnConnectionProxy implements Connection {
  @Override
    public void setAutoCommit(boolean autoCommit) throws SQLException {
        connection.setAutoCommit(false);
    }

    @Override
    public void commit() throws SQLException {
        //connection.commit();
    }

    @Override
    public void rollback() throws SQLException {
        //connection.rollback();
    }

    @Override
    public void close() throws SQLException {
        //connection.close();
    }
}

利用aop，在getConnection的时候做一个环绕，返回一个Connection，这个Connection是LcnConnectionProxy。直接接管了原来的datasource的Connection。
在原本连接close阶段假释放,将请求和连接关系保存起来Map<请求Id,Connection>。那么进行第2阶段时就可以找回之前的操作的连接了。

代理datasource，保持请求和连接的对应

2.3 补偿

补偿机制，用到了redis。做了标记(那些Tx-client是异常的),做记录(通知的具体事项或者需要执行的sql)

2.4 核心步骤

TX-manager
mysql执行sql：
        jar包下有对应的创表sql
添加jar:
        <dependency>
            <groupId>com.codingapi.txlcn</groupId>
            <artifactId>txlcn-tm</artifactId>
            <version>5.0.2.RELEASE</version>
        </dependency>

        <dependency>
            <groupId>com.codingapi.txlcn</groupId>
            <artifactId>txlcn-tc</artifactId>
            <version>5.0.2.RELEASE</version>
        </dependency>
        <dependency>
            <groupId>com.codingapi.txlcn</groupId>
            <artifactId>txlcn-txmsg-netty</artifactId>
            <version>5.0.2.RELEASE</version>
        </dependency>
添加配置:
        tx-lcn.manager.admin-key=xxx
        tx-lcn.logger.enabled=true
        tx-lcn.logger.driver-class-name=com.mysql.cj.jdbc.Driver
        tx-lcn.logger.jdbc-url=jdbc:mysql://localhost:3306/tx-manager?characterEncoding=UTF-8&serverTimezone=Asia/Shanghai
        tx-lcn.logger.username=xxx
        tx-lcn.logger.password=xxx
添加注解:
        启动类上@EnableTransactionManagerServer

TX-client
添加jar：
        <dependency>
            <groupId>com.codingapi.txlcn</groupId>
            <artifactId>txlcn-tc</artifactId>
            <version>5.0.2.RELEASE</version>
        </dependency>

        <dependency>
            <groupId>com.codingapi.txlcn</groupId>
            <artifactId>txlcn-txmsg-netty</artifactId>
            <version>5.0.2.RELEASE</version>
        </dependency>
添加配置:
        tx-lcn:
          client:
            manager-address: 127.0.0.1:8070
添加注解:
        启动类@EnableDistributedTransaction
        业务类@LcnTransaction

3. TXLCN-TCC

用在本身不带事务的场景，如redis

try
检查网络连接、检查业务、隔离资源
confirm
使用try阶段隔离的资源，进行真正事务提交。因为tcc引入补偿机制:重试和超时。因为有重试机制，必须幂等。而幂等操作肯定要一个唯一ID，即事务ID，相同事务ID，相同的事，即可以进行幂等操作。
cancel
回滚

3.1 核心步骤

/**
* TX-manager TX-client的添加jar、配置、启动类注解都与lcn一致
*/
//业务类
    @Transactional(rollbackFor = Exception.class)
    @TccTransaction
    public String add(@RequestBody TblOrder bean){

        
        return "xxx";
    }

    public String confirmAdd(TblOrder bean){

        System.out.println("xxx ");
        return "xxx";
    }

    //用来保存第1阶段操作过的id，其中key为(机器-方法-时间戳)等能唯一表示的key
    private static Map<String,Integer> maps = new HashMap<>();

    public String cancelAdd(TblOrder bean){
        //自己做回滚逻辑
        return "xxx";
    }

4. seata

它有4种模式，分别是AT、TCC、SAGA、XA模式

seata中角色

TC(事务协调者)
TM(事务发起者，实际也是RM)
RM(资源管理者)

4.1 AT模式

以下有3个库,分别模拟正常流程和异常流程

库	表
seata-server	global_table, lock_table , branch_table
a	yw, undo_log
b	yw, undo_log

seata-角色.png

正常流程

请求到达TM,TM向TC注册一个全局事务，获得一个全局锁,TC向seata-server的global_table插入一条数据
在TM执行自己业务逻辑时，向TC注册一个分支事务,TC向branch_table和lock_table各插入一条数据，TM获得一个本地锁，然后将业务数据插入a库的yw表和对应的undolog插入a库的undolog表，然后释放本地锁(不是删除库记录，这是容许其他线程竞争)
TM调用RM，RM向TC注册一个分支事务，TC向branch_table和lock_table各插入一条数据，RM获得一个本地锁，然后将业务数据插入b库的yw表和对应的undolog插入b库的undolog表，然后释放本地锁(不是删除库记录，这是容许其他线程竞争)
TM向TC发送commit,TC删除global_table、lock_table、branch_table记录,TC通知TM和RM提交，即通知它们分别删除自己的undo_log表记录

异常流程

请求到达TM,TM向TC注册一个全局事务，获得一个全局锁,TC向seata-server的global_table插入一条数据
在TM执行自己业务逻辑时，向TC注册一个分支事务,TC向branch_table和lock_table各插入一条数据，TM获得一个本地锁，然后将业务数据插入a库的yw表和对应的undolog插入a库的undolog表，然后释放本地锁(不是删除库记录，这是容许其他线程竞争)
TM调用RM，RM向TC注册一个分支事务，TC向branch_table和lock_table各插入一条数据，RM获得一个本地锁，然后将业务数据插入b库的yw表和对应的undolog插入b库的undolog表，然后释放本地锁(不是删除库记录，这是容许其他线程竞争)
如果此时调用链上还有一个RM，那么重复3的步骤，但是其中发生了异常
TM向TC发送roll back,TC通知TM和所有RM回滚，即通知它们分别执行自己的undo_log表记录，在执行undo_log前需要获取本地锁。然后TC删除global_table、lock_table、branch_table记录

4.2 TCC

4.2.1 空回滚

try未执行，执行cancel
解决:加事务控制表

字段	含义
tx_id	全局事务id
branch_id	分支id
state	状态(try、confirm、cancel)

执行cancel时,检查是否有try的记录，没的话进行空回滚

4.2.2 幂等

多次执行conform、cancel
解决:加事务控制表

字段	含义
tx_id	全局事务id
branch_id	分支id
state	状态(事务初始化、已提交、已回滚)

在执行前先检查库记录即可。执行过就不执行

4.2.3 悬挂

执行cancel在try之前
解决:加事务控制表

字段	含义
tx_id	全局事务id
branch_id	分支id
state	状态(事务初始化、已提交、已回滚)

cnacel的时候，如果原来没记录，执行空方法，再向里面插入一条已回滚记录，try执行的时候发现有已回滚记录，空try

网友评论

本文标题：分布式事务解决方案

本文链接：https://www.haomeiwen.com/subject/ldjinktx.html

延伸阅读

深度阅读

您也可以注册成为美文阅读网的作者，发表您的原创作品、分享您的心情！