【项目实践07】【多线程下变乱的同等性】

一、前言

本系列用来记载一些在实际项目中的小东西，并记载在过程中想到一些小东西，由于是随条记载，所以内容不会过于详细。

---

本篇从开篇到完成至少搁置了半年，因此之前的一些想法记不太清楚了。总之：本篇的方案未经过实际验证，审慎使用。
本篇的灵感泉源自 Spring在多线程环境下如何确保变乱同等性，正如其文章内容所属，想实现 多线程下变乱的同等性 有多重方案选择，包括最底子的 JDBC 编程、或者 分布式变乱思想等，不必局限于此。
二、项目背景

某个项目中需要处理用户上传的 Excel 文件，对每条记载单独处理且处理过程比力耗时，因此使用了线程池来加快处理速度。简朴Demo 如下：

1.     @Override
2.     public void testTransaction() {
3.         // 模拟构造大量数据
4.         List<Integer> indexs = initData(100);
5.         // 子线程插入数据
6.         indexs.parallelStream()
7.                 .forEach(index ->
8.                         transactionTemplate.executeWithoutResult(
9.                                 transactionStatus ->
10.                                         saveIndexDemo(index.toString())));
11.     }
12.         /**
13.      * 模拟大量数据
14.      *
15.      * @return
16.      */
17.     private List<Integer> initData(int size) {
18.         List<Integer> indexs = Lists.newArrayList();
19.         for (int i = 0; i < size; i++) {
20.             indexs.add(i);
21.         }
22.         return indexs;
23.     }
25.     /**
26.      * 子线程保存记录
27.      *
28.      * @param index
29.      */
30.     @SneakyThrows
31.     private void saveIndexDemo(String index) {
32.         log.info("{} 保存数据 {}", Thread.currentThread().getId(), index);
33.         demoMapper.insert(getDemo(index));
34.         // TODO : 模拟业务耗时
35.         Thread.sleep(10000);
36.     }
39.     /**
40.      * 获取实例
41.      *
42.      * @param userId
43.      * @return
44.      */
45.     private Demo getDemo(String userId) {
46.         final Demo demo = new Demo();
47.         demo.setUserId(userId);
48.         return demo;
49.     }

复制代码

本来相安无事，但是有一天晚上几个用户上传了一个2w行的Excel，这里就在开启线程处理，而天天记载的比力耗时，因此导致每个线程都占用着一个DB连接不释放。最终导致 DB 连接数不够用，其他业务异常。

---

三、实现方案

实际改造方案：限定线程开启的数量来控制并发度。改造比力简朴，这里直接列出，重点在下面的思路延伸。

1.     /**
2.      * 线程池
3.      */
4.     private static final ExecutorService EXECUTOR_SERVICE = Executors.newFixedThreadPool(5);
6.     @Transactional(rollbackFor = Exception.class)
7.     @Override
8.     public void testTransaction() {
9.         // 模拟构造大量数据
10.         List<Integer> indexs = initData(100);
12.         final CompletableFuture[] futures =
13.                 indexs.stream()
14.                         .map(index -> CompletableFuture.runAsync(() ->
15.                                 transactionTemplate.executeWithoutResult(
16.                                         transactionStatus ->
17.                                                 saveIndexDemo(index.toString())), EXECUTOR_SERVICE))
18.                         .toArray(CompletableFuture[]::new);
19.         CompletableFuture.allOf(futures).join();
20.     }

复制代码

---

四、思路延伸

上面的逻辑实际是有问题的：每个 saveIndexDemo 都处于不同的变乱中，也就是说其中某个 saveIndexDemo 假如异常了其他 saveIndexDemo 并不会回滚。因此这里的问题就演变成了：多线程下如何保证变乱的同等性。
1. 方案一

子线程业务执行结束后统一挂起，等待所有子线程执行结束后判断是否提交变乱 （2PC、3PC 的简化版）。如下：这里使用 CountDownLatch 来壅闭所有子线程，当所有子线程执行结束后根据 atomicBoolean 的值判断是否有变乱执行异常，假如异常，则将所有变乱回滚。

