多线程事务回滚方法

介绍

1.最近有一个大数据量插入的操作入库的业务场景，需要先做一些其他修改操作，然后在执行插入操作，由于插入数据可能会很多，用到多线程去拆分数据并行处理来提高响应时间，如果有一个线程执行失败，则全部回滚。

2.在spring中可以使用@Transactional注解去控制事务，使出现异常时会进行回滚，在多线程中，这个注解则不会生效，如果主线程需要先执行一些修改数据库的操作，当子线程在进行处理出现异常时，主线程修改的数据则不会回滚，导致数据错误。

案例演示

下面是事务不成功案例演示。

线程池配置

package com.mry.rollback.config;import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;/*** 线程池配置*/
public class ExecutorConfig {private static int maxPoolSize = Runtime.getRuntime().availableProcessors();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 queueSize = 500;int corePool = Math.min(5, maxPoolSize);return new ThreadPoolExecutor(corePool, maxPoolSize, 10000L, TimeUnit.MILLISECONDS,new LinkedBlockingQueue<>(queueSize),new ThreadPoolExecutor.AbortPolicy());}private ExecutorConfig(){}}

异常类

package com.mry.rollback.exception;import lombok.Data;/*** 异常类*/
@Data
public class ServiceException extends RuntimeException {private static final long serialVersionUID = 1L;private String msg;private int code = 500;public ServiceException(String msg) {super(msg);this.msg = msg;}public ServiceException(String msg, Throwable e) {super(msg, e);this.msg = msg;}public ServiceException(String msg, int code) {super(msg);this.msg = msg;this.code = code;}public ServiceException(String msg, int code, Throwable e) {super(msg, e);this.msg = msg;this.code = code;}}

实体类

package com.mry.rollback.entity;import com.baomidou.mybatisplus.annotation.TableName;
import lombok.Data;
import lombok.ToString;
import java.util.Date;@ToString
@Data
@TableName("employee")
public class Employee {private Integer employeeId;private Integer age;private String employeeName;private Date birthDate;private Integer gender;private String idNumber;private Date createTime;private Date updateTime;private Integer status;
}

控制层

package com.mry.rollback.controller;import com.mry.rollback.entity.Employee;
import com.mry.rollback.service.EmployeeService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import java.util.ArrayList;
import java.util.Calendar;
import java.util.List;@Slf4j
@RestController
@RequestMapping("/sys")
public class EmployeeController {@AutowiredEmployeeService employeeService;@GetMapping("/add")public String batchAddEmployee(){int size = 10;List<Employee> employeeDOList = new ArrayList<>(size);for (int i = 0; i<size;i++){Employee employee = new Employee();employee.setEmployeeName("lol"+i);employee.setAge(18);employee.setGender(1);employee.setBirthDate(Calendar.getInstance().getTime());employee.setIdNumber(i+"XX");employee.setStatus(1);employee.setCreateTime(Calendar.getInstance().getTime());employee.setUpdateTime(Calendar.getInstance().getTime());employeeDOList.add(employee);}try {employeeService.saveThread(employeeDOList);System.out.println("添加成功");}catch (Exception e){e.printStackTrace();}return "添加成功";}}

业务层

package com.mry.rollback.service;import com.mry.rollback.entity.Employee;
import java.util.List;public interface EmployeeService {public void saveThread(List<Employee> employeeList);
}

