Java Spring Boot 분산 트랜잭션을 구현하는 방법은 무엇입니까?

머리말

먼저 분산 트랜잭션이란 무엇입니까? 예를 들어 비즈니스 로직을 실행할 때 A 데이터 소스에서 데이터 변경을 수행할 때 각각 두 단계로 작업합니다. 소스, B 데이터 소스가 실행될 때 런타임 예외가 발생하면 데이터 소스 B의 작업을 롤백해야 하며, 이러한 상황은 결제 서비스에서 자주 발생합니다. 티켓 구매 서비스가 결국 결제에 실패하면 이전 작업을 모두 롤백해야 합니다. 이전 작업이 여러 데이터 소스에 분산된 경우 이는 일반적인 분산 트랜잭션 롤백입니다. Java로 분산 트랜잭션 해결 솔루션은 JTA(Java Transaction API)입니다. springboot는 Atomikos 또는 Bitronix 솔루션을 공식적으로 제공합니다.

실제로 대부분의 회사에서는 분산 트랜잭션을 구현하기 위해 메시지 대기열을 사용합니다.

이 기사는 springboot 환경에서 Atomikos +mysql+mybatis+tomcat/jetty를 통합하는 데 중점을 둡니다.

1. 프로젝트 종속성

pom.xml에atomikos의 springboot 관련 종속성을 추가합니다.

<!--分布式事务-->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-jta-atomikos</artifactId>
</dependency>

입력하려면 클릭하세요. 통합되어 있음을 알 수 있습니다: transactions-jms, transactions-jta, transactions-jdbc, javax.transaction-api</ code> code></p><p>2. 데이터 소스 구성<code>transactions-jms、transactions-jta、transactions-jdbc、javax.transaction-api

二、数据源配置

把数据源的相关配置项单独提炼到一个application.yml中：

注意：

• 这回我们的spring.datasource.type 是com.alibaba.druid.pool.xa.DruidXADataSource；

• spring.jta.transaction-manager-id的值在你的电脑中是唯一的，这个详细请阅读官方文档；

完整的yml文件如下：

spring:
datasource:
type: com.alibaba.druid.pool.xa.DruidXADataSource
druid:

systemDB:
name: systemDB
url: jdbc:mysql://localhost:3306/springboot-mybatis?useUnicode=true&characterEncoding=utf-8
username: root
password: root
# 下面为连接池的补充设置，应用到上面所有数据源中
# 初始化大小，最小，最大
initialSize: 5
minIdle: 5
maxActive: 20
# 配置获取连接等待超时的时间
maxWait: 60000
# 配置间隔多久才进行一次检测，检测需要关闭的空闲连接，单位是毫秒
timeBetweenEvictionRunsMillis: 60000
# 配置一个连接在池中最小生存的时间，单位是毫秒
minEvictableIdleTimeMillis: 30
validationQuery: SELECT 1
validationQueryTimeout: 10000
testWhileIdle: true
testOnBorrow: false
testOnReturn: false
# 打开PSCache，并且指定每个连接上PSCache的大小
poolPreparedStatements: true
maxPoolPreparedStatementPerConnectionSize: 20
filters: stat,wall
# 通过connectProperties属性来打开mergeSql功能；慢SQL记录
connectionProperties: druid.stat.mergeSql=true;druid.stat.slowSqlMillis=5000
# 合并多个DruidDataSource的监控数据
useGlobalDataSourceStat: true

businessDB:
name: businessDB

