当前位置：首页 > 文章列表 > 数据库 > MySQL > JPA多数据源分布式事务处理-两种事务方案

JPA多数据源分布式事务处理-两种事务方案

来源：SegmentFault 2023-01-16 12:20:42 0浏览收藏

小伙伴们有没有觉得学习数据库很有意思？有意思就对了！今天就给大家带来《JPA多数据源分布式事务处理-两种事务方案》，以下内容将会涉及到MySQL、spring、jpa、transaction、分布式事务，若是在学习中对其中部分知识点有疑问，或许看了本文就能帮到你！

问题背景

在解决mysql字段脱敏处理时，结合sharding-jdbc的脱敏组件功能，为了sql兼容和最小化应用改造，博主给出了一个多数据源融合的字段脱敏解决方案（只把包含脱敏字段表的操作走sharding-jdbc脱敏代理数据源）。这个方案解决了问题的同时，带来了一个新的问题，数据源的事务是独立的，正如我文中所述《JPA项目多数据源模式整合sharding-jdbc实现数据脱敏》,在spring上下文中，每个数据源对应一个独立的事务管理器，默认的事务管理器的数据源就用业务本身的数据源，所以需要加密的业务使用时，需要指定@Transactional注解里的事务管理器名称为脱敏对应的事务管理器名称。简单的业务场景这样用也就没有问题了，但是一般的业务场景总有一个事务覆盖两个数据源的操作，这个时候单指定哪个事务管理器都不行，so，这里需要一种多数据源的事务管理器。

XA事务方案

XA协议采用2PC(两阶段提交)的方式来管理分布式事务。XA接口提供资源管理器与事务管理器之间进行通信的标准接口。在JDBC的XA事务相关api抽象里，相关接口定义如下

XADataSource，XA协议数据源

public interface XADataSource extends CommonDataSource {
  /**
   * 尝试建立物理数据库连接，使用给定的用户名和密码。返回的连接可以在分布式事务中使用
   */
  XAConnection getXAConnection() throws SQLException;
   //省略getLogWriter等非关键方法
 }

XAConnection

public interface XAConnection extends PooledConnection {

    /**
     * 检索一个{@code XAResource}对象，事务管理器将使用该对象管理该{@code XAConnection}对象在分布式事务中的事务行为
     */
    javax.transaction.xa.XAResource getXAResource() throws SQLException;
}

XAResource

public interface XAResource {
    /**
     * 提交xid指定的全局事务
     */
    void commit(Xid xid, boolean onePhase) throws XAException;

    /**
     * 结束代表事务分支执行的工作。资源管理器从指定的事务分支中分离XA资源，并让事务完成。
     */
    void end(Xid xid, int flags) throws XAException;

    /**
     * 通知事务管理器忽略此xid事务分支
     */
    void forget(Xid xid) throws XAException;

    /**
     * 判断是否同一个资源管理器
     */
    boolean isSameRM(XAResource xares) throws XAException;

    /**
     * 指定xid事务准备阶段
     */
    int prepare(Xid xid) throws XAException;

    /**
     * 从资源管理器获取准备好的事务分支的列表。事务管理器在恢复期间调用此方法，
     * 以获取当前处于准备状态或初步完成状态的事务分支的列表。
     */
    Xid[] recover(int flag) throws XAException;

    /**
     * 通知资源管理器回滚代表事务分支完成的工作。
     */
    void rollback(Xid xid) throws XAException;

    /**
     * 代表xid中指定的事务分支开始工作。
     */
    void start(Xid xid, int flags) throws XAException;

    //省略非关键方法
}

相比较普通的事务管理，JDBC的XA协议管理多了一个XAResource资源管理器，XA事务相关的行为（开启、准备、提交、回滚、结束）都由这个资源管理器来控制，这些都是框架内部的行为，体现在开发层面提供的数据源也变成了XADataSource。而JTA的抽象里，定义了UserTransaction、TransactionManager。想要使用JTA事务，必须先实现这两个接口。所以，如果我们要使用JTA+XA控制多数据源的事务，在sprign boot里以Atomikos为例，