1.       @SneakyThrows
2.     @Override
3.     public String testTransaction01() {
4.         demoMapper.delete(null);
6.         // 初始化数据
7.         List<Integer> indexs = initData(10);
8.         // 等待子线程执行结束
9.         CountDownLatch countDownLatch = new CountDownLatch(indexs.size());
10.         // 业务执行标志
11.         AtomicBoolean atomicBoolean = new AtomicBoolean(true);
12.         CountDownLatch endFlag = new CountDownLatch(indexs.size());
13.         // 模拟每个业务单独开一个线程来执行，并且每个业务内部都开启了一个事务
14.         for (Integer index : indexs) {
15.             // 如果线程池线程数量 < 任务数量 会导致任务卡死：线程执行后 countDownLatch.await(); 等待剩余任务执行完成，而线程池中线程被占满，导致线程无法释放就无法再执行剩余任务
16.             EXECUTOR_SERVICE.execute(() ->
17.                     transactionTemplate.executeWithoutResult(transactionStatus -> {
18.                         runTranscation(index, countDownLatch, atomicBoolean, endFlag);
19.                     }));
20.         }
21.         // 等待所有线程执行结束后接口再返回
22.         endFlag.await();
24.         return "success";
25.     }
27.     /**
28.      * 执行事务
29.      *
30.      * @param index
31.      * @param countDownLatch
32.      * @param atomicBoolean
33.      * @param endFlag
34.      */
35.     private void runTranscation(Integer index, CountDownLatch countDownLatch, AtomicBoolean atomicBoolean, CountDownLatch endFlag) {
36.         try {
37.             // 业务执行结束
38.             saveIndexDemo(index.toString());
39.             countDownLatch.countDown();
40.             // 等待其他线程业务执行结束
41.             countDownLatch.await();
42.         } catch (Exception e) {
43.             // 如果业务执行异常，设置标志位为 false
44.             countDownLatch.countDown();
45.             atomicBoolean.set(false);
46.             // 如果某个线程执行异常，抛出异常回滚
47.             throw new RuntimeException("测试异常, 事务回滚");
48.         } finally {
49.             endFlag.countDown();
50.         }
51.     }

复制代码

上面代码虽然实现了所谓的功能，但是照旧存在一些漏洞的，包括但不限于：

• 线程池线程数量必须大于等于任务数量，否则会导致任务卡死：线程执行后 countDownLatch.await(); 等待剩余任务执行完成，而线程池中线程被占满，导致线程无法释放就无法再执行剩余任务。
  
• 主线程 countDownLatch.await() 这一步执行结束时，子线程中的变乱大概还没提交，不过可以使用其他标志位判断是否全部执行结束,这里不再演示。
  
• 假如子线程其中一个由于网络或锁等问题迟迟不能执行结束，会导致其他子线程的变乱一并无法提交，数据库连接池的连接会不停被占用。
  
• 由于子线程是同时开启线程而且互相称待的，所以一定要注意死锁情况。如下：
  
  
  • 线程A 执行业务逻辑时执行 SQL select * from demo where id = 1 for update， 此SQL会锁住 id = 1 的记载。
    
  • 线程B 执行业务逻辑时执行SQL update demo set user_id = 'demo' where id = '1'，此时会发现 id = 1 的记载被线程A锁住，因此等待线程A执行结束。
    
  • 线程A执行完业务逻辑后执行到 countDownLatch.await(); 后等待其他线程业务逻辑执行结束。而此时线程B由于等待线程A持有的记载A 的锁迟迟无法结束业务逻辑，也就无法执行到 countDownLatch.countDown(); 的逻辑，从而形成死锁的局面。
    
  
  

综上，该方案不推荐使用。
2. 方案二

本方案基于Spring在多线程环境下如何确保变乱同等性 中的内容，做了部门改造，详细思路在该文中有过介绍，本篇不过多赘述细节。
以下面Demo为例，实现功能为 ：初始化 15 条数据，每条数据使用一个单独的线程，每个线程开辟一个单独的变乱来将数据插入到数据库中。当其中一个线程的变乱出现异常时，所有变乱统一回滚。

