public void removeAuthorityModuleSeq(Integer authorityModuleId, IAuthorityService iAuthorityService, IRoleAuthorityService iRoleAuthorityService) {
    //1.查詢(xún)出當(dāng)前資源模塊下所有資源,查詢(xún)出來(lái)后進(jìn)行刪除
    deleteAuthoritiesOfCurrentAuthorityModule(authorityModuleId, iAuthorityService, iRoleAuthorityService);
    //2.查詢(xún)出當(dāng)前資源模塊下所有子模塊,遞歸查詢(xún),當(dāng)刪除完所有子模塊下的資源后,再刪除所有子模塊,最終刪除當(dāng)前資源模塊
    deleteSonAuthorityModuleUnderCurrentAuthorityModule(authorityModuleId, iAuthorityService, iRoleAuthorityService);
    //3.刪除當(dāng)前資源模塊
    removeById(authorityModuleId);
}

public void removeAuthorityModuleSeq(Integer authorityModuleId, IAuthorityService iAuthorityService, IRoleAuthorityService iRoleAuthorityService) {
    CompletableFuture.runAsync(()->{
        //兩個(gè)并行執(zhí)行的任務(wù)
        CompletableFuture<Void> future1 = CompletableFuture.runAsync(() ->
                deleteAuthoritiesOfCurrentAuthorityModule(authorityModuleId, iAuthorityService, iRoleAuthorityService),executor);
        CompletableFuture<Void> future2 = CompletableFuture.runAsync(() ->
                deleteSonAuthorityModuleUnderCurrentAuthorityModule(authorityModuleId, iAuthorityService, iRoleAuthorityService), executor);
        //等待兩個(gè)并行任務(wù)執(zhí)行完后,再執(zhí)行最后一個(gè)步驟
        CompletableFuture.allOf(future1,future2).thenRun(()->removeById(authorityModuleId)); 
    },executor);
}

public void removeAuthorityModuleSeq(Integer authorityModuleId, IAuthorityService iAuthorityService, IRoleAuthorityService iRoleAuthorityService) {
    CompletableFuture.runAsync(()->{
        //兩個(gè)并行執(zhí)行的任務(wù)
        CompletableFuture<Void> future1 = CompletableFuture.runAsync(() ->
                deleteAuthoritiesOfCurrentAuthorityModule(authorityModuleId, iAuthorityService, iRoleAuthorityService),executor);
        CompletableFuture<Void> future2 = CompletableFuture.runAsync(() ->
                deleteSonAuthorityModuleUnderCurrentAuthorityModule(authorityModuleId, iAuthorityService, iRoleAuthorityService), executor);
        //等待兩個(gè)并行任務(wù)執(zhí)行完后,再執(zhí)行最后一個(gè)步驟
        CompletableFuture.allOf(future1,future2).thenRun(()->removeById(authorityModuleId));
    },executor);
}

 TransactionStatus getTransaction(@Nullable TransactionDefinition definition)
   throws TransactionException;

