分布式学习笔记十一：分布式事务模型DTP

亦凉 2022-05-16 15:58 184阅读 0赞

### **一。 DTP简介** ###

X/Open DTP(X/Open Distributed Transaction Processing Reference Model) 是X/Open 这个组织定义的一套分布式事务的标准，也就是了定义了规范和API接口，由厂商进行具体的实现

X/Open DTP 定义了三个组件： AP，TM，RM  
AP(Application Program)：也就是应用程序，可以理解为使用DTP的程序  
RM(Resource Manager)：资源管理器，这里可以理解为一个DBMS系统，或者消息服务器管理系统，应用程序通过资源管理器对资源进行控制。资源必须实现XA定义的接口  
TM(Transaction Manager)：事务管理器，负责协调和管理事务，提供给AP应用程序编程接口（TX协议）以及管理资源管理器  
其中，AP 可以和TM 以及 RM 通信，TM 和 RM 互相之间可以通信，DTP模型里面定义了XA接口，TM 和 RM 通过XA接口进行双向通信，例如:TM通知RM提交事务或者回滚事务，RM把提交结果通知给TM。AP和RM之间则通过RM提供的Native API 进行资源控制，这个没有进行约API和规范，各个厂商自己实现自己的资源控制，比如Oracle自己的数据库驱动程序。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2N6Y251b251bw_size_16_color_FFFFFF_t_70][]

其中在DTP定了以下几个概念：

事务：一个事务是一个完整的工作单元，由多个独立的计算任务组成，这多个任务在逻辑上是原子的。  
全局事务：对于一次性操作多个资源管理器的事务，就是全局事务  
分支事务：在全局事务中，某一个资源管理器有自己独立的任务，这些任务的集合作为这个资源管理器的分支任务  
控制线程：用来表示一个工作线程，主要是关联AP,TM,RM三者的一个线程，也就是事务上下文环境。简单的说，就是需要标识一个全局事务以及分支事务的关系。

分布式事务模型 使用二阶段提交协议2PC(Two-phaseCommit) 实现多数据源事务处理

准备阶段  
事务协调者(事务管理器)给每个参与者(资源管理器)发送Prepare消息，每个参与者要么直接返回失败(如权限验证失败)，要么在本地执行事务，写本地的redo和undo日志，但不提交，到达一种“万事俱备，只欠东风”的状态。

提交阶段  
如果协调者收到了参与者的失败消息或者超时，直接给每个参与者发送回滚(Rollback)消息；否则，发送提交(Commit)消息；参与者根据协调者的指令执行提交或者回滚操作，释放所有事务处理过程中使用的锁资源。(注意:必须在最后阶段释放锁资源)

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2N6Y251b251bw_size_16_color_FFFFFF_t_70 1][]

出现部分资源失败后的处理情况

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2N6Y251b251bw_size_16_color_FFFFFF_t_70 2][]

二阶段提交看起来确实能够提供原子性的操作，但是不幸的事，二阶段提交还是有几个缺点的、

1、同步阻塞问题。执行过程中，所有参与节点都是事务阻塞型的。当参与者占有公共资源时，其他第三方节点访问公共资源不得不处于阻塞状态。  
2、单点故障。由于协调者的重要性，一旦协调者发生故障。参与者会一直阻塞下去。尤其在第二阶段，协调者发生故障，那么所有的参与者还都处于锁定事务资源的状态中，而无法继续完成事务操作。（如果是协调者挂掉，可以重新选举一个协调者，但是无法解决因为协调者宕机导致的参与者处于阻塞状态的问题）  
3、数据不一致。在二阶段提交的阶段二中，当协调者向参与者发送commit请求之后，发生了局部网络异常或者在发送commit请求过程中协调者发生了故障，这回导致只有一部分参与者接受到了commit请求。而在这部分参与者接到commit请求之后就会执行commit操作。但是其他部分未接到commit请求的机器则无法执行事务提交。于是整个分布式系统便出现了数据部一致性的现象。  
4、二阶段无法解决的问题：协调者再发出commit消息之后宕机，而唯一接收到这条消息的参与者同时也宕机了。那么即使协调者通过选举协议产生了新的协调者，这条事务的状态也是不确定的，没人知道事务是否被已经提交。