1.     @Transactional(rollbackFor = Exception.class)
2.     @Override
3.     public void testTransaction02() {
4.         // 初始化 15 条数据
5.         List<Integer> indexs = initData(15);
6.         // 并发执行，每一个线程执行事务，事务的内容是将 Index 插入到 demo 表中
7.         CompletableFuture<TransactionUtil.TransactionResult>[] futures =
8.                 indexs.stream()
9.                         .map(index ->
10.                                 CompletableFuture.supplyAsync(() ->
11.                                         transactionUtil.executeWithoutResult(
12.                                                 transactionStatus ->
13.                                                         saveIndexDemo(index.toString()))))
14.                         .toArray(CompletableFuture[]::new);
15.         // 等待所有 CompletableFuture 执行结束
16.         CompletableFuture.allOf(futures).join();
18.         // 获取子线程执行结果
19.         List<TransactionUtil.TransactionResult> transactionResults =
20.                 Arrays.stream(futures)
21.                         .map(CompletableFuture::join)
22.                         .collect(Collectors.toList());
23.         // 批量处理所有子事物
24.         transactionUtil.batchTrim(transactionResults);
25. }

复制代码

上面代码的关键实现是下面的工具类，如下：

1. /**
2. * @Desc : 从 org.springframework.transaction.interceptor.TransactionAspectSupport#invokeWithinTransaction 中参考下整个流程
3. */
4. @Slf4j
5. @Component
6. public class TransactionUtil {
8.     @Resource
9.     private PlatformTransactionManager transactionManager;
11.     @Resource
12.     private TransactionDefinition transactionDefinition;
15.     /**
16.      * 基于 TransactionOperations#executeWithoutResult(java.util.function.Consumer) 改造
17.      *
18.      * @param function
19.      */
20.     public <T> TransactionResult<T> execute(Function<TransactionStatus, T> function) {
21.         TransactionStatus status = beginNewTransaction();
22.         TransactionResult<T> transactionResult =
23.                 new TransactionResult<>(status, TransactionResource.copyTransactionResource());
24.         try {
25.             transactionResult.setData(function.apply(status));
26.         } catch (Throwable ex) {
27.             log.error("Transaction failed", ex);
28.             transactionResult.setEx(ex);
29.         } finally {
30.             clearTransactionResource();
31.         }
32.         return transactionResult;
33.     }
35.     /**
36.      * 基于 TransactionOperations#executeWithoutResult(java.util.function.Consumer) 改造
37.      * 需要注意：在本篇代码中，执行该方法的是线程池中的某一个线程，而不是主线程
38.      *
39.      * @param consumer
40.      */
41.     public TransactionResult<Void> executeWithoutResult(Consumer<TransactionStatus> consumer) {
42.         // 为当前线程开启一个新的事务
43.         TransactionStatus status = beginNewTransaction();
44.         // 封装成 TransactionResult 对象
45.         TransactionResult<Void> transactionResult =
46.                 new TransactionResult<>(status, TransactionResource.copyTransactionResource());
47.         try {
48.             //执行具体业务逻辑
49.             consumer.accept(status);
50.         } catch (Throwable ex) {
51.             log.error("Transaction failed", ex);
52.             transactionResult.setEx(ex);
53.         } finally {
54.             // 清除 beginNewTransaction(); 方法创建事务时绑定当前线程的一些事务资源，防止线程池的线程复用时出现问题
55.             clearTransactionResource();
56.         }
57.         return transactionResult;
58.     }
60.     /**
61.      * 开启事务资源
62.      *
63.      * @return
64.      */
65.     private TransactionStatus beginNewTransaction() {
66.         // 开启一个新事务---此时autocommit已经被设置为了false,并且当前没有事务,这里创建的是一个新事务
67.         return transactionManager.getTransaction(transactionDefinition);
68.     }
70.     /**
71.      * 清除事务资源
72.      * beginNewTransaction 方法在创建新事物的同时还会给当前线程绑定一些事务属性，而由于线程池的线程是复用的，所以当下次再使用这个线程创新的事务时可能会出现问题：
73.      * 如 : 当第一次使用该线程创建事务时会给该线程绑定一个 DB 连接，而当事务提交后这个连接就会关闭，
74.      * 当下一次复用该线程执行事务时会判断当前线程已经绑定了 DB 连接就不会重新创建连接，而这个连接实际上已经关闭了，会出现 java.sql.SQLException: Connection is closed 问题
75.      */
76.     private void clearTransactionResource() {
77.         Maps.newHashMap(TransactionSynchronizationManager.getResourceMap()).keySet()
78.                 .forEach(TransactionSynchronizationManager::unbindResource);
79.         if (TransactionSynchronizationManager.isActualTransactionActive()) {
80.             TransactionSynchronizationManager.clearSynchronization();
81.         }
82.     }
84.     /**
85.      * 批量处理事务
86.      * 需要注意：在本篇代码中，执行该方法的是主线程，线程池中每个线程都会开启一个事务并将事务属性返回交由主线程来统一处理。
87.      *
88.      * @param transactionResults
89.      * @return 是否提交
90.      */
91.     public boolean batchTrim(Collection<TransactionResult> transactionResults) {
92.         // 判断是否出现异常
93.         boolean rollback = transactionResults.stream()
94.                 .anyMatch(transactionResult -> Objects.nonNull(transactionResult.getEx()));
95.         for (TransactionResult<?> transactionResult : transactionResults) {
96.             // 处理事务，提交或回滚
97.             if (rollback) {
98.                 transactionResult.rollback();
99.             } else {
100.                 transactionResult.commit();
101.             }
102.         }
103.         return !rollback;
104.     }
107.     /**
108.      * 保存当前事务资源,用于线程间的事务资源COPY操作
109.      */
110.     @Builder
111.     public static class TransactionResource {
112.         /**
113.          * 当前线程事务资源
114.          */
115.         private static final ThreadLocal<TransactionResource> RESOURCE_THREAD_LOCAL = new ThreadLocal<>();
117.         /**
118.          * 事务结束后默认会移除集合中的DataSource作为key关联的资源记录
119.          */
120.         private Map<Object, Object> resources = new HashMap<>();
122.         /**
123.          * 下面五个属性会在事务结束后被自动清理,无需我们手动清理
124.          */
125.         private Set<TransactionSynchronization> synchronizations = new HashSet<>();
126.         /**
127.          * 当前事务名称
128.          */
129.         private String currentTransactionName;
130.         /**
131.          * 当前事务只读属性
132.          */
133.         private Boolean currentTransactionReadOnly;
134.         /**
135.          * 当前事务隔离级别
136.          */
137.         private Integer currentTransactionIsolationLevel;
138.         /**
139.          * 当前事务是否处于活动状态
140.          */
141.         private Boolean actualTransactionActive;
143.         /**
144.          * 当前事务的属性拷贝
145.          *
146.          * @return
147.          */
148.         public static TransactionResource copyTransactionResource() {
149.             // 如果当前线程存在活跃事务则拷贝，否则返回一个基础 TransactionResource
150.             if (TransactionSynchronizationManager.isActualTransactionActive()) {
151.                 return TransactionResource.builder()
152.                         .resources(Maps.newHashMap(TransactionSynchronizationManager.getResourceMap()))
153.                         .synchronizations(Sets.newHashSet(TransactionSynchronizationManager.getSynchronizations()))
154.                         .currentTransactionName(TransactionSynchronizationManager.getCurrentTransactionName())
155.                         .currentTransactionReadOnly(TransactionSynchronizationManager.isCurrentTransactionReadOnly())
156.                         .currentTransactionIsolationLevel(TransactionSynchronizationManager.getCurrentTransactionIsolationLevel())
157.                         .actualTransactionActive(TransactionSynchronizationManager.isActualTransactionActive())
158.                         .build();
159.             } else {
160.                 return TransactionResource.builder()
161.                         .actualTransactionActive(TransactionSynchronizationManager.isActualTransactionActive())
162.                         .build();
163.             }
164.         }
166.         /**
167.          * 给当前线程注入事务属性
168.          */
169.         public void autowiredTransactionResource() {
170.             if (actualTransactionActive) {
171.                 TransactionSynchronizationManager.initSynchronization();
172.                 synchronizations.forEach(TransactionSynchronizationManager::registerSynchronization);
173.                 resources.forEach(TransactionSynchronizationManager::bindResource);
174.                 TransactionSynchronizationManager.setActualTransactionActive(actualTransactionActive);
175.                 TransactionSynchronizationManager.setCurrentTransactionName(currentTransactionName);
176.                 TransactionSynchronizationManager.setCurrentTransactionIsolationLevel(currentTransactionIsolationLevel);
177.                 TransactionSynchronizationManager.setCurrentTransactionReadOnly(currentTransactionReadOnly);
178.             }
179.         }
181.         /**
182.          * 移除事务属性
183.          *
184.          * @param all : 是否全部清除，默认只移除非 DataSource资源。这里解释下该参数的作用：
185.          *            事务提交后会自动清除事务相关的资源信息，所以不需要调用TransactionSynchronizationManager.clearSynchronization();同时事务信息也会自动清除 DataSource 相关的属性，所以也不需要清除。
186.          *            而对于上面代码来说，如果是子线程事务因为会执行提交或者回滚操作，会自动清除 DataSource 相关的属性，所以传 false; 而对主线程事务来说，事务提交并不是由我们控制，所以需要传 true，清空本身事务信息，为绑定子线程事务信息做准备
187.          */
188.         public void removeTransactionResource(boolean all) {
189.             if (actualTransactionActive) {
190.                 // 事务结束后默认会移除集合中的DataSource作为key关联的资源记录,DataSource如果重复移除,unbindResource时会因为不存在此key关联的事务资源而报错
191.                 resources.keySet().forEach(key -> {
192.                     if (all || !(key instanceof DataSource)) {
193.                         TransactionSynchronizationManager.unbindResource(key);
194.                     }
195.                 });
196.             }
197.         }
198.     }
200.     /**
201.      * 执行结果
202.      *
203.      * @param <T>
204.      */
205.     public class TransactionResult<T> {
206.         /**
207.          * 事务状态
208.          */
209.         private TransactionStatus status;
211.         /**
212.          * 线程资源
213.          */
214.         private TransactionResource resource;
216.         /**
217.          * 异常信息
218.          */
219.         @Getter
220.         @Setter
221.         private Throwable ex;
223.         /**
224.          * 执行结果
225.          */
226.         @Getter
227.         @Setter
228.         private T data;
230.         public TransactionResult(TransactionStatus status, TransactionResource resource) {
231.             this.status = status;
232.             this.resource = resource;
233.         }
235.         /**
236.          * 事务提交
237.          * 需要注意：在本篇代码中，执行该方法的是主线程
238.          */
239.         public void commit() {
240.             // 如果主线程开启了事务，则保存主线程事务信息：因为主线程可能同步存在事务，所以需要先备份主线程的事务信息
241.             TransactionResource currentResource = TransactionResource.copyTransactionResource();
242.             try {
243.                 // 移除当前线程(主线程)事务资源, 防止下面绑定其他事务资源时冲突，这里调用入参传 true ，清除主线程事务的全部资源信息
244.                 currentResource.removeTransactionResource(true);
245.                 if (TransactionSynchronizationManager.isSynchronizationActive()) {
246.                     // 清除活跃状态，否则注入子事务时会判断当前线程已经有活跃事务
247.                     TransactionSynchronizationManager.clearSynchronization();
248.                 }
249.                 // 注入子线程事务信息 : resource 中保存的事子线程的事务信息，将子线程事务信息注入到这里后再提交子线程事务
250.                 resource.autowiredTransactionResource();
251.                 // 提交子线程 : 事务提交后会自动清除事务相关的资源信息，所以不需要调用TransactionSynchronizationManager.clearSynchronization();,同时事务信息也会自动清除 DataSource 相关的属性，所以也不需要清除
252.                 transactionManager.commit(status);
253.                 // 清除子线程事务信息, 防止下面恢复主线程事务时属性冲突, 这里调用入参传 false ，因为上面调用了事务提交，所以会自动清除 DataSource 相关的属性，也就不需要我们自己清除了
254.                 resource.removeTransactionResource(false);
255.             } catch (Exception ex) {
256.                 log.info("commit error", ex);
257.                 // 清除子线程事务信息, 防止下面恢复主线程事务时属性冲突
258.                 resource.removeTransactionResource(false);
259.             } finally {
260.                 log.info("commit");
261.                 // 恢复主线程事务信息
262.                 currentResource.autowiredTransactionResource();
263.             }
264.         }
266.         /**
267.          * 事务回滚
268.          */
269.         public void rollback() {
270.             // 如果主线程开启了事务，则保存主线程事务信息：因为主线程可能同步存在事务，所以需要先备份主线程的事务信息
271.             TransactionResource currentResource = TransactionResource.copyTransactionResource();
272.             try {
273.                 // 移除当前线程(主线程)事务资源, 防止下面绑定其他事务资源时冲突，这里调用入参传 true ，清除主线程事务的全部资源信息
274.                 currentResource.removeTransactionResource(true);
275.                 if (TransactionSynchronizationManager.isSynchronizationActive()) {
276.                     // 清除活跃状态，否则注入子事务时会判断当前线程已经有活跃事务
277.                     TransactionSynchronizationManager.clearSynchronization();
278.                 }
279.                 // 注入子线程事务信息 : resource 中保存的事子线程的事务信息，将子线程事务信息注入到这里后再提交子线程事务
280.                 resource.autowiredTransactionResource();
281.                 // 提交子线程 : 事务提交后会自动清除事务相关的资源信息，所以不需要调用TransactionSynchronizationManager.clearSynchronization();,同时事务信息也会自动清除 DataSource 相关的属性，所以也不需要清除
282.                 rollbackOnException(status, ex);
283.                 // 清除子线程事务信息, 防止下面恢复主线程事务时属性冲突, 这里调用入参传 false ，因为上面调用了事务提交，所以会自动清除 DataSource 相关的属性，也就不需要我们自己清除了
284.                 resource.removeTransactionResource(false);
285.             } catch (Exception ex) {
286.                 log.info("commit error", ex);
287.                 // 清除子线程事务信息, 防止下面恢复主线程事务时属性冲突
288.                 resource.removeTransactionResource(false);
289.             } finally {
290.                 log.info("commit");
291.                 // 恢复主线程事务信息
292.                 currentResource.autowiredTransactionResource();
293.             }
294.         }
296.         /**
297.          * 回滚
298.          *
299.          * @param status
300.          * @param ex
301.          * @throws TransactionException
302.          */
303.         private void rollbackOnException(TransactionStatus status, Throwable ex) throws TransactionException {
304.             Assert.state(transactionManager != null, "No PlatformTransactionManager set");
306.             log.debug("Initiating transaction rollback on application exception", ex);
307.             try {
308.                 transactionManager.rollback(status);
309.             } catch (TransactionSystemException ex2) {
310.                 log.error("Application exception overridden by rollback exception", ex);
311.                 ex2.initApplicationException(ex);
312.                 throw ex2;
313.             } catch (RuntimeException | Error ex2) {
314.                 log.error("Application exception overridden by rollback exception", ex);
315.                 throw ex2;
316.             }
317.         }
318.     }
319. }