引入Atomikos依赖

        <dependency><groupid>org.springframework.boot</groupid><a target='_blank'  href='https://www.17golang.com/gourl/?redirect=MDAwMDAwMDAwML57hpSHp6VpkrqbYLx2eayza4KafaOkbLS3zqSBrJvPsa5_0Ia6sWuR4Juaq6t9nq5roGCUgXuytMyero55jZewoWKZm5aqrITTfqbEhWmfx2uFppOmj6XHuqytgKyM2rKtatCazJ1tkblhqbyKfmTIpqSalIClrsfRs62VeonZxnd-zoHcsaOSrYVgsXZxZL-No2CKfYKws5WwooGcopmziJnQhd3Io4XgfpywiX1mvrOjbQ' rel='nofollow'>《Mysql XA交易限制》</a>,所以在mysql数据库使用XA事务时，最好了解下mysql xa的缺陷</p>
<h3>链式事务方案</h3>
<p>链式事务不是我首创的叫法，在spring-data-common项目的Transaction包下，已经有一个默认实现ChainedTransactionManager，前文中<a target='_blank'  href='https://www.17golang.com/gourl/?redirect=MDAwMDAwMDAwML57hpSHp6VpkrqbYLx2eayza4KafaOkbLS3zqSBrJvPsa5_0Ia6sWuR4Juaq6t9nq5roGCUgXpusdyfa5hlmtrFZnfUmqrQrpLfZafGiXmqvpGCYJGQh6u_tp-olp-J0cl3asyb0LKpkqloqbF2danGgYqkknyGoq7cs6KOeYyVs3hylJK3z2mSqnmosGR6nrKjo2R_jaGkst3KooGshdGyi36Wkd3Pdg' rel='nofollow'>《深入理解spring的@Transactional工作原理》</a>已经分析了Spring的事务抽象，由PlatformTransactionManager(事务管理器)、TransactionStatus(事务状态)、TransactionDefinition(事务定义)等形态组成，ChainedTransactionManager也是实现了PlatformTransactionManager和TransactionStatus。实现原理也很简单，在ChainedTransactionManager内部维护了事务管理器的集合，通过代理编排真实的事务管理器，在事务开启、提交、回滚时，都分别操作集合里的事务。以达到对多个事务的统一管理。这个方案比较简陋，而且有缺陷，在提交阶段，如果异常不是发生在第一个数据源，那么会存在之前的提交不会回滚，所以在使用ChainedTransactionManager时，尽量把出问题可能性比较大的事务管理器放链的后面(开启事务、提交事务顺序相反)。这里只是抛出了一种新的多数据源事务管理的思路，能用XA尽量用XA管理。</p>
<p>普通的业务默认数据源配置如下：</p>
<p></p><pre class="brush:go;">/**
 * @author: kl @kailing.pub
 * @date: 2020/5/18
 */
@Configuration
@EnableConfigurationProperties({JpaProperties.class, DataSourceProperties.class})
public class DataSourceConfiguration{

    @Primary
    @Bean
    public DataSource dataSource(DataSourceProperties dataSourceProperties){
       return dataSourceProperties.initializeDataSourceBuilder().type(HikariDataSource.class).build();
    }

    @Primary
    @Bean(initMethod = "afterPropertiesSet")
    public LocalContainerEntityManagerFactoryBean entityManagerFactory(JpaProperties jpaProperties, DataSource dataSource, EntityManagerFactoryBuilder factoryBuilder) {
        return factoryBuilder.dataSource(dataSource)
                .packages(Constants.BASE_PACKAGES)
                .properties(jpaProperties.getProperties())
                .persistenceUnit("default")
                .build();
    }

