JPA+Jta-atomikos实现分布式事务

左手的ㄟ右手 2022-12-10 07:10 135阅读 0赞

我在上一章节中介绍了分布式事务的相关理论知识，本章节中我会在SpringBoot中结合JPA，整合jta-atomikos来实现多数据源环境下的分布式事务。

## **一. 多数据源环境下分布式事务代码实现** ##

**代码环境:**

*  Spring Boot2.2.5
 *  mysql-connector-java8.0.11

### **创建数据库对应的表** ###

**db1数据库中的goods表:**

![fe5326421357236152745874a3b35a06.png][]

**db4数据库中的user表:**

![2f5afbcc2d8c50ffd44fcc08eee6e57a.png][]

### **1. 创建web项目** ###

我们按照之前的经验，创建一个web程序，并将之改造成Spring Boot项目，具体过程略。

![e5f451262dde3c40ba8738a945c586fa.png][]

### **2. 添加依赖包** ###

<dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-jta-atomikos</artifactId>
    </dependency>
    
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-data-jpa</artifactId>
    </dependency>
    
    <dependency>
        <groupId>mysql</groupId>
        <artifactId>mysql-connector-java</artifactId>
        
        <version>8.0.11</version>
    </dependency>
    
    <dependency>
        <groupId>com.alibaba</groupId>
        <artifactId>druid</artifactId>
        <version>1.1.12</version>
    </dependency>

要注意mysql-connector-java的版本不要太高，否则可能会产生一些异常。

### **3. 创建application.yml配置文件** ###

spring:
      jpa:
        database: mysql
        show-sql: true
        hibernate:
          ddl-auto: update
        database-platform: org.hibernate.dialect.MySQL57Dialect

### **4. 创建两个数据源对应的实体类** ###

**db1数据源的实体类Goods**

package com.yyg.boot.entity.db1;
    
    import lombok.Data;
    
    import javax.persistence.*;
    
    /**
     * @Author 一一哥Sun
     * @Date Created in 2020/4/3
     * @Description db1中的商品表
     */
    @Entity
    @Table(name = "goods")
    @Data
    public class Goods {
    
        @Id
        @GeneratedValue(strategy=GenerationType.IDENTITY)
        private Long id;
    
        private String name;
    
    }

**db4数据源的实体类User**

package com.yyg.boot.entity.db2;
    
    import lombok.Data;
    import lombok.ToString;
    
    import javax.persistence.*;
    import java.io.Serializable;
    import java.util.Date;
    
    /**
     * @Author 一一哥Sun
     * @Date Created in 2020/4/3
     * @Description db4中的用户表
     */
    @Entity
    @Table(name = "user")
    @Data
    @ToString
    public class User implements Serializable {
    
        @Id
        @GeneratedValue(strategy=GenerationType.IDENTITY)
        private Long id;
    
        @Column
        private String username;
    
        @Column
        private Date birthday;
    
        @Column
        private String sex;
    
        @Column
        private String address;
    
    }

### **5. 创建2个数据源仓库类** ###

**db1数据源仓库类GoodsRepository**

package com.yyg.boot.dao.db01;
    
    import com.yyg.boot.entity.db1.Goods;
    import org.springframework.data.jpa.repository.JpaRepository;
    import org.springframework.data.jpa.repository.JpaSpecificationExecutor;
    import org.springframework.stereotype.Repository;
    
    /**
     * @Author 一一哥Sun
     * @Date Created in 2020/4/3
     * @Description Description
     */
    @Repository
    public interface GoodsRepository extends JpaRepository<Goods, Long>,JpaSpecificationExecutor<Goods> {
    }

**db4数据源仓库类UserRepository**

package com.yyg.boot.dao.db02;
    
    import com.yyg.boot.entity.db2.User;
    import org.springframework.data.jpa.repository.JpaRepository;
    import org.springframework.data.jpa.repository.JpaSpecificationExecutor;
    import org.springframework.stereotype.Repository;
    
    /**
     * @Author 一一哥Sun
     * @Date Created in 2020/4/3
     * @Description Description
     */
    @Repository
    public interface UserRepository extends JpaRepository<User, Long>,JpaSpecificationExecutor<User> {
    }

### **6. 配置数据源及事务管理器等** ###

这里的配置是最关键的，一般代码出问题都是因为这里有问题。

