多线程下保证数据库事务

获取sqlSession

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;

@Component
public class SqlContext {
   
    @Resource
    private SqlSessionTemplate sqlSessionTemplate;

    public SqlSession getSqlSession(){
        SqlSessionFactory sqlSessionFactory = sqlSessionTemplate.getSqlSessionFactory();
        return sqlSessionFactory.openSession();
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

线程池配置

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() {

    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37

拆分新增List方法

    /**
     * 平均拆分list方法
     *
     * @param source 需拆分List
     * @param n      拆分份数
     * @param     返回值
     * @return
     */
    public static <T> List<List<T>> averageAssign(List<T> source, int n) {

        List<List<T>> result = new ArrayList<>();

        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
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30

多线程新增

    @Resource
    SqlContext sqlContext;

    /**
     * 测试多线程事务.
     *
     * @param employeeDOList
     */
    public void saveThread(List<Config> employeeDOList) throws SQLException {


        // 获取数据库连接,获取会话(内部自有事务)
        SqlSession sqlSession = sqlContext.getSqlSession();
        Connection connection = sqlSession.getConnection();
        try {

            // 设置手动提交
            connection.setAutoCommit(false);

            // 获取mapper层
            ConfigMapper employeeMapper = sqlSession.getMapper(ConfigMapper.class);

            //先做删除操作
            employeeMapper.deleteConfig(null);

            // 获取线程池
            ExecutorService service = ExecutorConfig.getThreadPool();


            List<Callable<Integer>> callableList = new ArrayList<>();
          
            // 拆分数据 5代表拆分成5份
            List<List<Config>> lists = averageAssign(employeeDOList, 5);

            for (int i = 0; i < lists.size(); i++) {
                List<Config> 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) {
            e.printStackTrace();
            // 失败回滚
            connection.rollback();
            throw new RuntimeException("出现异常");
        }
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60

参考

多线程事务如何回滚？说用 @Transactional 可以回去等通知了！