Spring如何在多线程下保持事务的一致性

Spring如何在多线程下保持事务的一致性

方法：每个线程都开启各自的事务去执行相关业务，等待所有线程的业务执行完成，统一提交或回滚。

下面我们通过具体的案例来演示Spring如何在多线程下保持事务的一致性。

1、项目结构

在这里插入图片描述

2、数据库SQL

CREATE TABLE `student` (`id` int(11) NOT NULL AUTO_INCREMENT,`name` varchar(255) NOT NULL DEFAULT '',PRIMARY KEY (`id`)
) ENGINE=InnoDB AUTO_INCREMENT=1 DEFAULT CHARSET=utf8mb4;

3、pom依赖

<?xml version="1.0" encoding="UTF-8"?>
<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 https://maven.apache.org/xsd/maven-4.0.0.xsd"><modelVersion>4.0.0</modelVersion><parent><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-parent</artifactId><version>2.5.6</version><relativePath/></parent><groupId>com.example</groupId><artifactId>Transaction</artifactId><version>0.0.1-SNAPSHOT</version><name>Transaction</name><description>Transaction</description><properties><java.version>1.8</java.version></properties><dependencies><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-web</artifactId></dependency><dependency><groupId>org.mybatis.spring.boot</groupId><artifactId>mybatis-spring-boot-starter</artifactId><version>2.0.0</version></dependency><dependency><groupId>mysql</groupId><artifactId>mysql-connector-java</artifactId></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-test</artifactId><scope>test</scope></dependency><dependency><groupId>org.projectlombok</groupId><artifactId>lombok</artifactId></dependency></dependencies><build><plugins><plugin><groupId>org.springframework.boot</groupId><artifactId>spring-boot-maven-plugin</artifactId></plugin></plugins></build></project>

4、配置文件

spring.datasource.jdbc-url=jdbc:mysql://localhost:3306/test?serverTimezone=UTC&useUnicode=true&characterEncoding=utf-8&useSSL=true
spring.datasource.username=root
spring.datasource.password=root
spring.datasource.driver-class-name=com.mysql.cj.jdbc.Driver

5、实体类

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"));}}

我们先进行测试，看数据库是否可以正常插入，执行完的结果：