> *这里针对两个数据库，分别设置两个数据源goodsDataSource与userDataSource。这里的DataSource必须是XADataSource类型的数据源才行，并且利用AtomikosDataSourceBean讲数据源配置进去。*  
> *还有分别设置实体类管理工厂goodsEntityManagerFactory与userEntityManagerFactory。*  
> *再分别设置事务管理器goodsTransactionManager与userTransactionManager。*  
> *还要创建一个用来整合多个事务管理器的JtaTransactionManager，负责回滚到多个数据源中。*  
> *但要注意区分是否是Primary配置。*

**db1数据源配置类GoodsDataSourceConfig**

package com.yyg.boot.config;
    
    import com.alibaba.druid.pool.xa.DruidXADataSource;
    import com.atomikos.jdbc.AtomikosDataSourceBean;
    import org.springframework.boot.autoconfigure.orm.jpa.JpaProperties;
    import org.springframework.boot.orm.jpa.EntityManagerFactoryBuilder;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    import org.springframework.context.annotation.Primary;
    import org.springframework.data.jpa.repository.config.EnableJpaRepositories;
    import org.springframework.orm.jpa.JpaTransactionManager;
    import org.springframework.orm.jpa.LocalContainerEntityManagerFactoryBean;
    import org.springframework.transaction.PlatformTransactionManager;
    
    import javax.annotation.Resource;
    import javax.sql.DataSource;
    import java.util.Objects;
    
    /**
     * @Author 一一哥Sun
     * @Date Created in 2020/4/18
     * @Description 商品数据源配置类
     */
    @Configuration
    @EnableJpaRepositories(basePackages = {"com.yyg.boot.dao.db01"}, entityManagerFactoryRef = "goodsEntityManagerFactory", transactionManagerRef = "goodsTransactionManager")
    public class GoodsDataSourceConfig {
    
        @Resource
        private JpaProperties jpaProperties;
    
        /**
         * 第一个数据源
         */
        @Primary
        @Bean(name = "goodsDataSource")
        public DataSource goodsDataSource() {
            DruidXADataSource druidXADataSource = new DruidXADataSource();
            druidXADataSource.setUrl("jdbc:mysql://localhost:3306/db1?characterEncoding=utf-8&useSSL=false&serverTimezone=UTC");
            druidXADataSource.setUsername("root");
            druidXADataSource.setPassword("syc");
            druidXADataSource.setDefaultAutoCommit(false);
            AtomikosDataSourceBean atomikosDataSourceBean = new AtomikosDataSourceBean();
            atomikosDataSourceBean.setXaDataSource(druidXADataSource);
            atomikosDataSourceBean.setUniqueResourceName("goodsDataSource");
            atomikosDataSourceBean.setPoolSize(5);
    
            return atomikosDataSourceBean;
        }
    
        /**
         * 商品实体类管理工厂
         */
        @Primary
        @Bean(name = "goodsEntityManagerFactory")
        public LocalContainerEntityManagerFactoryBean goodsEntityManagerFactory(EntityManagerFactoryBuilder builder) {
            return builder.dataSource(goodsDataSource())
                    .properties(jpaProperties.getProperties())
                    //设置实体类所在位置：类或包
                    .packages("com.yyg.boot.entity.db1")
                    //持久化单元名称
                    .persistenceUnit("goodsPersistenceUnit")
                    .build();
        }
    
        /**
         * 商品事务管理器
         */
        @Primary
        @Bean(name = "goodsTransactionManager")
        public PlatformTransactionManager goodsTransactionManager(EntityManagerFactoryBuilder builder) {
            return new JpaTransactionManager(Objects.requireNonNull(goodsEntityManagerFactory(builder).getObject()));
        }
    
    }

**db4数据源配置类GoodsDataSourceConfig**