url: jdbc:mysql://localhost:3306/springboot-mybatis2?useUnicode=true&characterEncoding=utf-8
username: root
password: root
# 下面为连接池的补充设置，应用到上面所有数据源中
# 初始化大小，最小，最大
initialSize: 5
minIdle: 5
maxActive: 20
# 配置获取连接等待超时的时间
maxWait: 60000
# 配置间隔多久才进行一次检测，检测需要关闭的空闲连接，单位是毫秒
timeBetweenEvictionRunsMillis: 60000
# 配置一个连接在池中最小生存的时间，单位是毫秒
minEvictableIdleTimeMillis: 30
validationQuery: SELECT 1
validationQueryTimeout: 10000
testWhileIdle: true
testOnBorrow: false
testOnReturn: false
# 打开PSCache，并且指定每个连接上PSCache的大小
poolPreparedStatements: true
maxPoolPreparedStatementPerConnectionSize: 20
filters: stat,wall
# 通过connectProperties属性来打开mergeSql功能；慢SQL记录
connectionProperties: druid.stat.mergeSql=true;druid.stat.slowSqlMillis=5000
# 合并多个DruidDataSource的监控数据
useGlobalDataSourceStat: true
#jta相关参数配置
jta:
log-dir: classpath:tx-logs
transaction-manager-id: txManager

三、数据源的注册

在DruidConfig.java中实现多个数据源的注册；分布式事务管理器的注册；druid的注册

package com.zjt.config;
import com.alibaba.druid.filter.stat.StatFilter;
import com.alibaba.druid.support.http.StatViewServlet;
import com.alibaba.druid.support.http.WebStatFilter;
import com.alibaba.druid.wall.WallConfig;
import com.alibaba.druid.wall.WallFilter;
import com.atomikos.icatch.jta.UserTransactionImp;
import com.atomikos.icatch.jta.UserTransactionManager;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.jta.atomikos.AtomikosDataSourceBean;
import org.springframework.boot.web.servlet.FilterRegistrationBean;
import org.springframework.boot.web.servlet.ServletRegistrationBean;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Primary;
import org.springframework.core.env.Environment;
import org.springframework.transaction.jta.JtaTransactionManager;
import javax.sql.DataSource;
import javax.transaction.UserTransaction;
import java.util.Properties;
/**
* Druid配置
*
*
*/
@Configuration
public class DruidConfig {
@Bean(name = "systemDataSource")
@Primary
@Autowired
public DataSource systemDataSource(Environment env){
AtomikosDataSourceBean ds = new AtomikosDataSourceBean();
Properties prop = build(env, "spring.datasource.druid.systemDB.");
ds.setXaDataSourceClassName("com.alibaba.druid.pool.xa.DruidXADataSource");
ds.setUniqueResourceName("systemDB");
ds.setPoolSize(5);
ds.setXaProperties(prop);
return ds;

}
@Autowired
@Bean(name = "businessDataSource")
public AtomikosDataSourceBean businessDataSource(Environment env){

AtomikosDataSourceBean ds = new AtomikosDataSourceBean();
Properties prop = build(env, "spring.datasource.druid.businessDB.");
ds.setXaDataSourceClassName("com.alibaba.druid.pool.xa.DruidXADataSource");
ds.setUniqueResourceName("businessDB");
ds.setPoolSize(5);
ds.setXaProperties(prop);

return ds;
}
/**
* 注入事物管理器
* @return
*/
@Bean(name = "xatx")
public JtaTransactionManager regTransactionManager (){
UserTransactionManager userTransactionManager = new UserTransactionManager();
UserTransaction userTransaction = new UserTransactionImp();
return new JtaTransactionManager(userTransaction, userTransactionManager);
}
private Properties build(Environment env, String prefix){
Properties prop = new Properties();
prop.put("url", env.getProperty(prefix + "url"));
prop.put("username", env.getProperty(prefix + "username"));
prop.put("password", env.getProperty(prefix + "password"));
prop.put("driverClassName", env.getProperty(prefix + "driverClassName", ""));
prop.put("initialSize", env.getProperty(prefix + "initialSize", Integer.class));
prop.put("maxActive", env.getProperty(prefix + "maxActive", Integer.class));
prop.put("minIdle", env.getProperty(prefix + "minIdle", Integer.class));
prop.put("maxWait", env.getProperty(prefix + "maxWait", Integer.class));
prop.put("poolPreparedStatements", env.getProperty(prefix + "poolPreparedStatements", Boolean.class));
prop.put("maxPoolPreparedStatementPerConnectionSize",
env.getProperty(prefix + "maxPoolPreparedStatementPerConnectionSize", Integer.class));
prop.put("maxPoolPreparedStatementPerConnectionSize",
env.getProperty(prefix + "maxPoolPreparedStatementPerConnectionSize", Integer.class));
prop.put("validationQuery", env.getProperty(prefix + "validationQuery"));
prop.put("validationQueryTimeout", env.getProperty(prefix + "validationQueryTimeout", Integer.class));
prop.put("testOnBorrow", env.getProperty(prefix + "testOnBorrow", Boolean.class));
prop.put("testOnReturn", env.getProperty(prefix + "testOnReturn", Boolean.class));
prop.put("testWhileIdle", env.getProperty(prefix + "testWhileIdle", Boolean.class));
prop.put("timeBetweenEvictionRunsMillis",
env.getProperty(prefix + "timeBetweenEvictionRunsMillis", Integer.class));
prop.put("minEvictableIdleTimeMillis", env.getProperty(prefix + "minEvictableIdleTimeMillis", Integer.class));
prop.put("filters", env.getProperty(prefix + "filters"));

return prop;
}
@Bean
public ServletRegistrationBean druidServlet(){
ServletRegistrationBean servletRegistrationBean = new ServletRegistrationBean(new StatViewServlet(), "/druid/*");
//控制台管理用户，加入下面2行 进入druid后台就需要登录
//servletRegistrationBean.addInitParameter("loginUsername", "admin");
//servletRegistrationBean.addInitParameter("loginPassword", "admin");
return servletRegistrationBean;
}
@Bean
public FilterRegistrationBean filterRegistrationBean(){
FilterRegistrationBean filterRegistrationBean = new FilterRegistrationBean();
filterRegistrationBean.setFilter(new WebStatFilter());
filterRegistrationBean.addUrlPatterns("/*");
filterRegistrationBean.addInitParameter("exclusions", "*.js,*.gif,*.jpg,*.png,*.css,*.ico,/druid/*");
filterRegistrationBean.addInitParameter("profileEnable", "true");
return filterRegistrationBean;
}
@Bean
public StatFilter statFilter(){
StatFilter statFilter = new StatFilter();
statFilter.setLogSlowSql(true); //slowSqlMillis用来配置SQL慢的标准，执行时间超过slowSqlMillis的就是慢。
statFilter.setMergeSql(true); //SQL合并配置
statFilter.setSlowSqlMillis(1000);//slowSqlMillis的缺省值为3000，也就是3秒。
return statFilter;
}
@Bean
public WallFilter wallFilter(){
WallFilter wallFilter = new WallFilter();
//允许执行多条SQL
WallConfig config = new WallConfig();
config.setMultiStatementAllow(true);
wallFilter.setConfig(config);
return wallFilter;
}

}

