贵阳论坛网站建设,冠县网站建设多少钱,产品设计考研学校,网站域名注册商查询Spring如何在多线程下保持事务的一致性
方法#xff1a;每个线程都开启各自的事务去执行相关业务#xff0c;等待所有线程的业务执行完成#xff0c;统一提交或回滚。
下面我们通过具体的案例来演示Spring如何在多线程下保持事务的一致性。
1、项目结构 2、数据库SQL
CR…Spring如何在多线程下保持事务的一致性
方法每个线程都开启各自的事务去执行相关业务等待所有线程的业务执行完成统一提交或回滚。
下面我们通过具体的案例来演示Spring如何在多线程下保持事务的一致性。
1、项目结构 2、数据库SQL
CREATE TABLE student (id int(11) NOT NULL AUTO_INCREMENT,name varchar(255) NOT NULL DEFAULT ,PRIMARY KEY (id)
) ENGINEInnoDB AUTO_INCREMENT1 DEFAULT CHARSETutf8mb4;3、pom依赖
?xml version1.0 encodingUTF-8?
project xmlnshttp://maven.apache.org/POM/4.0.0 xmlns:xsihttp://www.w3.org/2001/XMLSchema-instancexsi:schemaLocationhttp://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsdmodelVersion4.0.0/modelVersionparentgroupIdorg.springframework.boot/groupIdartifactIdspring-boot-starter-parent/artifactIdversion2.5.6/versionrelativePath//parentgroupIdcom.example/groupIdartifactIdTransaction/artifactIdversion0.0.1-SNAPSHOT/versionnameTransaction/namedescriptionTransaction/descriptionpropertiesjava.version1.8/java.version/propertiesdependenciesdependencygroupIdorg.springframework.boot/groupIdartifactIdspring-boot-starter-web/artifactId/dependencydependencygroupIdorg.mybatis.spring.boot/groupIdartifactIdmybatis-spring-boot-starter/artifactIdversion2.0.0/version/dependencydependencygroupIdmysql/groupIdartifactIdmysql-connector-java/artifactId/dependencydependencygroupIdorg.springframework.boot/groupIdartifactIdspring-boot-starter-test/artifactIdscopetest/scope/dependencydependencygroupIdorg.projectlombok/groupIdartifactIdlombok/artifactId/dependency/dependenciesbuildpluginsplugingroupIdorg.springframework.boot/groupIdartifactIdspring-boot-maven-plugin/artifactId/plugin/plugins/build/project4、配置文件
spring.datasource.jdbc-urljdbc:mysql://localhost:3306/test?serverTimezoneUTCuseUnicodetruecharacterEncodingutf-8useSSLtrue
spring.datasource.usernameroot
spring.datasource.passwordroot
spring.datasource.driver-class-namecom.mysql.cj.jdbc.Driver5、实体类
package com.example.transaction.model;import java.io.Serializable;/*** author tom*/
public class Student implements Serializable {private static final long serialVersionUID 1L;private int id;private String name;public int getId() {return id;}public void setId(int id) {this.id id;}public String getName() {return name;}public void setName(String name) {this.name name;}public Student(String name) {this.name name;}
}6、Mapper
package com.example.transaction.mapper;import com.example.transaction.model.Student;
import org.apache.ibatis.annotations.Insert;
import org.springframework.stereotype.Component;/*** author tom*/
Component
public interface StudentMapper {/*** 插入student* param student*/Insert(insert into student(name) VALUES(#{name}))void insert(Student student);
}7、数据源配置
package com.example.transaction.config;import org.mybatis.spring.annotation.MapperScan;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.boot.jdbc.DataSourceBuilder;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.jdbc.datasource.DataSourceTransactionManager;import javax.sql.DataSource;/*** author tom*/
Configuration
MapperScan(basePackages com.example.transaction.mapper)
public class DataSourceConfig {ConfigurationProperties(prefix spring.datasource)Beanpublic DataSource getDataSource() {return DataSourceBuilder.create().build();}Beanpublic DataSourceTransactionManager getTransactionManager(DataSource dataSource) {return new DataSourceTransactionManager(dataSource);}}8、测试
package com.example.transaction;import com.example.transaction.mapper.StudentMapper;
import com.example.transaction.model.Student;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;SpringBootTest
class TransactionApplicationTests {Autowiredprivate StudentMapper studentMapper;Testvoid contextLoads() {studentMapper.insert(new Student(John));}}我们先进行测试看数据库是否可以正常插入执行完的结果
idname1John
9、线程池
package com.example.transaction.config;import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;/*** author tom*/
public class ExecutorConfig {private final static int MAX_POOL_SIZE Runtime.getRuntime().availableProcessors();private final static int QUEUE_SIZE 500;private volatile static ExecutorService executorService;public static ExecutorService getThreadPool() {if (executorService null) {synchronized (ExecutorConfig.class) {if (executorService null) {executorService newThreadPool();}}}return executorService;}private static ExecutorService newThreadPool() {int corePool Math.min(5, MAX_POOL_SIZE);return new ThreadPoolExecutor(corePool, MAX_POOL_SIZE, 10000L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue(QUEUE_SIZE), new ThreadPoolExecutor.AbortPolicy());}private ExecutorConfig() {}
}10、多线程事务管理
package com.example.transaction.service;import com.example.transaction.config.ExecutorConfig;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.jdbc.datasource.DataSourceTransactionManager;
import org.springframework.stereotype.Service;
import org.springframework.transaction.TransactionStatus;
import org.springframework.transaction.support.DefaultTransactionDefinition;import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;/*** author tom*/
Service
public class MultiThreadingTransactionManager {/*** 数据源事务管理器*/private DataSourceTransactionManager dataSourceTransactionManager;Autowiredpublic void setUserService(DataSourceTransactionManager dataSourceTransactionManager) {this.dataSourceTransactionManager dataSourceTransactionManager;}/*** 用于判断子线程业务是否处理完成* 处理完成时threadCountDownLatch的值为0*/private CountDownLatch threadCountDownLatch;/*** 用于等待子线程全部完成后,子线程统一进行提交和回滚* 进行提交和回滚时mainCountDownLatch的值为0*/private final CountDownLatch mainCountDownLatch new CountDownLatch(1);/*** 是否提交事务,默认是true,当子线程有异常发生时,设置为false,回滚事务*/private final AtomicBoolean isSubmit new AtomicBoolean(true);public boolean execute(ListRunnable runnableList) {// 超时时间long timeout 30;setThreadCountDownLatch(runnableList.size());ExecutorService executorService ExecutorConfig.getThreadPool();runnableList.forEach(runnable - executorService.execute(() - executeThread(runnable, threadCountDownLatch, mainCountDownLatch, isSubmit)));// 等待子线程全部执行完毕try {// 若计数器变为零了,则返回 trueboolean isFinish threadCountDownLatch.await(timeout, TimeUnit.SECONDS);if (!isFinish) {// 如果还有为执行完成的就回滚isSubmit.set(false);System.out.println(存在子线程在预期时间内未执行完毕,任务将全部回滚);}} catch (Exception exception) {System.out.println(主线程发生异常,异常为: exception.getMessage());} finally {// 计数器减1,代表该主线程执行完毕mainCountDownLatch.countDown();}// 返回结果,是否执行成功,事务提交即为执行成功,事务回滚即为执行失败return isSubmit.get();}private void executeThread(Runnable runnable, CountDownLatch threadCountDownLatch, CountDownLatch mainCountDownLatch, AtomicBoolean isSubmit) {System.out.println(子线程: [ Thread.currentThread().getName() ]);// 判断别的子线程是否已经出现错误,错误别的线程已经出现错误,那么所有的都要回滚,这个子线程就没有必要执行了if (!isSubmit.get()) {System.out.println(整个事务中有子线程执行失败需要回滚, 子线程: [ Thread.currentThread().getName() ] 终止执行);// 计数器减1,代表该子线程执行完毕threadCountDownLatch.countDown();return;}// 开启事务DefaultTransactionDefinition defaultTransactionDefinition new DefaultTransactionDefinition();TransactionStatus transactionStatus dataSourceTransactionManager.getTransaction(defaultTransactionDefinition);try {// 执行业务逻辑runnable.run();} catch (Exception exception) {// 发生异常需要进行回滚,设置isSubmit为falseisSubmit.set(false);System.out.println(子线程: [ Thread.currentThread().getName() ]执行业务发生异常,异常为: exception.getMessage());} finally {// 计数器减1,代表该子线程执行完毕threadCountDownLatch.countDown();}try {// 等待主线程执行mainCountDownLatch.await();} catch (Exception exception) {System.out.println(子线程: [ Thread.currentThread().getName() ]等待提交或回滚异常,异常为: exception.getMessage());}try {// 提交if (isSubmit.get()) {dataSourceTransactionManager.commit(transactionStatus);System.out.println(子线程: [ Thread.currentThread().getName() ]进行事务提交);} else {dataSourceTransactionManager.rollback(transactionStatus);System.out.println(子线程: [ Thread.currentThread().getName() ]进行事务回滚);}} catch (Exception exception) {System.out.println(子线程: [ Thread.currentThread().getName() ]进行事务提交或回滚出现异常,异常为: exception.getMessage());}}private void setThreadCountDownLatch(int num) {this.threadCountDownLatch new CountDownLatch(num);}}11、正常插入
package com.example.transaction;import com.example.transaction.mapper.StudentMapper;
import com.example.transaction.model.Student;
import com.example.transaction.service.MultiThreadingTransactionManager;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;import java.util.ArrayList;
import java.util.List;SpringBootTest
public class TransactionApplicationTwoTests {Autowiredprivate StudentMapper studentMapper;Autowiredprivate MultiThreadingTransactionManager multiThreadingTransactionManager;Testvoid contextLoads() {ListStudent studentList new ArrayList();studentList.add(new Student(tom));studentList.add(new Student(marry));ListRunnable runnableList new ArrayList();studentList.forEach(student - runnableList.add(() - {System.out.println(当前线程[ Thread.currentThread().getName() ] 插入数据: student);try {studentMapper.insert(student);} catch (Exception e) {e.printStackTrace();}}));boolean isSuccess multiThreadingTransactionManager.execute(runnableList);System.out.println(isSuccess);}
}日志输出
......
子线程: [pool-1-thread-2]
子线程: [pool-1-thread-1]
2023-11-26 17:15:42.138 INFO 15736 --- [pool-1-thread-1] com.zaxxer.hikari.HikariDataSource : HikariPool-1 - Starting...
2023-11-26 17:15:42.319 INFO 15736 --- [pool-1-thread-1] com.zaxxer.hikari.HikariDataSource : HikariPool-1 - Start completed.
当前线程[pool-1-thread-2] 插入数据: com.example.transaction.model.Student1f52ee45
当前线程[pool-1-thread-1] 插入数据: com.example.transaction.model.Student238acf6d
true
子线程: [pool-1-thread-2]进行事务提交
子线程: [pool-1-thread-1]进行事务提交数据库中的数据
idname1John2tom3marry
12、异常插入
package com.example.transaction;import com.example.transaction.mapper.StudentMapper;
import com.example.transaction.model.Student;
import com.example.transaction.service.MultiThreadingTransactionManager;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;import java.util.ArrayList;
import java.util.List;SpringBootTest
public class TransactionApplicationThreeTests {Autowiredprivate StudentMapper studentMapper;Autowiredprivate MultiThreadingTransactionManager multiThreadingTransactionManager;Testvoid contextLoads() {ListStudent studentList new ArrayList();studentList.add(new Student(张三));studentList.add(new Student(李四));ListRunnable runnableList new ArrayList();studentList.forEach(student - runnableList.add(() - {System.out.println(当前线程[ Thread.currentThread().getName() ] 插入数据: student);try {studentMapper.insert(student);} catch (Exception e) {e.printStackTrace();}}));runnableList.add(() - System.out.println(1 / 0));boolean isSuccess multiThreadingTransactionManager.execute(runnableList);System.out.println(isSuccess);}
}日志输出
......
子线程: [pool-1-thread-1]
子线程: [pool-1-thread-2]
子线程: [pool-1-thread-3]
2023-11-26 17:19:45.876 INFO 11384 --- [pool-1-thread-2] com.zaxxer.hikari.HikariDataSource : HikariPool-1 - Starting...
2023-11-26 17:19:46.034 INFO 11384 --- [pool-1-thread-2] com.zaxxer.hikari.HikariDataSource : HikariPool-1 - Start completed.
子线程: [pool-1-thread-3]执行业务发生异常,异常为: / by zero
当前线程[pool-1-thread-1] 插入数据: com.example.transaction.model.Student6231e93c
当前线程[pool-1-thread-2] 插入数据: com.example.transaction.model.Student74568de7
false
子线程: [pool-1-thread-3]进行事务回滚
子线程: [pool-1-thread-2]进行事务回滚数据库中的数据
idname1John2tom3marry
从上面我们可以看出事务进行了回滚并没有插入到数据库中。
13、在主线程中统一进行事务的提交和回滚
这里将事务的回滚放在所有子线程执行完毕之后。
package com.example.transaction.service;import com.example.transaction.config.ExecutorConfig;
import lombok.Builder;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.jdbc.datasource.DataSourceTransactionManager;
import org.springframework.stereotype.Service;
import org.springframework.transaction.TransactionStatus;
import org.springframework.transaction.support.DefaultTransactionDefinition;
import org.springframework.transaction.support.TransactionSynchronization;
import org.springframework.transaction.support.TransactionSynchronizationManager;import javax.sql.DataSource;
import java.util.*;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;/*** author tom*/
Service
public class MultiThreadingTransactionManagerTwo {/*** 数据源事务管理器*/private DataSourceTransactionManager dataSourceTransactionManager;Autowiredpublic void setUserService(DataSourceTransactionManager dataSourceTransactionManager) {this.dataSourceTransactionManager dataSourceTransactionManager;}/*** 用于判断子线程业务是否处理完成* 处理完成时threadCountDownLatch的值为0*/private CountDownLatch threadCountDownLatch;/*** 是否提交事务,默认是true,当子线程有异常发生时,设置为false,回滚事务*/private final AtomicBoolean isSubmit new AtomicBoolean(true);public boolean execute(ListRunnable runnableList) {// 超时时间long timeout 30;ListTransactionStatus transactionStatusList Collections.synchronizedList(new ArrayList());ListTransactionResource transactionResourceList Collections.synchronizedList(new ArrayList());setThreadCountDownLatch(runnableList.size());ExecutorService executorService ExecutorConfig.getThreadPool();runnableList.forEach(runnable - executorService.execute(() - {try {// 执行业务逻辑executeThread(runnable, transactionStatusList, transactionResourceList);} catch (Exception exception) {exception.printStackTrace();// 执行异常,需要回滚isSubmit.set(false);} finally {threadCountDownLatch.countDown();}}));// 等待子线程全部执行完毕try {// 若计数器变为零了,则返回 trueboolean isFinish threadCountDownLatch.await(timeout, TimeUnit.SECONDS);if (!isFinish) {// 如果还有为执行完成的就回滚isSubmit.set(false);System.out.println(存在子线程在预期时间内未执行完毕,任务将全部回滚);}} catch (Exception exception) {exception.printStackTrace();}// 发生了异常则进行回滚操作,否则提交if (isSubmit.get()) {System.out.println(全部事务正常提交);for (int i 0; i runnableList.size(); i) {transactionResourceList.get(i).autoWiredTransactionResource();dataSourceTransactionManager.commit(transactionStatusList.get(i));transactionResourceList.get(i).removeTransactionResource();}} else {System.out.println(发生异常,全部事务回滚);for (int i 0; i runnableList.size(); i) {transactionResourceList.get(i).autoWiredTransactionResource();dataSourceTransactionManager.rollback(transactionStatusList.get(i));transactionResourceList.get(i).removeTransactionResource();}}// 返回结果,是否执行成功,事务提交即为执行成功,事务回滚即为执行失败return isSubmit.get();}private void executeThread(Runnable runnable, ListTransactionStatus transactionStatusList, ListTransactionResource transactionResourceList) {System.out.println(子线程: [ Thread.currentThread().getName() ]);DefaultTransactionDefinition defaultTransactionDefinition new DefaultTransactionDefinition();TransactionStatus transactionStatus dataSourceTransactionManager.getTransaction(defaultTransactionDefinition);// 开启新事务transactionStatusList.add(transactionStatus);// copy事务资源transactionResourceList.add(TransactionResource.copyTransactionResource());// 执行业务逻辑runnable.run();}private void setThreadCountDownLatch(int num) {this.threadCountDownLatch new CountDownLatch(num);}/*** 保存当前事务资源,用于线程间的事务资源COPY操作* p* Builder注解是Lombok库提供的一个注解,它可以用于自动生成Builder模式的代码,使用Builder注解可以简化创建对象实例的过程,并且可以使代码更加清晰和易于维护*/Builderprivate static class TransactionResource {// TransactionSynchronizationManager类内部默认提供了下面六个ThreadLocal属性,分别保存当前线程对应的不同事务资源// 保存当前事务关联的资源,默认只会在新建事务的时候保存当前获取到的DataSource和当前事务对应Connection的映射关系// 当然这里Connection被包装为了ConnectionHolder// 事务结束后默认会移除集合中的DataSource作为key关联的资源记录private MapObject, Object resources;//下面五个属性会在事务结束后被自动清理,无需我们手动清理// 事务监听者,在事务执行到某个阶段的过程中,会去回调监听者对应的回调接口(典型观察者模式的应用),默认为空集合private SetTransactionSynchronization synchronizations;// 存放当前事务名字private String currentTransactionName;// 存放当前事务是否是只读事务private Boolean currentTransactionReadOnly;// 存放当前事务的隔离级别private Integer currentTransactionIsolationLevel;// 存放当前事务是否处于激活状态private Boolean actualTransactionActive;/*** 对事务资源进行复制** return TransactionResource*/public static TransactionResource copyTransactionResource() {return TransactionResource.builder()//返回的是不可变集合.resources(TransactionSynchronizationManager.getResourceMap())//如果需要注册事务监听者,这里记得修改,我们这里不需要,就采用默认负责,spring事务内部默认也是这个值.synchronizations(new LinkedHashSet()).currentTransactionName(TransactionSynchronizationManager.getCurrentTransactionName()).currentTransactionReadOnly(TransactionSynchronizationManager.isCurrentTransactionReadOnly()).currentTransactionIsolationLevel(TransactionSynchronizationManager.getCurrentTransactionIsolationLevel()).actualTransactionActive(TransactionSynchronizationManager.isActualTransactionActive()).build();}/*** 使用*/public void autoWiredTransactionResource() {resources.forEach(TransactionSynchronizationManager::bindResource);//如果需要注册事务监听者,这里记得修改,我们这里不需要,就采用默认负责,spring事务内部默认也是这个值TransactionSynchronizationManager.initSynchronization();TransactionSynchronizationManager.setActualTransactionActive(actualTransactionActive);TransactionSynchronizationManager.setCurrentTransactionName(currentTransactionName);TransactionSynchronizationManager.setCurrentTransactionIsolationLevel(currentTransactionIsolationLevel);TransactionSynchronizationManager.setCurrentTransactionReadOnly(currentTransactionReadOnly);}/*** 移除*/public void removeTransactionResource() {// 事务结束后默认会移除集合中的DataSource作为key关联的资源记录// DataSource如果重复移除,unbindResource时会因为不存在此key关联的事务资源而报错resources.keySet().forEach(key - {if (!(key instanceof DataSource)) {TransactionSynchronizationManager.unbindResource(key);}});}}
}13.1 正常插入
package com.example.transaction;import com.example.transaction.mapper.StudentMapper;
import com.example.transaction.model.Student;
import com.example.transaction.service.MultiThreadingTransactionManagerTwo;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;import java.util.ArrayList;
import java.util.List;SpringBootTest
public class TransactionApplicationFourTests {Autowiredprivate StudentMapper studentMapper;Autowiredprivate MultiThreadingTransactionManagerTwo multiThreadingTransactionManagerTwo;Testvoid contextLoads() {ListStudent studentList new ArrayList();studentList.add(new Student(tom));studentList.add(new Student(marry));ListRunnable runnableList new ArrayList();studentList.forEach(student - runnableList.add(() - {System.out.println(当前线程[ Thread.currentThread().getName() ] 插入数据: student);try {studentMapper.insert(student);} catch (Exception e) {e.printStackTrace();}}));boolean isSuccess multiThreadingTransactionManagerTwo.execute(runnableList);System.out.println(isSuccess);}
}日志输出
......
子线程: [pool-1-thread-1]
子线程: [pool-1-thread-2]
2023-11-26 18:57:13.096 INFO 4280 --- [pool-1-thread-1] com.zaxxer.hikari.HikariDataSource : HikariPool-1 - Starting...
2023-11-26 18:57:13.256 INFO 4280 --- [pool-1-thread-1] com.zaxxer.hikari.HikariDataSource : HikariPool-1 - Start completed.
当前线程[pool-1-thread-2] 插入数据: com.example.transaction.model.Student6cf36c13
当前线程[pool-1-thread-1] 插入数据: com.example.transaction.model.Student7fc3efd5
全部事务正常提交
true数据库中的数据
idname1John2tom3marry6tom7marry
13.2 异常插入
package com.example.transaction;import com.example.transaction.mapper.StudentMapper;
import com.example.transaction.model.Student;
import com.example.transaction.service.MultiThreadingTransactionManager;
import com.example.transaction.service.MultiThreadingTransactionManagerTwo;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;import java.util.ArrayList;
import java.util.List;SpringBootTest
public class TransactionApplicationFiveTests {Autowiredprivate StudentMapper studentMapper;Autowiredprivate MultiThreadingTransactionManagerTwo multiThreadingTransactionManagerTwo;Testvoid contextLoads() {ListStudent studentList new ArrayList();studentList.add(new Student(张三));studentList.add(new Student(李四));ListRunnable runnableList new ArrayList();studentList.forEach(student - runnableList.add(() - {System.out.println(当前线程[ Thread.currentThread().getName() ] 插入数据: student);try {studentMapper.insert(student);} catch (Exception e) {e.printStackTrace();}}));runnableList.add(() - System.out.println(1 / 0));boolean isSuccess multiThreadingTransactionManagerTwo.execute(runnableList);System.out.println(isSuccess);}
}日志输出
子线程: [pool-1-thread-1]
子线程: [pool-1-thread-3]
子线程: [pool-1-thread-2]
2023-11-26 19:00:40.938 INFO 17920 --- [pool-1-thread-1] com.zaxxer.hikari.HikariDataSource : HikariPool-1 - Starting...
2023-11-26 19:00:41.097 INFO 17920 --- [pool-1-thread-1] com.zaxxer.hikari.HikariDataSource : HikariPool-1 - Start completed.
当前线程[pool-1-thread-1] 插入数据: com.example.transaction.model.Student2f7e458
当前线程[pool-1-thread-2] 插入数据: com.example.transaction.model.Student2b3ae8b
java.lang.ArithmeticException: / by zeroat com.example.transaction.TransactionApplicationFiveTests.lambda$contextLoads$2(TransactionApplicationFiveTests.java:37)at com.example.transaction.service.MultiThreadingTransactionManagerTwo.executeThread(MultiThreadingTransactionManagerTwo.java:107)at com.example.transaction.service.MultiThreadingTransactionManagerTwo.lambda$null$0(MultiThreadingTransactionManagerTwo.java:57)at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)at java.lang.Thread.run(Thread.java:745)
发生异常,全部事务回滚
false数据库中的数据
idname1John2tom3marry6tom7marry
14、使用CompletableFuture实现
package com.example.transaction.service;import lombok.Builder;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.jdbc.datasource.DataSourceTransactionManager;
import org.springframework.stereotype.Service;
import org.springframework.transaction.TransactionStatus;
import org.springframework.transaction.support.DefaultTransactionDefinition;
import org.springframework.transaction.support.TransactionSynchronization;
import org.springframework.transaction.support.TransactionSynchronizationManager;import javax.sql.DataSource;
import java.util.*;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.atomic.AtomicBoolean;/*** author tom*/
Service
public class MultiThreadingTransactionManagerThree {/*** 数据源事务管理器*/private DataSourceTransactionManager dataSourceTransactionManager;Autowiredpublic void setUserService(DataSourceTransactionManager dataSourceTransactionManager) {this.dataSourceTransactionManager dataSourceTransactionManager;}/*** 是否提交事务,默认是true,当子线程有异常发生时,设置为false,回滚事务*/private final AtomicBoolean isSubmit new AtomicBoolean(true);public boolean execute(ListRunnable runnableList) {ListTransactionStatus transactionStatusList Collections.synchronizedList(new ArrayList());ListTransactionResource transactionResourceList Collections.synchronizedList(new ArrayList());ListCompletableFuture? completableFutureList new ArrayList(runnableList.size());runnableList.forEach(runnable - completableFutureList.add(CompletableFuture.runAsync(() - {try {// 执行业务逻辑executeThread(runnable, transactionStatusList, transactionResourceList);} catch (Exception exception) {exception.printStackTrace();// 执行异常,需要回滚isSubmit.set(false);// 终止其它还未执行的任务completableFutureList.forEach(completableFuture - completableFuture.cancel(true));}})));// 等待子线程全部执行完毕try {// 阻塞直到所有任务全部执行结束,如果有任务被取消,这里会抛出异常,需要捕获CompletableFuture.allOf(completableFutureList.toArray(new CompletableFuture[]{})).get();} catch (Exception exception) {exception.printStackTrace();}// 发生了异常则进行回滚操作,否则提交if (!isSubmit.get()) {System.out.println(发生异常,全部事务回滚);for (int i 0; i runnableList.size(); i) {transactionResourceList.get(i).autoWiredTransactionResource();dataSourceTransactionManager.rollback(transactionStatusList.get(i));transactionResourceList.get(i).removeTransactionResource();}} else {System.out.println(全部事务正常提交);for (int i 0; i runnableList.size(); i) {transactionResourceList.get(i).autoWiredTransactionResource();dataSourceTransactionManager.commit(transactionStatusList.get(i));transactionResourceList.get(i).removeTransactionResource();}}// 返回结果,是否执行成功,事务提交即为执行成功,事务回滚即为执行失败return isSubmit.get();}private void executeThread(Runnable runnable, ListTransactionStatus transactionStatusList, ListTransactionResource transactionResourceList) {System.out.println(子线程: [ Thread.currentThread().getName() ]);DefaultTransactionDefinition defaultTransactionDefinition new DefaultTransactionDefinition();TransactionStatus transactionStatus dataSourceTransactionManager.getTransaction(defaultTransactionDefinition);// 开启新事务transactionStatusList.add(transactionStatus);// copy事务资源transactionResourceList.add(TransactionResource.copyTransactionResource());// 执行业务逻辑runnable.run();}/*** 保存当前事务资源,用于线程间的事务资源COPY操作* p* Builder注解是Lombok库提供的一个注解,它可以用于自动生成Builder模式的代码,使用Builder注解可以简化创建对象实例的过程,并且可以使代码更加清晰和易于维护*/Builderprivate static class TransactionResource {// TransactionSynchronizationManager类内部默认提供了下面六个ThreadLocal属性,分别保存当前线程对应的不同事务资源// 保存当前事务关联的资源,默认只会在新建事务的时候保存当前获取到的DataSource和当前事务对应Connection的映射关系// 当然这里Connection被包装为了ConnectionHolder// 事务结束后默认会移除集合中的DataSource作为key关联的资源记录private MapObject, Object resources;//下面五个属性会在事务结束后被自动清理,无需我们手动清理// 事务监听者,在事务执行到某个阶段的过程中,会去回调监听者对应的回调接口(典型观察者模式的应用),默认为空集合private SetTransactionSynchronization synchronizations;// 存放当前事务名字private String currentTransactionName;// 存放当前事务是否是只读事务private Boolean currentTransactionReadOnly;// 存放当前事务的隔离级别private Integer currentTransactionIsolationLevel;// 存放当前事务是否处于激活状态private Boolean actualTransactionActive;/*** 对事务资源进行复制** return TransactionResource*/public static TransactionResource copyTransactionResource() {return TransactionResource.builder()//返回的是不可变集合.resources(TransactionSynchronizationManager.getResourceMap())//如果需要注册事务监听者,这里记得修改,我们这里不需要,就采用默认负责,spring事务内部默认也是这个值.synchronizations(new LinkedHashSet()).currentTransactionName(TransactionSynchronizationManager.getCurrentTransactionName()).currentTransactionReadOnly(TransactionSynchronizationManager.isCurrentTransactionReadOnly()).currentTransactionIsolationLevel(TransactionSynchronizationManager.getCurrentTransactionIsolationLevel()).actualTransactionActive(TransactionSynchronizationManager.isActualTransactionActive()).build();}/*** 使用*/public void autoWiredTransactionResource() {resources.forEach(TransactionSynchronizationManager::bindResource);//如果需要注册事务监听者,这里记得修改,我们这里不需要,就采用默认负责,spring事务内部默认也是这个值TransactionSynchronizationManager.initSynchronization();TransactionSynchronizationManager.setActualTransactionActive(actualTransactionActive);TransactionSynchronizationManager.setCurrentTransactionName(currentTransactionName);TransactionSynchronizationManager.setCurrentTransactionIsolationLevel(currentTransactionIsolationLevel);TransactionSynchronizationManager.setCurrentTransactionReadOnly(currentTransactionReadOnly);}/*** 移除*/public void removeTransactionResource() {// 事务结束后默认会移除集合中的DataSource作为key关联的资源记录// DataSource如果重复移除,unbindResource时会因为不存在此key关联的事务资源而报错resources.keySet().forEach(key - {if (!(key instanceof DataSource)) {TransactionSynchronizationManager.unbindResource(key);}});}}
}14.1 正常插入
package com.example.transaction;import com.example.transaction.mapper.StudentMapper;
import com.example.transaction.model.Student;
import com.example.transaction.service.MultiThreadingTransactionManagerThree;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;import java.util.ArrayList;
import java.util.List;SpringBootTest
public class TransactionApplicationSixTests {Autowiredprivate StudentMapper studentMapper;Autowiredprivate MultiThreadingTransactionManagerThree multiThreadingTransactionManagerThree;Testvoid contextLoads() {ListStudent studentList new ArrayList();studentList.add(new Student(tom));studentList.add(new Student(marry));ListRunnable runnableList new ArrayList();studentList.forEach(student - runnableList.add(() - {System.out.println(当前线程[ Thread.currentThread().getName() ] 插入数据: student);try {studentMapper.insert(student);} catch (Exception e) {e.printStackTrace();}}));boolean isSuccess multiThreadingTransactionManagerThree.execute(runnableList);System.out.println(isSuccess);}
}日志输出
子线程: [ForkJoinPool.commonPool-worker-1]
子线程: [ForkJoinPool.commonPool-worker-2]
2023-11-26 19:17:00.674 INFO 12344 --- [onPool-worker-1] com.zaxxer.hikari.HikariDataSource : HikariPool-1 - Starting...
2023-11-26 19:17:00.815 INFO 12344 --- [onPool-worker-1] com.zaxxer.hikari.HikariDataSource : HikariPool-1 - Start completed.
当前线程[ForkJoinPool.commonPool-worker-2] 插入数据: com.example.transaction.model.Student25e1950b
当前线程[ForkJoinPool.commonPool-worker-1] 插入数据: com.example.transaction.model.Student57e8ff9a
全部事务正常提交
true数据库中的数据
idname1John2tom3marry6tom7marry10tom11marry
14.2 异常插入
package com.example.transaction;import com.example.transaction.mapper.StudentMapper;
import com.example.transaction.model.Student;
import com.example.transaction.service.MultiThreadingTransactionManagerThree;
import com.example.transaction.service.MultiThreadingTransactionManagerTwo;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;import java.util.ArrayList;
import java.util.List;SpringBootTest
public class TransactionApplicationSevenTests {Autowiredprivate StudentMapper studentMapper;Autowiredprivate MultiThreadingTransactionManagerThree multiThreadingTransactionManagerThree;Testvoid contextLoads() {ListStudent studentList new ArrayList();studentList.add(new Student(张三));studentList.add(new Student(李四));ListRunnable runnableList new ArrayList();studentList.forEach(student - runnableList.add(() - {System.out.println(当前线程[ Thread.currentThread().getName() ] 插入数据: student);try {studentMapper.insert(student);} catch (Exception e) {e.printStackTrace();}}));runnableList.add(() - System.out.println(1 / 0));boolean isSuccess multiThreadingTransactionManagerThree.execute(runnableList);System.out.println(isSuccess);}
}输出日志
子线程: [ForkJoinPool.commonPool-worker-2]
子线程: [ForkJoinPool.commonPool-worker-3]
子线程: [ForkJoinPool.commonPool-worker-1]
2023-11-26 19:19:01.862 INFO 15120 --- [onPool-worker-3] com.zaxxer.hikari.HikariDataSource : HikariPool-1 - Starting...
2023-11-26 19:19:02.016 INFO 15120 --- [onPool-worker-3] com.zaxxer.hikari.HikariDataSource : HikariPool-1 - Start completed.
当前线程[ForkJoinPool.commonPool-worker-1] 插入数据: com.example.transaction.model.Student3155d2ee
当前线程[ForkJoinPool.commonPool-worker-2] 插入数据: com.example.transaction.model.Student5ff9bde5
java.lang.ArithmeticException: / by zeroat com.example.transaction.TransactionApplicationSevenTests.lambda$contextLoads$2(TransactionApplicationSevenTests.java:37)at com.example.transaction.service.MultiThreadingTransactionManagerThree.executeThread(MultiThreadingTransactionManagerThree.java:90)at com.example.transaction.service.MultiThreadingTransactionManagerThree.lambda$null$1(MultiThreadingTransactionManagerThree.java:45)at java.util.concurrent.CompletableFuture$AsyncRun.run(CompletableFuture.java:1626)at java.util.concurrent.CompletableFuture$AsyncRun.exec(CompletableFuture.java:1618)at java.util.concurrent.ForkJoinTask.doExec(ForkJoinTask.java:289)at java.util.concurrent.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1056)at java.util.concurrent.ForkJoinPool.runWorker(ForkJoinPool.java:1692)at java.util.concurrent.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:157)
java.util.concurrent.ExecutionException: java.util.concurrent.CancellationExceptionat java.util.concurrent.CompletableFuture.reportGet(CompletableFuture.java:357)at java.util.concurrent.CompletableFuture.get(CompletableFuture.java:1895)
......
com.example.transaction.service.MultiThreadingTransactionManagerThree.lambda$null$1(MultiThreadingTransactionManagerThree.java:51)at java.util.concurrent.CompletableFuture$AsyncRun.run(CompletableFuture.java:1626)at java.util.concurrent.CompletableFuture$AsyncRun.exec(CompletableFuture.java:1618)at java.util.concurrent.ForkJoinTask.doExec(ForkJoinTask.java:289)at java.util.concurrent.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1056)at java.util.concurrent.ForkJoinPool.runWorker(ForkJoinPool.java:1692)at java.util.concurrent.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:157)
发生异常,全部事务回滚
false数据库中的数据
idname1John2tom3marry6tom7marry10tom11marry
至此结束。