package com.yyg.boot.config;
    
    import com.alibaba.druid.pool.xa.DruidXADataSource;
    import com.atomikos.jdbc.AtomikosDataSourceBean;
    import org.springframework.boot.autoconfigure.orm.jpa.JpaProperties;
    import org.springframework.boot.orm.jpa.EntityManagerFactoryBuilder;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    import org.springframework.data.jpa.repository.config.EnableJpaRepositories;
    import org.springframework.orm.jpa.JpaTransactionManager;
    import org.springframework.orm.jpa.LocalContainerEntityManagerFactoryBean;
    import org.springframework.transaction.PlatformTransactionManager;
    
    import javax.annotation.Resource;
    import javax.sql.DataSource;
    import java.util.Objects;
    
    /**
     * @Author 一一哥Sun
     * @Date Created in 2020/4/18
     * @Description 商品数据源配置类
     */
    @Configuration
    @EnableJpaRepositories(basePackages = {"com.yyg.boot.dao.db02"},entityManagerFactoryRef = "userEntityManagerFactory",transactionManagerRef = "userTransactionManager")
    public class UserDataSourceConfig {
    
        /**
         * 自动注入jpa配置
         */
        @Resource
        private JpaProperties jpaProperties;
    
        @Bean(name = "userDataSource")
        public DataSource db4DataSource() {
            DruidXADataSource druidXADataSource = new DruidXADataSource();
            druidXADataSource.setUrl("jdbc:mysql://localhost:3306/db4?characterEncoding=utf-8&useSSL=false&serverTimezone=UTC");
            druidXADataSource.setUsername("root");
            druidXADataSource.setPassword("syc");
            druidXADataSource.setDefaultAutoCommit(false);
    
            AtomikosDataSourceBean atomikosDataSourceBean = new AtomikosDataSourceBean();
            atomikosDataSourceBean.setXaDataSource(druidXADataSource);
            atomikosDataSourceBean.setUniqueResourceName("userDataSource");
            atomikosDataSourceBean.setPoolSize(5);
    
            return atomikosDataSourceBean;
        }
    
        /**
         * 将数据源、连接池、以及其他配置策略进行封装返回给事务管理器
         * 自动装配时当出现多个Bean候选者时，被注解为@Primary的Bean将作为首选者，否则将抛出异常
        */
        @Bean(name = "userEntityManagerFactory")
        public LocalContainerEntityManagerFactoryBean userEntityManagerFactory(EntityManagerFactoryBuilder builder){
          return builder.dataSource(db4DataSource())
                  .properties(jpaProperties.getProperties())
                  //设置实体类所在位置：类或包
                  .packages("com.yyg.boot.entity.db2")
                  //持久化单元名称
                  .persistenceUnit("userPersistenceUnit")
                  .build();
        }
    
        /**
          * 返回数据源的事务管理器
          */
        @Bean(name = "userTransactionManager")
        public PlatformTransactionManager userTransactionManager(EntityManagerFactoryBuilder builder){
            return new JpaTransactionManager(Objects.requireNonNull(userEntityManagerFactory(builder).getObject()));
        }
    
    }

**创建用来整合多数据源的事务管理器JtaTransactionManager**

package com.yyg.boot.config;
    
    import com.atomikos.icatch.jta.UserTransactionImp;
    import com.atomikos.icatch.jta.UserTransactionManager;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    import org.springframework.context.annotation.Primary;
    import org.springframework.transaction.jta.JtaTransactionManager;
    
    import javax.transaction.UserTransaction;
    
    /**
     * @Author 一一哥Sun
     * @Date Created in 2020/4/19
     * @Description Description
     */
    @Configuration
    public class JtaTransactionManagerConfig {
    
        @Primary
        @Bean(name = "jtaTransactionManager")
        public JtaTransactionManager regTransactionManager () {
            UserTransactionManager userTransactionManager = new UserTransactionManager();
            UserTransaction userTransaction = new UserTransactionImp();
            return new JtaTransactionManager(userTransaction, userTransactionManager);
        }
    
    }

### **7. 创建Service及其实现类** ###

> *注意:*  
> *此处一定要通过@Transactional注解来明确指明所用的事务管理器,否则当多数据源时可能只能执行一个数据里的增删改操作，且此处应该设置Spring的事务传播属性为propagation =Propagation.NEVER，不使用单个数据源自己的事务，统一交由jtaTransactionManager的事务来管理。*

**定义GoodsService接口**

package com.yyg.boot.service;
    
    import com.yyg.boot.entity.db1.Goods;
    
    import java.util.List;
    
    /**
     * @Author 一一哥Sun
     * @Date Created in 2020/4/7
     * @Description Description
     */
    public interface GoodsService {
    
        List<Goods> findAll();
    
        Goods addGoods(Goods goods);
    
    }