四、配置数据源对应的sqlSessionFactory

分别配置每个数据源对应的sqlSessionFactory，以及MapperScan扫描的包

MybatisDatasourceConfig.java

package com.zjt.config;
import com.zjt.util.MyMapper;
import org.apache.ibatis.session.SqlSessionFactory;
import org.mybatis.spring.SqlSessionFactoryBean;
import org.mybatis.spring.SqlSessionTemplate;
import org.mybatis.spring.annotation.MapperScan;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.io.support.PathMatchingResourcePatternResolver;
import org.springframework.core.io.support.ResourcePatternResolver;
import javax.sql.DataSource;
/**
*
* @description
*/
@Configuration
// 精确到 mapper 目录，以便跟其他数据源隔离
@MapperScan(basePackages = "com.zjt.mapper", markerInterface = MyMapper.class, sqlSessionFactoryRef = "sqlSessionFactory")
public class MybatisDatasourceConfig {
@Autowired
@Qualifier("systemDataSource")
private DataSource ds;
@Bean
public SqlSessionFactory sqlSessionFactory() throws Exception {
SqlSessionFactoryBean factoryBean = new SqlSessionFactoryBean();
factoryBean.setDataSource(ds);
//指定mapper xml目录
ResourcePatternResolver resolver = new PathMatchingResourcePatternResolver();
factoryBean.setMapperLocations(resolver.getResources("classpath:mapper/*.xml"));
return factoryBean.getObject();
}
@Bean
public SqlSessionTemplate sqlSessionTemplate() throws Exception {
SqlSessionTemplate template = new SqlSessionTemplate(sqlSessionFactory()); // 使用上面配置的Factory
return template;
}

//关于事务管理器，不管是JPA还是JDBC等都实现自接口 PlatformTransactionManager
// 如果你添加的是 spring-boot-starter-jdbc 依赖，框架会默认注入 DataSourceTransactionManager 实例。
//在Spring容器中，我们手工注解@Bean 将被优先加载，框架不会重新实例化其他的 PlatformTransactionManager 实现类。
/*@Bean(name = "transactionManager")
@Primary
public DataSourceTransactionManager masterTransactionManager() {
//MyBatis自动参与到spring事务管理中，无需额外配置，只要org.mybatis.spring.SqlSessionFactoryBean引用的数据源
// 与DataSourceTransactionManager引用的数据源一致即可，否则事务管理会不起作用。
return new DataSourceTransactionManager(ds);
}*/
}