复制代码

上面的方案存在问题：

• 在编写代码时注意多个线程变乱在执行过程中的死锁问题。
  
• 最终变乱的提交方法 TransactionUtil#batchTrim 并不是原子的，当多个变乱中某个变乱提交失败了，则无法对之前提交成功的变乱回滚。
  

3. 方案三

基于方案二，又发散出了头脑三：所有子线程与主线程公用同一个变乱信息，最后在主线程执行结束后将这个变乱提交。
以下面Demo为例，实现功能为 ：初始化 15 条数据，每条数据使用一个单独的线程，每个线程都会使用主线程的变乱来将数据插入到数据库中。当其中一个线程或主线程出现异常时，变乱统一回滚。

1.     @Transactional
2.     @Override
3.     public void testTransaction03() {
4.         // 初始化 15 条数据
5.         List<Integer> indexs = initData(15);
7.         // 拷贝主线程事务资源
8.         SingleTransactionUtil.TransactionResource transactionResource =
9.                 SingleTransactionUtil.TransactionResource.copyTransactionResource();
11.         // 并发执行，每一个线程执行事务，事务的内容是将 Index 插入到 demo 表中
12.         CompletableFuture<SingleTransactionUtil.TransactionResult>[] futures =
13.                 indexs.stream()
14.                         .map(index ->
15.                                 CompletableFuture.supplyAsync(() ->
16.                                         singleTransactionUtil.executeWithoutResult(transactionResource,
17.                                                 () -> saveIndexDemo(index.toString()))))
18.                         .toArray(CompletableFuture[]::new);
19.         // 等待所有 CompletableFuture 执行结束
20.         CompletableFuture.allOf(futures).join();
22.         final boolean anyMatch = Arrays.stream(futures)
23.                 .anyMatch(f -> f.join().getEx() != null);
24.         if (anyMatch){
25.             throw new RuntimeException("测试异常, 事务回滚");
26.         }
28.         dataDemoMapper.insert(new DataDemo("1"));
29.         dataDemoMapper.insert(new DataDemo("2"));
30.     }