###    **二。DTP编程模型JTA** ###

JTA(Java Transaction API)是符合X/Open DTP的一个编程模型，事务管理和资源管理器支架也是用了XA协议。

1》TA的构成  
a、高层应用事务界定接口，供事务客户界定事务边界的  
b、X/Open XA协议(资源之间的一种标准化的接口)的标准Java映射，它可以使事务性的资源管理器参与由外部事务管理器控制的事务中  
c、高层事务管理器接口，允许应用程序服务器为其管理的应用程序界定事务的边界

2》JTA的主要接口　  
位于javax.transaction包中  
a、UserTransaction接口：让应用程序得以控制事务的开始、挂起、提交、回滚等。由Java客户端程序或EJB调用。  
b、TransactionManager 接口：用于应用服务器管理事务状态  
c、Transaction接口：用于执行相关事务操作  
d、XAResource接口：用于在分布式事务环境下，协调事务管理器和资源管理器的工作  
e、Xid接口：为事务标识符的Java映射  
注：前3个接口位于Java EE版的类库 javaee.jar 中，Java SE中没有提供！UserTransaction是编程常用的接口  
注意的是JTA只提供了接口，没有具体的实现。  
JTS(Java Transaction Service)是服务OTS的JTA的实现。简单的说JTS实现了JTA接口，并且符合OTS的规范。  
JTA的事务周期可横跨多个JDBC Connection生命周期，对众多Connection进行调度，实现其事务性要求。  
JTA可以处理任何提供符合XA接口的资源。包括：JDBC连接，数据库，JMS，商业对象等等。  
3》JTA实现

JOTM(Java Open Transaction Manager)是ObjectWeb的一个开源JTA实现，它本身也是开源应用程序服务器JOnAS(Java Open Application Server)的一部分，为其提供JTA分布式事务的功能。官网：http://jotm.objectweb.org

Atomikos 是一个为Java平台提供增值服务的并且开源类事务管理器提供 以下是包括在这个开源版本中的一些功能：  
1 全面崩溃 / 重启恢复  
2 兼容标准的SUN公司JTA API  
3 嵌套事务  
4 为XA和非XA提供内置的JDBC适配器  
官网：https://www.atomikos.com/

### **三。Spring集成JTA** ###

因为spring的  
<tx:annotation-driven transaction-manager="springTransactionManager"/>  只能指定一个事务管理器 所以多数据源下 肯定不能使用同一个事务管理器  
必须使用jta事务

1》使用jtom配置jta实现  
模拟环境   
  mysql 下存在表 mymoney 存在用户zs 余额 1000 建表sql

> CREATE DATABASE IF NOT EXISTS dtp;  
> USE dtp;  
> CREATE TABLE mymoney(  
>   id INT PRIMARY KEY AUTO\_INCREMENT,  
>   NAME VARCHAR(20),  
>   lostedmoney INT  
> );  
> INSERT INTO mymoney(id,NAME,lostedmoney) VALUES(1,'zs',1000);

oracle下存在表mymoney 存在用户ls 余额300 建表sql

>   
> oracle下 scott账号下 创建相同表  
> CREATE TABLE mymoney(  
>   id number PRIMARY KEY ,  
>   NAME VARCHAR2(20),  
>   lostedmoney number  
> );  
> INSERT INTO mymoney(id,NAME,lostedmoney) VALUES(2,'ls',300);

模拟zs给ls转账 自然涉及到多数据源分布式事务  
配置spring支持jtom的环境 （使用xapool管理数据源）  
maven依赖：

