多執行緒環境下的 sql 事務問題
阿新 • • 發佈：2022-04-01
背景介紹
1，最近有一個大資料量插入的操作入庫的業務場景，需要先做一些其他修改操作，然後在執行插入操作，由於插入資料可能會很多，用到多執行緒去拆分資料並行處理來提高響應時間，如果有一個執行緒執行失敗，則全部回滾。
2，在spring中可以使用@Transactional註解去控制事務，使出現異常時會進行回滾，在多執行緒中，這個註解則不會生效，如果主執行緒需要先執行一些修改資料庫的操作，當子執行緒在進行處理出現異常時，主執行緒修改的資料則不會回滾，導致資料錯誤。
3，下面用一個簡單示例演示多執行緒事務。
公用的類和方法
/** * 平均拆分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; }
/** 執行緒池配置 * @version V1.0 */ 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(){} }
/** 獲取sqlSession * @author 86182 * @version V1.0 */ @Component public class SqlContext { @Resource private SqlSessionTemplate sqlSessionTemplate; public SqlSession getSqlSession(){ SqlSessionFactory sqlSessionFactory = sqlSessionTemplate.getSqlSessionFactory(); return sqlSessionFactory.openSession(); } }
示例事務不成功操作
/** * 測試多執行緒事務. * @param employeeDOList */ @Override @Transactional public void saveThread(List<EmployeeDO> employeeDOList) { try { //先做刪除操作,如果子執行緒出現異常,此操作不會回滾 this.getBaseMapper().delete(null); //獲取執行緒池 ExecutorService service = ExecutorConfig.getThreadPool(); //拆分資料,拆分5份 List<List<EmployeeDO>> lists=averageAssign(employeeDOList, 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<EmployeeDO> list = lists.get(i); threadArray[i] = new Thread(() -> { try { //最後一個執行緒丟擲異常 if (!atomicBoolean.get()){ throw new ServiceException("001","出現異常"); } //批量新增,mybatisPlus中自帶的batch方法 this.saveBatch(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(); } }
資料庫中存在一條資料：
//測試用例 @RunWith(SpringRunner.class) @SpringBootTest(classes = { ThreadTest01.class, MainApplication.class}) public class ThreadTest01 { @Resource private EmployeeBO employeeBO; /** * 測試多執行緒事務. * @throws InterruptedException */ @Test public void MoreThreadTest2() throws InterruptedException { int size = 10; List<EmployeeDO> employeeDOList = new ArrayList<>(size); for (int i = 0; i<size;i++){ EmployeeDO employeeDO = new EmployeeDO(); employeeDO.setEmployeeName("lol"+i); employeeDO.setAge(18); employeeDO.setGender(1); employeeDO.setIdNumber(i+"XX"); employeeDO.setCreatTime(Calendar.getInstance().getTime()); employeeDOList.add(employeeDO); } try { employeeBO.saveThread(employeeDOList); System.out.println("新增成功"); }catch (Exception e){ e.printStackTrace(); } } }
測試結果：
可以發現子執行緒組執行時，有一個執行緒執行失敗，其他執行緒也會丟擲異常，但是主執行緒中執行的刪除操作，沒有回滾，@Transactional註解沒有生效。
使用sqlSession控制手動提交事務
@Resource SqlContext sqlContext; /** * 測試多執行緒事務. * @param employeeDOList */ @Override public void saveThread(List<EmployeeDO> employeeDOList) throws SQLException { // 獲取資料庫連線,獲取會話(內部自有事務) SqlSession sqlSession = sqlContext.getSqlSession(); Connection connection = sqlSession.getConnection(); try { // 設定手動提交 connection.setAutoCommit(false); //獲取mapper EmployeeMapper employeeMapper = sqlSession.getMapper(EmployeeMapper.class); //先做刪除操作 employeeMapper.delete(null); //獲取執行器 ExecutorService service = ExecutorConfig.getThreadPool(); List<Callable<Integer>> callableList = new ArrayList<>(); //拆分list List<List<EmployeeDO>> lists=averageAssign(employeeDOList, 5); AtomicBoolean atomicBoolean = new AtomicBoolean(true); for (int i =0;i<lists.size();i++){ if (i==lists.size()-1){ atomicBoolean.set(false); } List<EmployeeDO> list = lists.get(i); //使用返回結果的callable去執行, Callable<Integer> callable = () -> { //讓最後一個執行緒丟擲異常 if (!atomicBoolean.get()){ throw new ServiceException("001","出現異常"); } return employeeMapper.saveBatch(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(); } }
// sql <insert id="saveBatch" parameterType="List"> INSERT INTO employee (employee_id,age,employee_name,birth_date,gender,id_number,creat_time,update_time,status) values <foreach collection="list" item="item" index="index" separator=","> ( #{item.employeeId}, #{item.age}, #{item.employeeName}, #{item.birthDate}, #{item.gender}, #{item.idNumber}, #{item.creatTime}, #{item.updateTime}, #{item.status} ) </foreach> </insert>
資料庫中一條資料：
測試結果：丟擲異常，
刪除操作的資料回滾了，資料庫中的資料依舊存在，說明事務成功了。
成功操作示例：
@Resource SqlContext sqlContext; /** * 測試多執行緒事務. * @param employeeDOList */ @Override public void saveThread(List<EmployeeDO> employeeDOList) 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<EmployeeDO>> lists=averageAssign(employeeDOList, 5); for (int i =0;i<lists.size();i++){ List<EmployeeDO> list = lists.get(i); Callable<Integer> callable = () -> employeeMapper.saveBatch(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); } }
測試結果：
資料庫中資料：
刪除的刪除了，新增的新增成功了，測試成功。