protected Object invokeWithinTransaction(Method method, @Nullable Class<?> targetClass,
   final InvocationCallback invocation) throws Throwable {
  //TransactionAttributeSource內(nèi)部保存著當(dāng)前類(lèi)某個(gè)方法對(duì)應(yīng)的TransactionAttribute---事務(wù)屬性源
  //可以看做是一個(gè)存放TransactionAttribute與method方法映射的池子
  TransactionAttributeSource tas = getTransactionAttributeSource();
  //獲取當(dāng)前事務(wù)方法對(duì)應(yīng)的TransactionAttribute
  final TransactionAttribute txAttr = (tas != null ? tas.getTransactionAttribute(method, targetClass) : null);
  //定位TransactionManager
  final TransactionManager tm = determineTransactionManager(txAttr);
        .....
        //類(lèi)型轉(zhuǎn)換為局部事務(wù)管理器
  PlatformTransactionManager ptm = asPlatformTransactionManager(tm);
  final String joinpointIdentification = methodIdentification(method, targetClass, txAttr);
 
  if (txAttr == null || !(ptm instanceof CallbackPreferringPlatformTransactionManager)) {
   //TransactionManager根據(jù)TransactionAttribute創(chuàng)建事務(wù)后返回
   //TransactionInfo封裝了當(dāng)前事務(wù)的信息--包括TransactionStatus
   TransactionInfo txInfo = createTransactionIfNecessary(ptm, txAttr, joinpointIdentification);
 
   Object retVal;
   try {
    //繼續(xù)執(zhí)行過(guò)濾器鏈---過(guò)濾鏈最終會(huì)調(diào)用目標(biāo)方法
    //因此可以理解為這里是調(diào)用目標(biāo)方法
    retVal = invocation.proceedWithInvocation();
   }
   catch (Throwable ex) {
    //目標(biāo)方法拋出異常則進(jìn)行判斷是否需要回滾
    completeTransactionAfterThrowing(txInfo, ex);
    throw ex;
   }
   finally {
       //清除當(dāng)前事務(wù)信息
    cleanupTransactionInfo(txInfo);
   }
            ...
            //正常返回,那么就正常提交事務(wù)唄(當(dāng)然還是需要判斷TransactionStatus狀態(tài)先)
   commitTransactionAfterReturning(txInfo);
   return retVal;
  }
  ...

public class TransactionMain {
    public static void main(String[] args) throws ClassNotFoundException, SQLException {
        test();
    }
 
    private static void test() {
        DataSource dataSource = getDS();
        JdbcTransactionManager jtm = new JdbcTransactionManager(dataSource);
        //JdbcTransactionManager根據(jù)TransactionDefinition信息來(lái)進(jìn)行一些連接屬性的設(shè)置
        //包括隔離級(jí)別和傳播行為等
        DefaultTransactionDefinition transactionDef = new DefaultTransactionDefinition();
        //開(kāi)啟一個(gè)新事務(wù)---此時(shí)autocommit已經(jīng)被設(shè)置為了false,并且當(dāng)前沒(méi)有事務(wù),這里創(chuàng)建的是一個(gè)新事務(wù)
        TransactionStatus ts = jtm.getTransaction(transactionDef);
        //進(jìn)行業(yè)務(wù)邏輯操作
        try {
            update(dataSource);
            jtm.commit(ts);
        }catch (Exception e){
            jtm.rollback(ts);
            System.out.println("發(fā)生異常,我已回滾");
        }
    }
 
    private static void update(DataSource dataSource) throws Exception {
        JdbcTemplate jt = new JdbcTemplate();
        jt.setDataSource(dataSource);
        jt.update("UPDATE Department SET Dname=\"大忽悠\" WHERE id=6");
        throw new Exception("我是來(lái)?yè)v亂的");
    }
}

package com.user.util;
 
import lombok.RequiredArgsConstructor;
import org.springframework.jdbc.datasource.DataSourceTransactionManager;
import org.springframework.stereotype.Component;
import org.springframework.transaction.TransactionStatus;
import org.springframework.transaction.support.DefaultTransactionDefinition;
 
import javax.sql.DataSource;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Executor;
import java.util.concurrent.atomic.AtomicBoolean;
 
/**
 * 多線程事務(wù)一致性管理 <br>
 * 聲明式事務(wù)管理無(wú)法完成,此時(shí)我們只能采用初期的編程式事務(wù)管理才行
 * @author 大忽悠
 * @create 2022/10/19 21:34
 */
@Component
@RequiredArgsConstructor
public class MultiplyThreadTransactionManager {
    /**
     * 如果是多數(shù)據(jù)源的情況下,需要指定具體是哪一個(gè)數(shù)據(jù)源
     */
    private final DataSource dataSource;
 
    /**
     * 執(zhí)行的是無(wú)返回值的任務(wù)
     * @param tasks 異步執(zhí)行的任務(wù)列表
     * @param executor 異步執(zhí)行任務(wù)需要用到的線程池,考慮到線程池需要隔離,這里強(qiáng)制要求傳
     */
    public void runAsyncButWaitUntilAllDown(List<Runnable> tasks, Executor executor) {
        if(executor==null){
            throw new IllegalArgumentException("線程池不能為空");
        }
        DataSourceTransactionManager transactionManager = getTransactionManager();
        //是否發(fā)生了異常
        AtomicBoolean ex=new AtomicBoolean();
 
        List<CompletableFuture> taskFutureList=new ArrayList<>(tasks.size());
        List<TransactionStatus> transactionStatusList=new ArrayList<>(tasks.size());
 
        tasks.forEach(task->{
            taskFutureList.add(CompletableFuture.runAsync(
                    () -> {
                        try{
                            //1.開(kāi)啟新事務(wù)
                            transactionStatusList.add(openNewTransaction(transactionManager));
                            //2.異步任務(wù)執(zhí)行
                            task.run();
                        }catch (Throwable throwable){
                            //打印異常
                            throwable.printStackTrace();
                            //其中某個(gè)異步任務(wù)執(zhí)行出現(xiàn)了異常,進(jìn)行標(biāo)記
                            ex.set(Boolean.TRUE);
                            //其他任務(wù)還沒(méi)執(zhí)行的不需要執(zhí)行了
                            taskFutureList.forEach(completableFuture -> completableFuture.cancel(true));
                        }
                    }
                    , executor)
            );
        });
 
        try {
            //阻塞直到所有任務(wù)全部執(zhí)行結(jié)束---如果有任務(wù)被取消,這里會(huì)拋出異常滴,需要捕獲
            CompletableFuture.allOf(taskFutureList.toArray(new CompletableFuture[]{})).get();
        } catch (InterruptedException | ExecutionException e) {
            e.printStackTrace();
        }
 
        //發(fā)生了異常則進(jìn)行回滾操作,否則提交
        if(ex.get()){
            System.out.println("發(fā)生異常,全部事務(wù)回滾");
            transactionStatusList.forEach(transactionManager::rollback);
        }else {
            System.out.println("全部事務(wù)正常提交");
            transactionStatusList.forEach(transactionManager::commit);
        }
    }
 
    private TransactionStatus openNewTransaction(DataSourceTransactionManager transactionManager) {
        //JdbcTransactionManager根據(jù)TransactionDefinition信息來(lái)進(jìn)行一些連接屬性的設(shè)置
        //包括隔離級(jí)別和傳播行為等
        DefaultTransactionDefinition transactionDef = new DefaultTransactionDefinition();
        //開(kāi)啟一個(gè)新事務(wù)---此時(shí)autocommit已經(jīng)被設(shè)置為了false,并且當(dāng)前沒(méi)有事務(wù),這里創(chuàng)建的是一個(gè)新事務(wù)
        return transactionManager.getTransaction(transactionDef);
    }
 
    private DataSourceTransactionManager getTransactionManager() {
        return new DataSourceTransactionManager(dataSource);
    }
}

public void test(){
    List<Runnable> tasks=new ArrayList<>();
 
    tasks.add(()->{
       userMapper.deleteById(26);
    });
 
    tasks.add(()->{
        signMapper.deleteById(10);
    });
 
    multiplyThreadTransactionManager.runAsyncButWaitUntilAllDown(tasks, Executors.newCachedThreadPool());
}

No value for key [HikariDataSource (HikariPool-1)] bound to thread [main]
解釋: 無(wú)法在當(dāng)前線程綁定的threadLocal中尋找到HikariDataSource作為key,對(duì)應(yīng)關(guān)聯(lián)的資源對(duì)象ConnectionHolder

   //保存當(dāng)前事務(wù)關(guān)聯(lián)的資源--默認(rèn)只會(huì)在新建事務(wù)的時(shí)候保存當(dāng)前獲取到的DataSource和當(dāng)前事務(wù)對(duì)應(yīng)Connection的映射關(guān)系--當(dāng)然這里Connection被包裝為了ConnectionHolder
 private static final ThreadLocal<Map<Object, Object>> resources =
   new NamedThreadLocal<>("Transactional resources");
    //事務(wù)監(jiān)聽(tīng)者--在事務(wù)執(zhí)行到某個(gè)階段的過(guò)程中，會(huì)去回調(diào)監(jiān)聽(tīng)者對(duì)應(yīng)的回調(diào)接口(典型觀察者模式的應(yīng)用)---默認(rèn)為空集合
 private static final ThreadLocal<Set<TransactionSynchronization>> synchronizations =
   new NamedThreadLocal<>("Transaction synchronizations");
   //見(jiàn)名知意: 存放當(dāng)前事務(wù)名字
 private static final ThreadLocal<String> currentTransactionName =
   new NamedThreadLocal<>("Current transaction name");
   //見(jiàn)名知意: 存放當(dāng)前事務(wù)是否是只讀事務(wù)
 private static final ThreadLocal<Boolean> currentTransactionReadOnly =
   new NamedThreadLocal<>("Current transaction read-only status");
   //見(jiàn)名知意: 存放當(dāng)前事務(wù)的隔離級(jí)別
 private static final ThreadLocal<Integer> currentTransactionIsolationLevel =
   new NamedThreadLocal<>("Current transaction isolation level");
   //見(jiàn)名知意: 存放當(dāng)前事務(wù)是否處于激活狀態(tài)
 private static final ThreadLocal<Boolean> actualTransactionActive =
   new NamedThreadLocal<>("Actual transaction active");

package com.user.util;
 
import lombok.Builder;
import lombok.RequiredArgsConstructor;
import org.springframework.jdbc.datasource.DataSourceTransactionManager;
import org.springframework.stereotype.Component;
import org.springframework.transaction.TransactionStatus;
import org.springframework.transaction.support.DefaultTransactionDefinition;
import org.springframework.transaction.support.TransactionSynchronization;
import org.springframework.transaction.support.TransactionSynchronizationManager;
import javax.sql.DataSource;
import java.util.*;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Executor;
import java.util.concurrent.atomic.AtomicBoolean;
 
/**
 * 多線程事務(wù)一致性管理 <br>
 * 聲明式事務(wù)管理無(wú)法完成,此時(shí)我們只能采用初期的編程式事務(wù)管理才行
 * @author 大忽悠
 * @create 2022/10/19 21:34
 */
@Component
@RequiredArgsConstructor
public class MultiplyThreadTransactionManager {
    /**
     * 如果是多數(shù)據(jù)源的情況下,需要指定具體是哪一個(gè)數(shù)據(jù)源
     */
    private final DataSource dataSource;
 
    /**
     * 執(zhí)行的是無(wú)返回值的任務(wù)
     * @param tasks 異步執(zhí)行的任務(wù)列表
     * @param executor 異步執(zhí)行任務(wù)需要用到的線程池,考慮到線程池需要隔離,這里強(qiáng)制要求傳
     */
    public void runAsyncButWaitUntilAllDown(List<Runnable> tasks, Executor executor) {
        if(executor==null){
            throw new IllegalArgumentException("線程池不能為空");
        }
        DataSourceTransactionManager transactionManager = getTransactionManager();
        //是否發(fā)生了異常
        AtomicBoolean ex=new AtomicBoolean();
 
        List<CompletableFuture> taskFutureList=new ArrayList<>(tasks.size());
        List<TransactionStatus> transactionStatusList=new ArrayList<>(tasks.size());
        List<TransactionResource> transactionResources=new ArrayList<>(tasks.size());
 
        tasks.forEach(task->{
            taskFutureList.add(CompletableFuture.runAsync(
                    () -> {
                        try{
                            //1.開(kāi)啟新事務(wù)
                            transactionStatusList.add(openNewTransaction(transactionManager));
                            //2.copy事務(wù)資源
                         transactionResources.add(TransactionResource.copyTransactionResource());
                            //3.異步任務(wù)執(zhí)行
                            task.run();
                        }catch (Throwable throwable){
                            //打印異常
                            throwable.printStackTrace();
                            //其中某個(gè)異步任務(wù)執(zhí)行出現(xiàn)了異常,進(jìn)行標(biāo)記
                            ex.set(Boolean.TRUE);
                            //其他任務(wù)還沒(méi)執(zhí)行的不需要執(zhí)行了
                            taskFutureList.forEach(completableFuture -> completableFuture.cancel(true));
                        }
                    }
                    , executor)
            );
        });
 
        try {
            //阻塞直到所有任務(wù)全部執(zhí)行結(jié)束---如果有任務(wù)被取消,這里會(huì)拋出異常滴,需要捕獲
            CompletableFuture.allOf(taskFutureList.toArray(new CompletableFuture[]{})).get();
        } catch (InterruptedException | ExecutionException e) {
            e.printStackTrace();
        }
 
        //發(fā)生了異常則進(jìn)行回滾操作,否則提交
        if(ex.get()){
            System.out.println("發(fā)生異常,全部事務(wù)回滾");
            for (int i = 0; i < tasks.size(); i++) {
                transactionResources.get(i).autoWiredTransactionResource();
                transactionManager.rollback(transactionStatusList.get(i));
                transactionResources.get(i).removeTransactionResource();
            }
        }else {
            System.out.println("全部事務(wù)正常提交");
            for (int i = 0; i < tasks.size(); i++) {
                transactionResources.get(i).autoWiredTransactionResource();
                transactionManager.commit(transactionStatusList.get(i));
                transactionResources.get(i).removeTransactionResource();
            }
        }
    }
 
    private TransactionStatus openNewTransaction(DataSourceTransactionManager transactionManager) {
        //JdbcTransactionManager根據(jù)TransactionDefinition信息來(lái)進(jìn)行一些連接屬性的設(shè)置
        //包括隔離級(jí)別和傳播行為等
        DefaultTransactionDefinition transactionDef = new DefaultTransactionDefinition();
        //開(kāi)啟一個(gè)新事務(wù)---此時(shí)autocommit已經(jīng)被設(shè)置為了false,并且當(dāng)前沒(méi)有事務(wù),這里創(chuàng)建的是一個(gè)新事務(wù)
        return transactionManager.getTransaction(transactionDef);
    }
 
    private DataSourceTransactionManager getTransactionManager() {
        return new DataSourceTransactionManager(dataSource);
    }
 
    /**
     * 保存當(dāng)前事務(wù)資源,用于線程間的事務(wù)資源COPY操作
     */
    @Builder
    private static class TransactionResource{
        //事務(wù)結(jié)束后默認(rèn)會(huì)移除集合中的DataSource作為key關(guān)聯(lián)的資源記錄
        private  Map<Object, Object> resources = new HashMap<>();
 
        //下面五個(gè)屬性會(huì)在事務(wù)結(jié)束后被自動(dòng)清理,無(wú)需我們手動(dòng)清理
        private  Set<TransactionSynchronization> synchronizations =new HashSet<>();
 
        private  String currentTransactionName;
 
        private Boolean currentTransactionReadOnly;
 
        private Integer currentTransactionIsolationLevel;
 
        private Boolean actualTransactionActive;
 
        public static TransactionResource copyTransactionResource(){
            return TransactionResource.builder()
                    //返回的是不可變集合
                    .resources(TransactionSynchronizationManager.getResourceMap())
                    //如果需要注冊(cè)事務(wù)監(jiān)聽(tīng)者,這里記得修改--我們這里不需要,就采用默認(rèn)負(fù)責(zé)--spring事務(wù)內(nèi)部默認(rèn)也是這個(gè)值
                    .synchronizations(new LinkedHashSet<>())
                    .currentTransactionName(TransactionSynchronizationManager.getCurrentTransactionName())
                    .currentTransactionReadOnly(TransactionSynchronizationManager.isCurrentTransactionReadOnly())
                    .currentTransactionIsolationLevel(TransactionSynchronizationManager.getCurrentTransactionIsolationLevel())
                    .actualTransactionActive(TransactionSynchronizationManager.isActualTransactionActive())
                    .build();
        }
 
        public void autoWiredTransactionResource(){
             resources.forEach(TransactionSynchronizationManager::bindResource);
             //如果需要注冊(cè)事務(wù)監(jiān)聽(tīng)者,這里記得修改--我們這里不需要,就采用默認(rèn)負(fù)責(zé)--spring事務(wù)內(nèi)部默認(rèn)也是這個(gè)值
             TransactionSynchronizationManager.initSynchronization();
             TransactionSynchronizationManager.setActualTransactionActive(actualTransactionActive);
             TransactionSynchronizationManager.setCurrentTransactionName(currentTransactionName);
             TransactionSynchronizationManager.setCurrentTransactionIsolationLevel(currentTransactionIsolationLevel);
             TransactionSynchronizationManager.setCurrentTransactionReadOnly(currentTransactionReadOnly);
        }
 
        public void removeTransactionResource() {
            //事務(wù)結(jié)束后默認(rèn)會(huì)移除集合中的DataSource作為key關(guān)聯(lián)的資源記錄
            //DataSource如果重復(fù)移除,unbindResource時(shí)會(huì)因?yàn)椴淮嬖诖薻ey關(guān)聯(lián)的事務(wù)資源而報(bào)錯(cuò)
            resources.keySet().forEach(key->{
                if(!(key instanceof  DataSource)){
                    TransactionSynchronizationManager.unbindResource(key);
                }
            });
        }
    }
}

@SpringBootTest(classes = UserMain.class)
public class Test {
    @Resource
    private UserMapper userMapper;
    @Resource
    private SignMapper signMapper;
    @Resource
    private MultiplyThreadTransactionManager multiplyThreadTransactionManager;
 
    @SneakyThrows
    @org.junit.jupiter.api.Test
    public void test(){
        List<Runnable> tasks=new ArrayList<>();
 
        tasks.add(()->{
                userMapper.deleteById(26);
                throw new RuntimeException("我就要拋出異常!");
        });
 
        tasks.add(()->{
            signMapper.deleteById(10);
        });
 
        multiplyThreadTransactionManager.runAsyncButWaitUntilAllDown(tasks, Executors.newCachedThreadPool());
    }
 
}