多线程事务如何保证效率和原子性

ゝ一纸荒年。 2023-09-23 23:43 15阅读 0赞

## 多线程事务 ##

> 在Spring开发时，遇到一个从Excel表导入数据到数据库的需求，当然最简单的方法就是先使用EasyExcel把数据读出到集合中，然后依次插入到数据库中。
> 
> 但如何保证效率，原子性呢？我们一步步优化方案。这里不会引入不必要的组件，而是自己模拟类似的思想。

**方法1：依次顺序插入**

void test() {
            List<User> users = getAllUsers();
            users.forEach(user -> userService.save(user));
        }
    复制代码

**方法2：使用批处理，一次操作中执行多条SQL**

void test() {
        List<User> users = getAllUsers();
        userService.saveBatch(users);
    }
    复制代码

**方法3：使用多线程+批处理，每个线程插入多条数据**

> 需要注意的一点，Spring容器不允许线程注入，也就是没办法在多线程直接使用Bean操作，例如：
> 
>     void testThread() {
>     // 下面两种方式是无效的，不会执行任何东西
>     Runnable runnable = () -> {
>              userService.save(new User());
>          };
>     // 方法1
>     new Thread(runnable).start();
>     // 方法2
>     Executors.newFixedThreadPool(1).submit(runnable);
>     }
>     复制代码
> 
> 我们需要下面的方式进行执行
> 
>     void testThread() {
>          Runnable runnable = () -> {
>              userService.save(new User());
>          };
>          ExecutorService executorService = Executors.newFixedThreadPool(1);
>          CompletableFuture<Void> future = CompletableFuture.runAsync(runnable, executorService);
>          future.join();
>      }
>     
>     复制代码

void testThread() {
        int threadSize = 5;
        ExecutorService executorService = Executors.newFixedThreadPool(threadSize);
        List<List<User>> list = new ArrayList<>();
        for (int i = 0; i < threadSize; i++) {
            // 我们假设数据拆分为五分
            list.add(getAllUsers());
        }
        for (List<User> users : list) {
            CompletableFuture.runAsync(()->{
                userService.saveBatch(users);
            },executorService).join();
        }
        System.out.println("插入成功");
    }
    复制代码

**方法4：这时候速度已经很快了，但是如果其中一个线程插入数据时发生错误进行回滚，其他线程是无法得知的，因为事务是针对线程的，所以这里我们需要用一些方式保证每个线程之间的状态是被共享的。**

// UserService#saveUserSyn()
    @Override
    public boolean saveUserSyn(List<User> users, CountDownLatch threadLatch, CountDownLatch mainLatch, UserError hasError) {
            TransactionStatus transactionStatus = dataSourceTransactionManager.getTransaction(transactionDefinition);
            System.out.println("子线程：" + Thread.currentThread().getName());
            try {
                users.forEach(this::save);
            } catch (Throwable e) {
                hasError.setHasError(true);
            } finally {
                threadLatch.countDown(); // 切换到主线程执行
            }
            try {
                mainLatch.await();  //等待主线程执行
            } catch (Throwable e) {
                hasError.setHasError(true);
            }
            // 判断是否有错误，如有错误 就回滚事务
            if (hasError.isHasError()) {
                dataSourceTransactionManager.rollback(transactionStatus);
            } else {
                dataSourceTransactionManager.commit(transactionStatus);
            }
            return true;
        }
    复制代码

// 测试方法
    @Test
        void userSaveSyn() {
    
            List<User> userList = getAllUsers();
    
            // 添加一个错误数据
            User user = new User();
            user.setUserAccount(null);
            user.setUserPassword("123456");
            userList.add(user);
    
            // 线程数量
            final Integer threadCount = 4;
    
            //每个线程处理的数据量
            final Integer dataPartionLength = (userList.size() + threadCount - 1) / threadCount;
    
            // 创建多线程处理任务
            ExecutorService studentThreadPool = Executors.newFixedThreadPool(threadCount);
            CountDownLatch threadLatchs = new CountDownLatch(threadCount); // 用于计算子线程提交数量
            CountDownLatch mainLatch = new CountDownLatch(1); // 用于判断主线程是否提交
    
            for (int i = 0; i < threadCount; i++) {
                // 每个线程处理的数据
                List<User> threadDatas = userList.stream()
                        .skip(i * dataPartionLength).limit(dataPartionLength)
                        .collect(Collectors.toList());
                studentThreadPool.execute(() -> {
                    userService.saveUserSyn(threadDatas, threadLatchs, mainLatch, hasError);
                });
            }
    
            try {
                // 倒计时锁设置超时时间 30s
                boolean await = threadLatchs.await(30, TimeUnit.SECONDS);
                if (!await) { // 等待超时，事务回滚
                    hasError.setHasError(true);
                }
            } catch (Throwable e) {
                e.printStackTrace();
                hasError.setHasError(true);
            }
            mainLatch.countDown(); // 切换到子线程执行
            studentThreadPool.shutdown(); //关闭线程池
            System.out.println("主线程完成");
        }
    复制代码

这里我们使用`CountDownLatch` 和 `Volatile`来解决这个问题。

[CountDownLatch的语法与原理讲解][CountDownLatch]

### `Volatile`保证线程间数据的可见性 ###

2PC（两阶段提交），这个属于分布式事务的一个理论，这里模拟了这样的业务场景，大致流程为：

*  每个线程开启事务，插入数据，但不提交，向主线程通知说，我这里已经好了
 *  主线程等待一段时间，看是否所有的子线程都没问题了。如果超时也算是异常
 *  如果没有异常，主线程向所有子线程通知，可以提交事务
 *  如果有异常，主线程向所有子线程通知，进行回滚操作
 *  而中间使用Volatile修饰的hasError对象进行传达，是否出现异常。需要注意如果只是传递普通的boolean对象，可能会发生不一致的情况，我测试时没法通过。
 *  CountDownLatch则保证子线程在主线程没有通知前，是不能提交事务的。

这里细心些就会发现，即便是主线程通知子线程可以提交了，子线程依然有可能出现提交失败的可能，那其他线程提交事务是无法得知这边的失败的消息的。这就是我们其实无法在一个Java进程中保证多线程的原子性。

作者：啵啵肠  
原文链接：https://juejin.cn/post/7195078084830167097

[CountDownLatch]: https://link.juejin.cn?target=https%3A%2F%2Fzhuanlan.zhihu.com%2Fp%2F148231820