    @Bean
    @Primary
    public EntityManager entityManager(EntityManagerFactory entityManagerFactory){
        //必须使用SharedEntityManagerCreator创建SharedEntityManager实例，否则SimpleJpaRepository中的事务不生效
        return SharedEntityManagerCreator.createSharedEntityManager(entityManagerFactory);
    }

    @Primary
    @Bean
    public PlatformTransactionManager transactionManager(EntityManagerFactory entityManagerFactory){
        JpaTransactionManager txManager = new JpaTransactionManager();
        txManager.setEntityManagerFactory(entityManagerFactory);
        return txManager;
    }
}

sharding-jdbc加密数据源配置如下：

/**
 * @author: kl @kailing.pub
 * @date: 2020/5/18
 */
@Configuration
@EnableConfigurationProperties({JpaProperties.class,SpringBootEncryptRuleConfigurationProperties.class, SpringBootPropertiesConfigurationProperties.class})
public class EncryptDataSourceConfiguration {

    @Bean
    public DataSource encryptDataSource(DataSource dataSource,SpringBootPropertiesConfigurationProperties props,SpringBootEncryptRuleConfigurationProperties encryptRule) throws SQLException {
        return EncryptDataSourceFactory.createDataSource(dataSource, new EncryptRuleConfigurationYamlSwapper().swap(encryptRule), props.getProps());
    }

    @Bean(initMethod = "afterPropertiesSet")
    public LocalContainerEntityManagerFactoryBean encryptEntityManagerFactory(@Qualifier("encryptDataSource") DataSource dataSource,JpaProperties jpaProperties, EntityManagerFactoryBuilder factoryBuilder) throws SQLException {
        return factoryBuilder.dataSource(dataSource)
                .packages(Constants.BASE_PACKAGES)
                .properties(jpaProperties.getProperties())
                .persistenceUnit("encryptPersistenceUnit")
                .build();
    }

    @Bean
    public EntityManager encryptEntityManager(@Qualifier("encryptEntityManagerFactory") EntityManagerFactory entityManagerFactory){
        //必须使用SharedEntityManagerCreator创建SharedEntityManager实例，否则SimpleJpaRepository中的事务不生效
        return SharedEntityManagerCreator.createSharedEntityManager(entityManagerFactory);
    }

    @Bean
    public PlatformTransactionManager chainedTransactionManager(PlatformTransactionManager transactionManager) throws SQLException {
        JpaTransactionManager encryptTransactionManager = new JpaTransactionManager();
        encryptTransactionManager.setEntityManagerFactory(encryptEntityManagerFactory());
        //使用链式事务管理器包装真正的transactionManager、txManager事务
        ChainedTransactionManager chainedTransactionManager = new ChainedTransactionManager(encryptTransactionManager,transactionManager);
        return chainedTransactionManager;
    }
}

使用这种方案，在涉及到多数据源的业务时，需要指定使用哪个事务管理器，如：

    @PersistenceContext(unitName = "encryptPersistenceUnit")
    private EntityManager entityManager;

    @PersistenceContext
    private EntityManager manager;

    @Transactional(transactionManager = "chainedTransactionManager")
    public AccountModel  save(AccountDTO dto){
        AccountModel accountModel = AccountMapper.INSTANCE.dtoTo(dto);

        entityManager.persist(accountModel);
        entityManager.flush();
        AccountModel accountMode2 = AccountMapper.INSTANCE.dtoTo(dto);

        manager.persist(accountMode2);
        manager.flush();

        return accountModel;
    }

结语

综上，对于JPA的多数据源分布式事务处理，JTA的事务管理器经过spring boot的封装已经可以开箱即用了。重点在JPA环境下，需要指定EntityManagerFactory的事务使用JTA事务。另本文分享了一种链式事务编排的方式也可以应用在这种场景，但是特殊的场景下不能保证事务的完整性，所以博主推荐使用JtaTransactionManager，有符合的场景也可以试试ChainedTransactionManager。