**GoodsServiceImpl实现**

package com.yyg.boot.service.impl;
    
    import com.yyg.boot.dao.db01.GoodsRepository;
    import com.yyg.boot.entity.db1.Goods;
    import com.yyg.boot.service.GoodsService;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.stereotype.Service;
    import org.springframework.transaction.annotation.Propagation;
    import org.springframework.transaction.annotation.Transactional;
    import org.springframework.transaction.interceptor.TransactionAspectSupport;
    
    import java.util.List;
    
    /**
     * @Author 一一哥Sun
     * @Date Created in 2020/4/7
     * @Description Description
     *
     * 注意:此处一定要通过@Transactional注解来明确指明所用的事务管理器,否则当多数据源时可能只能执行一个数据里的增删改操作.
     * 且此处应该设置Spring的事务传播属性为propagation =Propagation.NEVER,不使用单个数据源自己的事务,统一交由jtaTransactionManager的事务来管理.
     */
    @Service
    @Transactional(transactionManager = "goodsTransactionManager",rollbackFor = Exception.class,propagation =Propagation.NEVER)
    public class GoodsServiceImpl implements GoodsService {
    
        @Autowired
        private GoodsRepository goodsRepository;
    
        @Override
        public List<Goods> findAll() {
            return goodsRepository.findAll();
        }
    
        @Override
        public Goods addGoods(Goods goods) {
            try {
                return goodsRepository.save(goods);
            }catch (Exception e){
                //强制手动事务回滚
                TransactionAspectSupport.currentTransactionStatus().setRollbackOnly();
            }
            return null;
        }
    
    }

**定义UserService接口**

package com.yyg.boot.service;
    
    import com.yyg.boot.entity.db2.User;
    
    import java.util.List;
    
    /**
     * @Author 一一哥Sun
     * @Date Created in 2020/4/7
     * @Description Description
     */
    public interface UserService {
    
        List<User> findAll();
    
        User addUser(User user);
    
    }

**UserServiceImpl实现**

package com.yyg.boot.service.impl;
    
    import com.yyg.boot.dao.db02.UserRepository;
    import com.yyg.boot.entity.db2.User;
    import com.yyg.boot.service.UserService;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.stereotype.Service;
    import org.springframework.transaction.annotation.Propagation;
    import org.springframework.transaction.annotation.Transactional;
    import org.springframework.transaction.interceptor.TransactionAspectSupport;
    
    import java.util.List;
    
    /**
     * @Author 一一哥Sun
     * @Date Created in 2020/4/7
     * @Description Description
     * 注意:此处一定要通过@Transactional注解来明确指明所用的事务管理器,否则当多数据源时可能只能执行一个数据里的增删改操作.
     * 且此处应该设置Spring的事务传播属性为propagation =Propagation.NEVER,不使用单个数据源自己的事务,统一交由jtaTransactionManager的事务来管理.
     */
    @Service
    @Transactional(transactionManager = "userTransactionManager",rollbackFor = Exception.class,propagation =Propagation.NEVER)
    public class UserServiceImpl implements UserService {
    
        @Autowired
        private UserRepository userRepository;
    
        @Override
        public List<User> findAll() {
            return userRepository.findAll();
        }
    
        @Override
        public User addUser(User user) {
            try {
                return userRepository.save(user);
            }catch (Exception e){
                //强制手动事务回滚
                TransactionAspectSupport.currentTransactionStatus().setRollbackOnly();
            }
            return null;
        }
    
    }

### **8. 创建Controller进行测试** ###