package com.mry.rollback.service.impl;import com.mry.rollback.config.ExecutorConfig;
import com.mry.rollback.entity.Employee;
import com.mry.rollback.exception.ServiceException;
import com.mry.rollback.mapper.EmployeeMapper;
import com.mry.rollback.service.EmployeeService;
import com.mry.rollback.util.ThreadUtil;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.atomic.AtomicBoolean;@Slf4j
@Service("employeeService")
public class EmployeeServiceImpl implements EmployeeService {@AutowiredEmployeeMapper employeeMapper;@Override@Transactionalpublic void saveThread(List<Employee> employeeList) {try {//先做删除操作,如果子线程出现异常,此操作不会回滚employeeMapper.delete(null);//获取线程池ExecutorService service = ExecutorConfig.getThreadPool();//拆分数据,拆分5份List<List<Employee>> lists = ThreadUtil.averageAssign(employeeList, 5);//执行的线程Thread []threadArray = new Thread[lists.size()];//监控子线程执行完毕,再执行主线程,要不然会导致主线程关闭,子线程也会随着关闭CountDownLatch countDownLatch = new CountDownLatch(lists.size());AtomicBoolean atomicBoolean = new AtomicBoolean(true);for (int i =0;i<lists.size();i++){if (i==lists.size()-1){atomicBoolean.set(false);}List<Employee> list  = lists.get(i);threadArray[i] =  new Thread(() -> {try {//最后一个线程抛出异常if (!atomicBoolean.get()){throw new ServiceException("出现异常",001);}//批量添加,mybatisPlus中自带的batch方法employeeMapper.saveBatchEmployee(list);}finally {countDownLatch.countDown();}});}for (int i = 0; i <lists.size(); i++){service.execute(threadArray[i]);}//当子线程执行完毕时,主线程再往下执行countDownLatch.await();System.out.println("添加完毕");}catch (Exception e){log.info("error",e);throw new ServiceException("出现异常",002);}finally {//connection.close();}}
}

mapper

package com.mry.rollback.mapper;import com.baomidou.mybatisplus.core.mapper.BaseMapper;
import com.mry.rollback.entity.Employee;
import org.apache.ibatis.annotations.Insert;
import org.apache.ibatis.annotations.Param;
import org.springframework.stereotype.Repository;
import java.util.List;@Repository
public interface EmployeeMapper extends BaseMapper<Employee> {@Insert("<script>" +"insert into employee (age, employee_name, birth_date, gender, id_number, create_time, update_time, status) " +"values " +"<foreach collection='employeeList' item='employee' index='index' separator=','>" +"(#{employee.age}, #{employee.employeeName}, #{employee.birthDate}, #{employee.gender}, #{employee.idNumber}, #{employee.createTime}, #{employee.updateTime}, #{employee.status})" +"</foreach>" +"</script>")public void saveBatchEmployee(@Param("employeeList") List<Employee> employeeList);}

工具类

package com.mry.rollback.util;import java.util.ArrayList;
import java.util.List;public class ThreadUtil {/*** 平均拆分list方法.* @param source* @param n* @param <T>* @return*/public static <T> List<List<T>> averageAssign(List<T> source,int n){List<List<T>> result=new ArrayList<List<T>>();int remaider=source.size()%n;int number=source.size()/n;int offset=0;//偏移量for(int i=0;i<n;i++){List<T> value=null;if(remaider>0){value=source.subList(i*number+offset, (i+1)*number+offset+1);remaider--;offset++;}else{value=source.subList(i*number+offset, (i+1)*number+offset);}result.add(value);}return result;}}

验证

1.数据库中存在一条数据：

2.请求接口：http://127.0.0.1:8866/sys/add
3.控制信息：

4.数据库信息：

注意：可以发现子线程组执行时，有一个线程执行失败，其他线程也会抛出异常，但是主线程中执行的删除操作，没有回滚，@Transactional注解没有生效。

解决方案

使用sqlSession控制手动提交事务

SqlSessionTemplate注入容器中

package com.mry.rollback.config;import org.apache.ibatis.session.SqlSession;
import org.apache.ibatis.session.SqlSessionFactory;
import org.mybatis.spring.SqlSessionTemplate;
import org.springframework.stereotype.Component;
import javax.annotation.Resource;/*** 获取sqlSession*/
@Component
public class SqlContext {@Resourceprivate SqlSessionTemplate sqlSessionTemplate;public SqlSession getSqlSession(){SqlSessionFactory sqlSessionFactory = sqlSessionTemplate.getSqlSessionFactory();return sqlSessionFactory.openSession();}}

改造业务层

package com.mry.rollback.service.impl;import com.mry.rollback.config.ExecutorConfig;
import com.mry.rollback.config.SqlContext;
import com.mry.rollback.entity.Employee;
import com.mry.rollback.exception.ServiceException;
import com.mry.rollback.mapper.EmployeeMapper;
import com.mry.rollback.service.EmployeeService;
import com.mry.rollback.util.ThreadUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.ibatis.session.SqlSession;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import javax.annotation.Resource;
import java.sql.Connection;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
import java.util.concurrent.atomic.AtomicBoolean;@Slf4j
@Service("employeeService")
public class EmployeeServiceImpl implements EmployeeService {@AutowiredEmployeeMapper employeeMapper;@ResourceSqlContext sqlContext;@Overridepublic void saveThread(List<Employee> employeeList) throws SQLException {// 获取数据库连接,获取会话(内部自有事务)SqlSession sqlSession = sqlContext.getSqlSession();Connection connection = sqlSession.getConnection();try {// 设置手动提交connection.setAutoCommit(false);//获取mapperEmployeeMapper employeeMapper = sqlSession.getMapper(EmployeeMapper.class);//先做删除操作employeeMapper.delete(null);//获取执行器ExecutorService service = ExecutorConfig.getThreadPool();List<Callable<Integer>> callableList  = new ArrayList<>();//拆分listList<List<Employee>> lists= ThreadUtil.averageAssign(employeeList, 5);AtomicBoolean atomicBoolean = new AtomicBoolean(true);for (int i =0;i<lists.size();i++){if (i==lists.size()-1){atomicBoolean.set(false);}List<Employee> list  = lists.get(i);//使用返回结果的callable去执行,Callable<Integer> callable = () -> {//让最后一个线程抛出异常if (!atomicBoolean.get()){throw new ServiceException("出现异常",001);}return employeeMapper.saveBatchEmployee(list);};callableList.add(callable);}//执行子线程List<Future<Integer>> futures = service.invokeAll(callableList);for (Future<Integer> future:futures) {//如果有一个执行不成功,则全部回滚if (future.get()<=0){connection.rollback();return;}}connection.commit();System.out.println("添加完毕");}catch (Exception e){connection.rollback();log.info("error",e);throw new ServiceException("出现异常",002);}finally {connection.close();}}
}

验证

1.数据库中存在一条数据：

2.请求接口：http://127.0.0.1:8877/sys/add
3.控制信息：

4.数据库信息：

注意：删除操作的数据回滚了，数据库中的数据依旧存在，说明事务成功了。

成功操作示例业务层改造

package com.mry.rollback.service.impl;import com.mry.rollback.config.ExecutorConfig;
import com.mry.rollback.config.SqlContext;
import com.mry.rollback.entity.Employee;
import com.mry.rollback.exception.ServiceException;
import com.mry.rollback.mapper.EmployeeMapper;
import com.mry.rollback.service.EmployeeService;
import com.mry.rollback.util.ThreadUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.ibatis.session.SqlSession;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import javax.annotation.Resource;
import java.sql.Connection;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
import java.util.concurrent.atomic.AtomicBoolean;@Slf4j
@Service("employeeService")
public class EmployeeServiceImpl implements EmployeeService {@AutowiredEmployeeMapper employeeMapper;@ResourceSqlContext sqlContext;@Overridepublic void saveThread(List<Employee> employeeList) throws SQLException {// 获取数据库连接,获取会话(内部自有事务)SqlSession sqlSession = sqlContext.getSqlSession();Connection connection = sqlSession.getConnection();try {// 设置手动提交connection.setAutoCommit(false);EmployeeMapper employeeMapper = sqlSession.getMapper(EmployeeMapper.class);//先做删除操作employeeMapper.delete(null);ExecutorService service = ExecutorConfig.getThreadPool();List<Callable<Integer>> callableList  = new ArrayList<>();List<List<Employee>> lists=ThreadUtil.averageAssign(employeeList, 5);for (int i =0;i<lists.size();i++){List<Employee> list  = lists.get(i);Callable<Integer> callable = () -> employeeMapper.saveBatchEmployee(list);callableList.add(callable);}//执行子线程List<Future<Integer>> futures = service.invokeAll(callableList);for (Future<Integer> future:futures) {if (future.get()<=0){connection.rollback();return;}}connection.commit();System.out.println("添加完毕");}catch (Exception e){connection.rollback();log.info("error",e);throw new ServiceException("出现异常",002);// throw new ServiceException(ExceptionCodeEnum.EMPLOYEE_SAVE_OR_UPDATE_ERROR);}}}

控制台日志输出：

数据库中数据：

注意：删除的删除了，添加的添加成功了，测试成功。