<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
        xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
        <modelVersion>4.0.0</modelVersion>
        <groupId>cn.et</groupId>
        <artifactId>spring</artifactId>
        <version>0.0.1-SNAPSHOT</version>
        <dependencies>
            <dependency>
                <groupId>org.springframework</groupId>
                <artifactId>spring-context</artifactId>
                <version>4.3.13.RELEASE</version>
            </dependency>
            <dependency>
                <groupId>org.springframework</groupId>
                <artifactId>spring-tx</artifactId>
                <version>4.3.13.RELEASE</version>
            </dependency>
            <dependency>
                <groupId>org.springframework</groupId>
                <artifactId>spring-aop</artifactId>
                <version>4.3.13.RELEASE</version>
            </dependency>
            <dependency>
                <groupId>org.springframework</groupId>
                <artifactId>spring-jdbc</artifactId>
                <version>4.3.13.RELEASE</version>
            </dependency>
            
            <dependency>
                <groupId>jotm</groupId>
                <artifactId>jotm</artifactId>
                <version>2.0.10</version>
            </dependency>
            
            <dependency>
                <groupId>org.ow2.jotm</groupId>
                <artifactId>jotm-core</artifactId>
                <version>2.2.3</version>
            </dependency>
            
            <dependency>
                <groupId>org.ow2.jotm</groupId>
                <artifactId>jotm-datasource</artifactId>
                <version>2.2.3</version>
            </dependency>
            
            <dependency>
                <groupId>org.ow2.jotm</groupId>
                <artifactId>jotm-standalone</artifactId>
                <version>2.2.3</version>
            </dependency>
            
            <dependency>
                <groupId>javax.resource</groupId>
                <artifactId>javax.resource-api</artifactId>
                <version>1.7</version>
            </dependency>
            
            <dependency>
                <groupId>com.experlog</groupId>
                <artifactId>xapool</artifactId>
                <version>1.5.0</version>
            </dependency>
            
            <dependency>
                <groupId>mysql</groupId>
                <artifactId>mysql-connector-java</artifactId>
                <version>5.1.44</version>
            </dependency>
            <dependency>
                <groupId>com.jslsolucoes</groupId>
                <artifactId>ojdbc6</artifactId>
                <version>11.2.0.1.0</version>
            </dependency>
        </dependencies>
    </project>

添加spring配置文件 配置 两个数据源 并且将两个支持xa的数据源 绑定到一个资源管理器 UserTranscation中

<?xml version="1.0" encoding="UTF-8"?>  
    <beans xmlns="http://www.springframework.org/schema/beans"  
        xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" 
        xmlns:aop="http://www.springframework.org/schema/aop"  
        xmlns:tx="http://www.springframework.org/schema/tx"
        xmlns:context="http://www.springframework.org/schema/context"  
        xsi:schemaLocation="    
            http://www.springframework.org/schema/beans     
            http://www.springframework.org/schema/beans/spring-beans-4.0.xsd    
            http://www.springframework.org/schema/tx     
            http://www.springframework.org/schema/tx/spring-tx-4.0.xsd    
            http://www.springframework.org/schema/aop  
            http://www.springframework.org/schema/aop/spring-aop-4.0.xsd    
            http://www.springframework.org/schema/context     
            http://www.springframework.org/schema/context/spring-context-4.0.xsd    
    ">  
      <context:component-scan base-package="cn"></context:component-scan>
      <bean id="txCurrent" class="org.objectweb.jotm.Current"></bean>
     
      
      <bean id="xaMysqlDataSource" class="org.enhydra.jdbc.pool.StandardXAPoolDataSource" destroy-method="shutdown">
         <property name="dataSource">
              
              <bean class="org.enhydra.jdbc.standard.StandardXADataSource" destroy-method="shutdown">
                  <property name="driverName" value="com.mysql.jdbc.Driver"></property>
                <property name="url" value="jdbc:mysql://localhost:3306/dtp"></property>
                
                
                <property name="transactionManager" ref="txCurrent"></property>
              </bean>
               
         </property>
         <property name="user" value="root"></property>
         <property name="password" value="123456"></property>
      </bean>
       <bean id="mysqlJdbcTemplate" class="org.springframework.jdbc.core.JdbcTemplate">  
            <property name="dataSource" ref="xaMysqlDataSource" />  
       </bean>  
       
       
       <bean id="xaOracleDataSource" class="org.enhydra.jdbc.pool.StandardXAPoolDataSource" destroy-method="shutdown">
         <property name="dataSource">
              
              <bean class="org.enhydra.jdbc.standard.StandardXADataSource" destroy-method="shutdown">
                  <property name="driverName" value="oracle.jdbc.OracleDriver"></property>
                <property name="url" value="jdbc:oracle:thin:@localhost:1521:orcl"></property>
                
                <property name="transactionManager" ref="txCurrent"></property>
              </bean>
         </property>
         <property name="user" value="scott"></property>
         <property name="password" value="tiger"></property>
      </bean>
       <bean id="oracleJdbcTemplate" class="org.springframework.jdbc.core.JdbcTemplate">  
            <property name="dataSource" ref="xaOracleDataSource" />  
       </bean>  
       
       
       
       <bean id="springTransactionManager" class="org.springframework.transaction.jta.JtaTransactionManager">  
            <property name="userTransaction" ref="txCurrent" />  
       </bean>  
       <tx:annotation-driven transaction-manager="springTransactionManager"/>  
    </beans>