idname
1John

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(List<Runnable> 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() {List<Student> studentList = new ArrayList<>();studentList.add(new Student("tom"));studentList.add(new Student("marry"));List<Runnable> 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.Student@1f52ee45
当前线程：[pool-1-thread-1] 插入数据: com.example.transaction.model.Student@238acf6d
true
子线程: [pool-1-thread-2]进行事务提交
子线程: [pool-1-thread-1]进行事务提交

数据库中的数据：

idname
1John
2tom
3marry

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() {List<Student> studentList = new ArrayList<>();studentList.add(new Student("张三"));studentList.add(new Student("李四"));List<Runnable> 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.Student@6231e93c
当前线程：[pool-1-thread-2] 插入数据: com.example.transaction.model.Student@74568de7
false
子线程: [pool-1-thread-3]进行事务回滚
子线程: [pool-1-thread-2]进行事务回滚

数据库中的数据：

idname
1John
2tom
3marry

从上面我们可以看出事务进行了回滚，并没有插入到数据库中。

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(List<Runnable> runnableList) {// 超时时间long timeout = 30;List<TransactionStatus> transactionStatusList = Collections.synchronizedList(new ArrayList<>());List<TransactionResource> 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, List<TransactionStatus> transactionStatusList, List<TransactionResource> 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 Map<Object, Object> resources;//下面五个属性会在事务结束后被自动清理,无需我们手动清理// 事务监听者,在事务执行到某个阶段的过程中,会去回调监听者对应的回调接口(典型观察者模式的应用),默认为空集合private Set<TransactionSynchronization> 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() {List<Student> studentList = new ArrayList<>();studentList.add(new Student("tom"));studentList.add(new Student("marry"));List<Runnable> 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.Student@6cf36c13
当前线程：[pool-1-thread-1] 插入数据: com.example.transaction.model.Student@7fc3efd5
全部事务正常提交
true

数据库中的数据：

idname
1John
2tom
3marry
6tom
7marry

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() {List<Student> studentList = new ArrayList<>();studentList.add(new Student("张三"));studentList.add(new Student("李四"));List<Runnable> 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.Student@2f7e458
当前线程：[pool-1-thread-2] 插入数据: com.example.transaction.model.Student@2b3ae8b
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

数据库中的数据：

idname
1John
2tom
3marry
6tom
7marry

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(List<Runnable> runnableList) {List<TransactionStatus> transactionStatusList = Collections.synchronizedList(new ArrayList<>());List<TransactionResource> transactionResourceList = Collections.synchronizedList(new ArrayList<>());List<CompletableFuture<?>> 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, List<TransactionStatus> transactionStatusList, List<TransactionResource> 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 Map<Object, Object> resources;//下面五个属性会在事务结束后被自动清理,无需我们手动清理// 事务监听者,在事务执行到某个阶段的过程中,会去回调监听者对应的回调接口(典型观察者模式的应用),默认为空集合private Set<TransactionSynchronization> 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() {List<Student> studentList = new ArrayList<>();studentList.add(new Student("tom"));studentList.add(new Student("marry"));List<Runnable> 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.Student@25e1950b
当前线程：[ForkJoinPool.commonPool-worker-1] 插入数据: com.example.transaction.model.Student@57e8ff9a
全部事务正常提交
true

数据库中的数据：

idname
1John
2tom
3marry
6tom
7marry
10tom
11marry

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() {List<Student> studentList = new ArrayList<>();studentList.add(new Student("张三"));studentList.add(new Student("李四"));List<Runnable> 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.Student@3155d2ee
当前线程：[ForkJoinPool.commonPool-worker-2] 插入数据: com.example.transaction.model.Student@5ff9bde5
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

数据库中的数据：

idname
1John
2tom
3marry
6tom
7marry
10tom
11marry

至此，结束。

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.mzph.cn/news/174034.shtml

如若内容造成侵权/违法违规/事实不符，请联系多彩编程网进行投诉反馈email:809451989@qq.com，一经查实，立即删除！

相关文章

自动标注好用吗？基于SAM和Label Studio搭建半自动实例分割标注平台

自动标注好用吗？基于SAM和Label Studio搭建半自动实例分割标注平台

文章目录 一、半自动标注二、缺点三、安装方法1、 python版本要求2、下载playground3、SAM安装4、SAM权重下载5、安装label-studio-ml6、启动SAM接口7、SAM启动日志8、安装并启动label-studio9、label-studio启动日志 四、半自动标注使用方法1、创建project并导入数据2、标签设…
阅读更多...
P8A002-CIA安全模型-配置Linux描述网络安全CIA模型之可用性案例

P8A002-CIA安全模型-配置Linux描述网络安全CIA模型之可用性案例

【预备知识】 可用性（Availability） 数据可用性是一种以使用者为中心的设计概念，易用性设计的重点在于让产品的设计能够符合使用者的习惯与需求。以互联网网站的设计为例，希望让使用者在浏览的过程中不会产生压力或感到挫折，并能让使用者在使用网站功能时，能用最少的努力…
阅读更多...
请问大家在都在什么场景用到嵌入式数据库？

请问大家在都在什么场景用到嵌入式数据库？

请问大家在都在什么场景用到嵌入式数据库&#xff1f; 嵌入式数据库在许多场景中都有广泛的应用。这些数据库通常被设计成轻量级、占用资源少且易于集成到其他应用程序中。以下是一些常见的场景和领域&#xff0c;在这些场景中嵌入式数据库被广泛使用&#xff1a;最近很多小伙伴…
阅读更多...
记录一个mqtt错误

记录一个mqtt错误

在vue-admin-template 中引入mqtt 安装不报错&#xff0c;引入试过 import mqtt from mqtt import * as mqtt from mqtt/dist/mqtt.min; import {connect} from mqtt 一直报错&#xff1a; 就表示不理解&#xff0c;网上也没查到相应的资料&#xff0c;请告诉我我不是第一个遇…
阅读更多...
FTP服务器搭建

FTP服务器搭建

1.FTP服务器概述 FTP服务器&#xff08;File Transfer Protocol Server&#xff09;是一种提供文件传输服务的服务器。FTP是一种标准的网络协议&#xff0c;用于在计算机之间进行文件传输。FTP服务器允许用户通过FTP协议上传、下载、删除和管理文件&#xff0c;从而使文件在不同…
阅读更多...
【面经八股】搜广推方向：面试记录（三）

【面经八股】搜广推方向：面试记录（三）

【面经&八股】搜广推方向：面试记录（三） 文章目录 【面经&八股】搜广推方向：面试记录（三）1. 编程题1.1 大数乘法1.2 大数加法2. 项目介绍3. 有了解过的广告推荐模型吗4. 广告模型回归问题1. 编程题 上来直接写编程题，有点儿懵逼。 1.1 大数乘法 可以参考 该博…
阅读更多...
如何在Node.js和Express中设置TypeScript（2023年）

如何在Node.js和Express中设置TypeScript（2023年）

如何在Node.js和Express中设置TypeScript&#xff08;2023年&#xff09; 在这篇文章中&#xff0c;我们将介绍在Express应用程序中设置TypeScript的最佳方法&#xff0c;了解与之相关的基本限制。 文章目录 如何在Node.js和Express中设置TypeScript&#xff08;2023年&#x…
阅读更多...
中国城市数字经济发展对环境污染的影响及机理研究（2011-2021年）

中国城市数字经济发展对环境污染的影响及机理研究（2011-2021年）

参照张翱祥&#xff08;2022&#xff09;的做法&#xff0c;本团队对来自南方经济《中国城市数字经济发展对环境污染的影响及机理研究》一文中的基准回归部分进行复刻 参考赵涛&#xff08;2020&#xff09;计算&#xff0c;PM2.5根据Atmospheric Composition Analysis Group计…
阅读更多...
案例033:基于微信小程序的商品展示系统设计与实现

案例033:基于微信小程序的商品展示系统设计与实现

文末获取源码 开发语言&#xff1a;Java 框架&#xff1a;SSM JDK版本&#xff1a;JDK1.8 数据库&#xff1a;mysql 5.7 开发软件&#xff1a;eclipse/myeclipse/idea Maven包&#xff1a;Maven3.5.4 小程序框架&#xff1a;uniapp 小程序开发软件&#xff1a;HBuilder X 小程序…
阅读更多...
巴菲特清仓Paytm，亏损62亿卢比

巴菲特清仓Paytm，亏损62亿卢比

KlipC报道&#xff1a;伯克希尔哈萨韦清仓其在印度“支付宝”Paytm的股份。 KlipC的合伙人Andi D表示&#xff1a;“据公开资料显示&#xff0c;伯克希尔.哈萨韦于2018年斥3亿美元巨资收购Paytm2.6%的股份&#xff0c;沃伦巴菲特公司于2021年出售Paytm 价值3600万美元的股份&am…
阅读更多...
卡码网语言基础课 | 14. 链表的基础操作Ⅱ

卡码网语言基础课 | 14. 链表的基础操作Ⅱ

题目&#xff1a; 构建一个单向链表&#xff0c;链表中包含一组整数数据&#xff0c;输出链表中的第 m 个元素&#xff08;m 从 1 开始计数&#xff09;。 要求&#xff1a; 1. 使用自定义的链表数据结构 2. 提供一个 linkedList 类来管理链表&#xff0c;包含构建链表、输出…
阅读更多...
【多属性对象“{a:1,b:2}”】与【单属性对象的数组“[{a:1},{b:2}]”】的相互转换

【多属性对象“{a:1,b:2}”】与【单属性对象的数组“[{a:1},{b:2}]”】的相互转换

前端开发的某些场景&#xff08;比如用echarts开发某些可视化图表&#xff09;经常需要将【多属性对象&#xff0c;如“{a:1,b:2}”】与【单属性对象的数组&#xff0c;如“[{a:1},{b:2}]”】做相互转换&#xff0c;以下是不通过循环&#xff0c;简洁实现这种转换的方法&#x…
阅读更多...
支持向量机的算法原理

支持向量机的算法原理

支持向量机&#xff08;Support Vector Machine&#xff0c;简称SVM&#xff09;是机器学习领域中一种常用的分类算法&#xff0c;它基于统计学习理论和结构风险最小化原则&#xff0c;具有很强的理论基础和良好的分类性能。本文将详细介绍支持向量机的算法原理&#xff0c;并解…
阅读更多...
【C语言】优化通讯录管理系统2

【C语言】优化通讯录管理系统2

本篇博客是基于上一篇博客写出来的&#xff0c;了解上一篇博客 大家好&#xff0c;我是苏貝&#xff0c;本篇博客带大家再次优化上一篇的通讯录&#xff0c;实现将录入的数据在程序退出后存储到文件中&#xff0c;在下一次程序开始时打开文件获取数据&#xff0c;如果你觉得我写…
阅读更多...
好用到难以置信的全域BI：揭秘店铺服务从优秀到卓越的3个办法

好用到难以置信的全域BI：揭秘店铺服务从优秀到卓越的3个办法

双11刚结束&#xff0c;一些平时易忽略的问题被放大出来&#xff0c;发现问题不可怕&#xff0c;可怕的是无视。如果您还没想好接下来怎么调整&#xff0c;本篇介绍的「全域BI-服务」定会给您一些思路。 过往&#xff0c;传统客服服务的管理仍停留于人工操作阶段&#xff0c;企…
阅读更多...
冯·诺依曼体系结构和操作系统

冯·诺依曼体系结构和操作系统

目录 一、冯诺依曼体系结构 1、初见结构 2、对体系结构的理解 3、总结 二、操作系统 1、概念 2、作用 一、冯诺依曼体系结构 1、初见结构 数学家冯诺依曼提出了计算机制造的三个基本原则&#xff0c;即采用二进制逻辑、程序存储执行以及计算机由五个部分组成&#xff08…
阅读更多...
sqli-labs(5)

sqli-labs(5)

23. 判断是注释符被过滤了我们用‘1’‘1来闭合后面的’ 这里不能使用order by来判断列数直接通过union select来判断 -1 union select 1,2,3 and 11 -1 union select 1,(select group_concat(table_name) from information_schema.tables where table_schemasecurity) ,3 an…
阅读更多...
如何使用cpolar+Jellyfin自建私人影音平台【内网穿透】

如何使用cpolar+Jellyfin自建私人影音平台【内网穿透】

&#x1f3a5; 个人主页&#xff1a;深鱼~ &#x1f525;收录专栏&#xff1a;cpolar &#x1f304;欢迎 &#x1f44d;点赞✍评论⭐收藏 文章目录 1. 前言2. Jellyfin服务网站搭建2.1. Jellyfin下载和安装2.2. Jellyfin网页测试 3.本地网页发布3.1 cpolar的安装和注册3.2 Cpo…
阅读更多...
SpringBoot参数校验@Validated和@Valid的使用

SpringBoot参数校验@Validated和@Valid的使用

1、Validated和Valid区别 Validated&#xff1a;可以用在类、方法和方法参数上。但是不能用在成员属性&#xff08;字段&#xff09;上Valid&#xff1a;可以用在方法、构造函数、方法参数和成员属性&#xff08;字段&#xff09;上 2、引入依赖 Spring Boot 2.3 1 之前&…
阅读更多...
Linux多线程基本概念

Linux多线程基本概念

目录 ​编辑 1.什么是进程&#xff0c;线程&#xff0c;并发&#xff0c;并行 优点 缺点 什么资源是线程应该私有的呢 为什么线程切换成本更低呢 3.线程控制 pthread_create lpthread选项 makefile 代码实现 ps -aL 什么是LWP 轻量级进程ID与进程ID之间的区别 LWP与pthr…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023