MybatisDatasource2Config.java

package com.zjt.config;
import com.zjt.util.MyMapper;
import org.apache.ibatis.session.SqlSessionFactory;
import org.mybatis.spring.SqlSessionFactoryBean;
import org.mybatis.spring.SqlSessionTemplate;
import org.mybatis.spring.annotation.MapperScan;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.io.support.PathMatchingResourcePatternResolver;
import org.springframework.core.io.support.ResourcePatternResolver;
import javax.sql.DataSource;
/**
*
* @description
*/
@Configuration
// 精确到 mapper 目录，以便跟其他数据源隔离
@MapperScan(basePackages = "com.zjt.mapper2", markerInterface = MyMapper.class, sqlSessionFactoryRef = "sqlSessionFactory2")
public class MybatisDatasource2Config {
@Autowired
@Qualifier("businessDataSource")
private DataSource ds;
@Bean
public SqlSessionFactory sqlSessionFactory2() throws Exception {
SqlSessionFactoryBean factoryBean = new SqlSessionFactoryBean();
factoryBean.setDataSource(ds);
//指定mapper xml目录
ResourcePatternResolver resolver = new PathMatchingResourcePatternResolver();
factoryBean.setMapperLocations(resolver.getResources("classpath:mapper2/*.xml"));
return factoryBean.getObject();

}
@Bean
public SqlSessionTemplate sqlSessionTemplate2() throws Exception {
SqlSessionTemplate template = new SqlSessionTemplate(sqlSessionFactory2()); // 使用上面配置的Factory
return template;
}
//关于事务管理器，不管是JPA还是JDBC等都实现自接口 PlatformTransactionManager
// 如果你添加的是 spring-boot-starter-jdbc 依赖，框架会默认注入 DataSourceTransactionManager 实例。
//在Spring容器中，我们手工注解@Bean 将被优先加载，框架不会重新实例化其他的 PlatformTransactionManager 实现类。
/*@Bean(name = "transactionManager2")
@Primary
public DataSourceTransactionManager masterTransactionManager() {
//MyBatis自动参与到spring事务管理中，无需额外配置，只要org.mybatis.spring.SqlSessionFactoryBean引用的数据源
// 与DataSourceTransactionManager引用的数据源一致即可，否则事务管理会不起作用。
return new DataSourceTransactionManager(ds);
}*/
}

由于我们本例中只使用一个事务管理器：xatx，故就不在使用TxAdviceInterceptor.java和TxAdvice2Interceptor.java中配置的事务管理器了；有需求的童鞋可以自己配置其他的事务管理器；(见DruidConfig.java中查看)

五、测试接口

新建分布式业务测试接口JtaTestService.java和实现类JtaTestServiceImpl.java

其实就是一个很简单的test01()方法，在该方法中我们分别先后调用classService.saveOrUpdateTClass(tClass);和teacherService.saveOrUpdateTeacher(teacher);