添加zs的数据处理类

package cn.et.dao;
     
    import javax.annotation.Resource;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.jdbc.core.JdbcTemplate;
    import org.springframework.stereotype.Repository;
    import org.springframework.transaction.annotation.Transactional;
    @Transactional
    @Repository
    public class UserZsDao {
        @Autowired
        private UserLsDao userLsDao;
        @Resource(name="mysqlJdbcTemplate")
        JdbcTemplate mysqlJdbcTemplate;
        public void zsMinus(int money){
            //zs的mysql扣款
            String sql="update MYMONEY set lostedmoney=lostedmoney-"+money+" where id=1";
            mysqlJdbcTemplate.execute(sql);
            //调用ls的oracle加钱
            userLsDao.lsAdd(money);
            //模拟出现异常 查看数据两个操作数据是否都回滚
            String a=null;
            a.toCharArray();
        }
    }

处理ls的数据处理类

package cn.et.dao;
     
    import javax.annotation.Resource;
     
    import org.springframework.jdbc.core.JdbcTemplate;
    import org.springframework.stereotype.Repository;
    import org.springframework.transaction.annotation.Transactional;
    @Transactional
    @Repository
    public class UserLsDao {
        
        @Resource(name="oracleJdbcTemplate")
        JdbcTemplate oracleJdbcTemplate;
        
        public void lsAdd(int money){
            String sql="update MYMONEY set lostedmoney=lostedmoney+"+money+" where id=2";
            oracleJdbcTemplate.execute(sql);
        }
    }

添加测试类

package cn.et.dao;
     
    import org.springframework.context.support.ClassPathXmlApplicationContext;
     
    public class Test {
     
        public static void main(String[] args) {
            ClassPathXmlApplicationContext cpxac=new ClassPathXmlApplicationContext("spring.xml");
            UserZsDao userZsDao=(UserZsDao)cpxac.getBean("userZsDao");
            userZsDao.zsMinus(100);
            cpxac.close();
        }
     
    }

运行后 发现出现异常正常回滚

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2N6Y251b251bw_size_16_color_FFFFFF_t_70]: /images/20220503/2861b001a76d4e33a4a24e19ad716b51.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2N6Y251b251bw_size_16_color_FFFFFF_t_70 1]: /images/20220503/e68dd4f6da5742b5a915a7575cfdafa4.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2N6Y251b251bw_size_16_color_FFFFFF_t_70 2]: /images/20220503/451304815a5b4fa2ba174978e9706f6f.png