package com.yyg.boot.web;
    
    import com.yyg.boot.entity.db1.Goods;
    import com.yyg.boot.entity.db2.User;
    import com.yyg.boot.service.GoodsService;
    import com.yyg.boot.service.UserService;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.http.ResponseEntity;
    import org.springframework.transaction.annotation.Transactional;
    import org.springframework.web.bind.annotation.GetMapping;
    import org.springframework.web.bind.annotation.RestController;
    
    import java.util.Date;
    
    /**
     * @Author 一一哥Sun
     * @Date Created in 2020/4/3
     * @Description Description
     */
    @RestController
    public class TransactionController {
    
        @Autowired
        private UserService userService;
    
        @Autowired
        private GoodsService goodsService;
    
        @Transactional(value = "jtaTransactionManager", rollbackFor = Exception.class)
        @GetMapping("/add")
        public ResponseEntity addTest() {
            Goods goods = new Goods();
            goods.setId(44L);
            goods.setName("iPhone se");
            goodsService.addGoods(goods);
    
            User user = new User();
            user.setSex("男");
            user.setUsername("一一哥");
            user.setAddress("上海");
            user.setBirthday(new Date());
            userService.addUser(user);
    
            return ResponseEntity.ok("添加成功");
        }
    
    //在该接口中故意制造一个除零异常
    @Transactional(value = "jtaTransactionManager", rollbackFor = Exception.class)
        @GetMapping("/add2")
        public ResponseEntity addTest2() {
            Goods goods = new Goods();
            goods.setId(44L);
            goods.setName("iPhone se");
            goodsService.addGoods(goods);
    
            User user = new User();
            user.setSex("男");
            user.setUsername("一一哥");
            user.setAddress("上海");
            user.setBirthday(new Date());
            userService.addUser(user);
    
            int i=10/0;
    
            return ResponseEntity.ok("添加成功");
        }
    
    }

### **9. 创建入口类** ###

package com.yyg.boot;
    
    import org.springframework.boot.SpringApplication;
    import org.springframework.boot.autoconfigure.SpringBootApplication;
    import org.springframework.transaction.annotation.EnableTransactionManagement;
    
    /**
     * @Author 一一哥Sun
     * @Date Created in 2020/4/18
     * @Description Description
     * (exclude = {DataSourceAutoConfiguration.class,DataSourceTransactionManagerAutoConfiguration.class})
     */
    @SpringBootApplication
    //手动进行事务管理
    @EnableTransactionManagement
    public class TransactionApplication {
    
        public static void main(String[] args){
            SpringApplication.run(TransactionApplication.class,args);
        }
    
    }

### **10. 完整代码结构** ###

![9b0339e55de2a88b9003492f5c9fa41a.png][]

![3bab788cd9b33613933b6bfb6092cd18.png][]

### **11. 启动项目进行测试** ###

**测试add接口**

![8da93dc3acca7ac224b273bfe0c76152.png][]

可以看到，当插入数据没有异常发生时，两个数据库都可以正常插入数据。

![0472e20a618630d5210956586da5c809.png][]

![bc03ab49b89a731833787a64a5d0ce0b.png][]

**测试add2接口**

![4f7d81a8b66d623b07de87a95623942b.png][]

因为我在add2接口中故意制造了一个除零异常，此时代码会发生异常。

![f4995b74d22859062cadfa75e3ab578f.png][]

我们再看数据库中，会发现并没有产生新数据，还是和之前的数据一样，说明多数据源环境下，利用jta-atomikos实现了分布式事务的代码实现。

![0472e20a618630d5210956586da5c809.png][]

![bc03ab49b89a731833787a64a5d0ce0b.png][]

[fe5326421357236152745874a3b35a06.png]: /images/20221123/3efc83440b28451d9975b9c44e14b289.png
[2f5afbcc2d8c50ffd44fcc08eee6e57a.png]: /images/20221123/d0107cd99c204a4fa30fe00540cf3b81.png
[e5f451262dde3c40ba8738a945c586fa.png]: /images/20221123/a73b6d8435e24309b61eb6508e3dbbfc.png
[9b0339e55de2a88b9003492f5c9fa41a.png]: /images/20221123/9a8b71f18a744bc1b64c68a2d7f932be.png
[3bab788cd9b33613933b6bfb6092cd18.png]: /images/20221123/75ddce256c004556b33bce7062554178.png
[8da93dc3acca7ac224b273bfe0c76152.png]: /images/20221123/c2c560c7d514470b94571c736373dd84.png
[0472e20a618630d5210956586da5c809.png]: /images/20221123/144805b574aa4d579dc840db02830d4c.png
[bc03ab49b89a731833787a64a5d0ce0b.png]: /images/20221123/c67784a269014c2b876d7e11f2d35866.png
[4f7d81a8b66d623b07de87a95623942b.png]: /images/20221123/007a530632484498ab586344d296f947.png
[f4995b74d22859062cadfa75e3ab578f.png]: /images/20221123/7199b8993e3743ef8f0dc2eb856bb76d.png