实现先后操作两个数据源：然后我们可以自己debug跟踪事务的提交时机，此外，也可以在在两个方法全执行结束之后，手动制造一个运行时异常，来检查分布式事务是否全部回滚；

注意：

在实现类的方法中我使用的是：

@Transactional(transactionManager = "xatx", propagation = Propagation.REQUIRED, rollbackFor = { java.lang.RuntimeException.class })从而指定了使用哪个事务管理器，事务隔离级别(一般都用我这个默认的)，回滚的条件(一般可以使用Exception)，这三个可以自己根据业务实际修改；

package com.zjt.service3;
import java.util.Map;
public interface JtaTestService {
public Map<String,Object> test01();
}

package com.zjt.service3.impl;

import com.zjt.entity.TClass;
import com.zjt.entity.Teacher;
import com.zjt.service.TClassService;
import com.zjt.service2.TeacherService;
import com.zjt.service3.JtaTestService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Propagation;
import org.springframework.transaction.annotation.Transactional;

import java.util.LinkedHashMap;
import java.util.Map;

@Service("jtaTestServiceImpl")
public class JtaTestServiceImpl implements JtaTestService{

@Autowired
@Qualifier("teacherServiceImpl")
private TeacherService teacherService;
@Autowired
@Qualifier("tclassServiceImpl")
private TClassService tclassService;
@Override
@Transactional(transactionManager = "xatx", propagation = Propagation.REQUIRED, rollbackFor = { java.lang.RuntimeException.class })
public Map<String, Object> test01() {
LinkedHashMap<String,Object> resultMap=new LinkedHashMap<String,Object>();
TClass tClass=new TClass();
tClass.setName("8888");
tclassService.saveOrUpdateTClass(tClass);
Teacher teacher=new Teacher();
teacher.setName("8888");
teacherService.saveOrUpdateTeacher(teacher);
System.out.println(1/0);
resultMap.put("state","success");
resultMap.put("message","分布式事务同步成功");
return resultMap;
}
}

六、建立JtaTestContoller.java

建立JtaTestContoller.java，接受一个来自前端的http请求，触发JtaTestService 的test01方法

package com.zjt.web;
import com.zjt.service3.JtaTestService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.stereotype.Controller;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.ResponseBody;
import java.util.LinkedHashMap;
import java.util.Map;
@Controller
@RequestMapping("/jtaTest")
public class JtaTestContoller {
@Autowired
@Qualifier("jtaTestServiceImpl")
private JtaTestService taTestService;
@ResponseBody
@RequestMapping("/test01")
public Map<String,Object> test01(){
LinkedHashMap<String,Object> resultMap=new LinkedHashMap<String,Object>();
try {
return taTestService.test01();
}catch (Exception e){
resultMap.put("state","fail");
resultMap.put("message","分布式事务同步失败");
return resultMap;
}
}
}

七、在test.ftl中增加一个按钮来测试

//分布式事务测试
$("#JTATest").click(function(){
$.ajax({
type: "POST",
url: "${basePath!}/jtaTest/test01",
data: {} ,
async: false,
error: function (request) {
layer.alert("与服务器连接失败/(ㄒoㄒ)/~~");
return false;
},
success: function (data) {
if (data.state == 'fail') {
layer.alert(data.message);
return false;
}else if(data.state == 'success'){
layer.alert(data.message);
}
}
});
});
<button class="layui-btn" id="JTATest">同时向班级和老师表插入名为8888的班级和老师</button>

八、启动服务，验证结果

点击这个按钮，跳转到controller：

当正常执行了sql语句之后，我们可以发现数据库并没有变化，因为整个方法的事务还没有走完，当我们走到1/0这步时：

抛出运行时异常，并被spring事务拦截器拦截，并捕获异常：

在this.completeTransactionAfterThrowing(txInfo, var16);

데이터 소스의 관련 구성 항목을 별도로 application.yml로 추출합니다.

참고:

🎜