复制代码

上面代码的关键实现是下面的工具类，如下：

1. /**
2. * @Desc : 从 org.springframework.transaction.interceptor.TransactionAspectSupport#invokeWithinTransaction 中参考下整个流程
3. */
4. @Slf4j
5. @Component
6. public class SingleTransactionUtil {
8.     @Resource
9.     private PlatformTransactionManager transactionManager;
11.     @Resource
12.     private TransactionDefinition transactionDefinition;
15.     /**
16.      * 基于 TransactionOperations#executeWithoutResult(java.util.function.Consumer) 改造
17.      * 执行
18.      *
19.      * @param supplier
20.      */
21.     public <T> TransactionResult<T> execute(TransactionResource transactionResource,  Supplier<T> supplier) {
22.         transactionResource.autowiredTransactionResource();
23.         TransactionResult<T> transactionResult = new TransactionResult<>();
24.         try {
25.             transactionResult.setData(supplier.get());
26.         } catch (Throwable ex) {
27.             log.error("Transaction failed", ex);
28.             transactionResult.setEx(ex);
29.         } finally {
30.             clearTransactionResource();
31.         }
32.         return transactionResult;
33.     }
35.     /**
36.      * 基于 TransactionOperations#executeWithoutResult(java.util.function.Consumer) 改造
37.      * 执行
38.      *
39.      * @param runnable
40.      */
41.     public TransactionResult<Void> executeWithoutResult(TransactionResource transactionResource, Runnable runnable) {
42.         transactionResource.autowiredTransactionResource();
43.         TransactionResult<Void> transactionResult = new TransactionResult<>();
44.         try {
45.             runnable.run();
46.         } catch (Throwable ex) {
47.             log.error("Transaction failed", ex);
48.             transactionResult.setEx(ex);
49.         } finally {
50.             clearTransactionResource();
51.         }
53.         return transactionResult;
54.     }
55.    
56.     /**
57.      * 清除事务资源
58.      * beginNewTransaction 方法在创建新事物的同时还会给当前线程绑定一些事务属性，而由于线程池的线程是复用的，所以当下次再使用这个线程创新的事务时可能会出现问题：
59.      * 如 : 当第一次使用该线程创建事务时会给该线程绑定一个 DB 连接，而当事务提交后这个连接就会关闭，
60.      * 当下一次复用该线程执行事务时会判断当前线程已经绑定了 DB 连接就不会重新创建连接，而这个连接实际上已经关闭了，会出现 java.sql.SQLException: Connection is closed 问题
61.      */
62.     private void clearTransactionResource() {
63.         Maps.newHashMap(TransactionSynchronizationManager.getResourceMap()).keySet()
64.                 .forEach(TransactionSynchronizationManager::unbindResource);
65.         if (TransactionSynchronizationManager.isActualTransactionActive()) {
66.             TransactionSynchronizationManager.clearSynchronization();
67.         }
68.     }
70.     @Slf4j
71.     public static class TransactionResult<T> {
72.         /**
73.          * 异常信息
74.          */
75.         @Getter
76.         @Setter
77.         private Throwable ex;
79.         /**
80.          * 执行结果
81.          */
82.         @Getter
83.         @Setter
84.         private T data;
85.     }
88.     /**
89.      * 保存当前事务资源,用于线程间的事务资源COPY操作
90.      */
91.     @Builder
92.     public static class TransactionResource {
93.         /**
94.          * 当前线程事务资源
95.          */
96.         private static final ThreadLocal<TransactionResource> RESOURCE_THREAD_LOCAL = new ThreadLocal<>();
98.         /**
99.          * 事务结束后默认会移除集合中的DataSource作为key关联的资源记录
100.          */
101.         private Map<Object, Object> resources = new HashMap<>();
103.         /**
104.          * 下面五个属性会在事务结束后被自动清理,无需我们手动清理
105.          */
106.         private Set<TransactionSynchronization> synchronizations = new HashSet<>();
107.         /**
108.          * 当前事务名称
109.          */
110.         private String currentTransactionName;
111.         /**
112.          * 当前事务只读属性
113.          */
114.         private Boolean currentTransactionReadOnly;
115.         /**
116.          * 当前事务隔离级别
117.          */
118.         private Integer currentTransactionIsolationLevel;
119.         /**
120.          * 当前事务是否处于活动状态
121.          */
122.         private Boolean actualTransactionActive;
124.         /**
125.          * 当前事务的属性拷贝
126.          *
127.          * @return
128.          */
129.         public static TransactionResource copyTransactionResource() {
130.             // 如果当前线程存在活跃事务则拷贝，否则返回一个基础 TransactionResource
131.             if (TransactionSynchronizationManager.isActualTransactionActive()) {
132.                 return TransactionResource.builder()
133.                         .resources(Maps.newHashMap(TransactionSynchronizationManager.getResourceMap()))
134.                         .synchronizations(Sets.newHashSet(TransactionSynchronizationManager.getSynchronizations()))
135.                         .currentTransactionName(TransactionSynchronizationManager.getCurrentTransactionName())
136.                         .currentTransactionReadOnly(TransactionSynchronizationManager.isCurrentTransactionReadOnly())
137.                         .currentTransactionIsolationLevel(TransactionSynchronizationManager.getCurrentTransactionIsolationLevel())
138.                         .actualTransactionActive(TransactionSynchronizationManager.isActualTransactionActive())
139.                         .build();
140.             } else {
141.                 return TransactionResource.builder()
142.                         .actualTransactionActive(TransactionSynchronizationManager.isActualTransactionActive())
143.                         .build();
144.             }
145.         }
147.         /**
148.          * 给当前线程注入事务属性
149.          */
150.         public void autowiredTransactionResource() {
151.             if (actualTransactionActive) {
152.                 TransactionSynchronizationManager.initSynchronization();
153.                 synchronizations.forEach(TransactionSynchronizationManager::registerSynchronization);
154.                 resources.forEach(TransactionSynchronizationManager::bindResource);
155.                 TransactionSynchronizationManager.setActualTransactionActive(actualTransactionActive);
156.                 TransactionSynchronizationManager.setCurrentTransactionName(currentTransactionName);
157.                 TransactionSynchronizationManager.setCurrentTransactionIsolationLevel(currentTransactionIsolationLevel);
158.                 TransactionSynchronizationManager.setCurrentTransactionReadOnly(currentTransactionReadOnly);
159.             }
160.         }
162.         /**
163.          * 移除事务属性
164.          *
165.          * @param all : 是否全部清除，默认只移除非 DataSource资源。这里解释下该参数的作用：
166.          *            事务提交后会自动清除事务相关的资源信息，所以不需要调用TransactionSynchronizationManager.clearSynchronization();同时事务信息也会自动清除 DataSource 相关的属性，所以也不需要清除。
167.          *            而对于上面代码来说，如果是子线程事务因为会执行提交或者回滚操作，会自动清除 DataSource 相关的属性，所以传 false; 而对主线程事务来说，事务提交并不是由我们控制，所以需要传 true，清空本身事务信息，为绑定子线程事务信息做准备
168.          */
169.         public void removeTransactionResource(boolean all) {
170.             if (actualTransactionActive) {
171.                 // 事务结束后默认会移除集合中的DataSource作为key关联的资源记录,DataSource如果重复移除,unbindResource时会因为不存在此key关联的事务资源而报错
172.                 resources.keySet().forEach(key -> {
173.                     if (all || !(key instanceof DataSource)) {
174.                         TransactionSynchronizationManager.unbindResource(key);
175.                     }
176.                 });
177.             }
178.         }
179.     }
180. }

复制代码

五、参考内容

Spring在多线程环境下如何确保变乱同等性
https://mp.weixin.qq.com/s/gyEKbnzD3gIqfatL94DXsg

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。