• 🎜이번 spring.datasource.type은 com.alibaba.druid.pool.xa.DruidXADataSource입니다;🎜
• 🎜 spring.jta.transaction-manager-id 값은 컴퓨터에서 고유합니다. 자세한 내용은 공식 문서를 읽어보세요. 🎜

🎜🎜🎜🎜전체 yml 파일은 다음과 같습니다. 🎜🎜rrreee🎜 3. 데이터 소스 등록🎜🎜DruidConfig.java에 여러 데이터 소스 등록, druid🎜rrreee🎜등록 4. 데이터 소스에 해당하는 sqlSessionFactory를 구성합니다.🎜🎜🎜각 데이터 소스에 해당하는 sqlSessionFactory를 구성합니다. , 그리고 MapperScan🎜🎜🎜🎜MybatisDatasourceConfig.java🎜🎜rrreee🎜🎜MybatisDatasource2Config.java🎜🎜rrreee🎜에 의해 스캔된 패키지 이 예에서는 하나의 트랜잭션 관리자인 xatx만 사용하므로 TxAdviceInterceptor.java 및 <code>TxAdvice2Interceptor.java는 트랜잭션 관리자에서 구성됩니다. 필요한 사람은 다른 트랜잭션 관리자를 직접 구성할 수 있습니다. (보려면 DruidConfig.java 참조) 🎜🎜 5. 테스트 인터페이스 🎜🎜🎜Create 새로운 분산 비즈니스 테스트 인터페이스 JtaTestService.java 및 구현 클래스 JtaTestServiceImpl.java🎜🎜🎜사실 이는 classService.saveOrUpdateTClass(tClass)를 호출하는 매우 간단한 test01() 메소드입니다. teacherService.saveOrUpdateTeacher(teacher);🎜🎜 두 가지 데이터 소스의 작업을 연속적으로 실현합니다. 그런 다음 트랜잭션 제출 시간을 직접 디버깅하고 추적할 수 있습니다. 또한 두 가지 방법을 사용하여 수행할 수도 있습니다. 실행 후에는 런타임 예외를 수동으로 생성하여 모든 분산 트랜잭션이 롤백되는지 확인합니다. 🎜🎜🎜🎜클래스 구현 방법에서는 다음을 사용합니다. 🎜🎜rrreeerrreeerrreee🎜 6. JtaTestContoller를 생성합니다. java🎜🎜🎜JtaTestContoller.java를 생성하고 프런트 엔드에서 http 요청을 수락한 다음 JtaTestService🎜🎜rrreee🎜Seven의 test01 메서드를 실행하여 test🎜rrreee🎜Eight에 버튼을 추가합니다. 결과를 확인하세요🎜🎜 🎜컨트롤러로 이동하려면 이 버튼을 클릭하세요: 🎜🎜🎜🎜🎜SQL 문을 정상적으로 실행한 후 아직 전체 메소드의 트랜잭션이 완료되지 않았기 때문에 데이터베이스가 변경되지 않은 것을 확인할 수 있습니다. 1/0 단계에 도달하면 : 🎜🎜🎜🎜🎜 런타임 예외가 발생하고 스프링에 의해 차단됩니다. 트랜잭션 인터셉터가 예외를 가로채고 캡처합니다. 🎜🎜🎜🎜🎜는 this.completeTransactionAfterThrowing(txInfo, var16); 메소드의 모든 트랜잭션을 롤백합니다: 🎜🎜🎜22:09:04.243 logback [http- nio-8080-exec- 5] INFO c.a.i.imp.CompositeTransactionImp - 트랜잭션 192.168.1.103.tm0000400006🎜🎜🎜의 롤백() 완료 이때 데이터베이스 검증을 다시 열어도 여전히 변경 사항이 없어 분산되었음을 증명합니다. 트랜잭션 구성이 성공했습니다. 🎜

위 내용은 Java Spring Boot 분산 트랜잭션을 구현하는 방법은 무엇입니까?의 상세 내용입니다. 자세한 내용은 PHP 중국어 웹사이트의 기타 관련 기사를 참조하세요!