SpringBoot中使用线程池控制主线程与子线程事务的全过程

// ThreadLocal在Spring事务管理中的应用示例
public abstract class TransactionSynchronizationManager {
    private static final ThreadLocal<Map<Object, Object>> resources = 
        new NamedThreadLocal<>("Transactional resources");
    private static final ThreadLocal<Set<TransactionSynchronization>> synchronizations = 
        new NamedThreadLocal<>("Transaction synchronizations");
    private static final ThreadLocal<String> currentTransactionName = 
        new NamedThreadLocal<>("Current transaction name");
    private static final ThreadLocal<Boolean> currentTransactionReadOnly = 
        new NamedThreadLocal<>("Current transaction read-only status");
    private static final ThreadLocal<Integer> currentTransactionIsolationLevel = 
        new NamedThreadLocal<>("Current transaction isolation level");
    private static final ThreadLocal<Boolean> actualTransactionActive = 
        new NamedThreadLocal<>("Actual transaction active");
}

─────────────────────────────────────────────────────────────┐
│                    Spring Transaction Architecture           │
├─────────────────────────────────────────────────────────────┤
│  @Transactional                                             │
│         │                                                   │
│         ▼                                                   │
│  TransactionInterceptor                                     │
│         │                                                   │
│         ▼                                                   │
│  PlatformTransactionManager                                 │
│         │                                                   │
│         ▼                                                   │
│  DataSourceTransactionManager / JpaTransactionManager / etc │
│         │                                                   │
│         ▼                                                   │
│  JDBC Connection / JPA EntityManager                        │
└─────────────────────────────────────────────────────────────┘

@Service
public class UserService {
    @Autowired
    private UserRepository userRepository;
    @Transactional
    public User createUser(User user) {
        // 业务逻辑
        return userRepository.save(user);
    }
    @Transactional(readOnly = true)
    public User findUserById(Long id) {
        return userRepository.findById(id).orElse(null);
    }
}

@Service
public class OrderService {
    @Autowired
    private PlatformTransactionManager transactionManager;
    @Autowired
    private TransactionTemplate transactionTemplate;
    public void processOrder(Long orderId) {
        // 方式1：使用TransactionTemplate
        transactionTemplate.execute(status -> {
            // 业务逻辑
            return null;
        });
        // 方式2：使用PlatformTransactionManager
        TransactionDefinition def = new DefaultTransactionDefinition();
        TransactionStatus status = transactionManager.getTransaction(def);
        try {
            // 业务逻辑
            transactionManager.commit(status);
        } catch (Exception e) {
            transactionManager.rollback(status);
            throw e;
        }
    }
}

@Service
public class ComplexService {
    @Transactional(propagation = Propagation.REQUIRED)
    public void methodA() {
        // 方法A的业务逻辑
        methodB(); // 调用方法B
    }
    @Transactional(propagation = Propagation.REQUIRES_NEW)
    public void methodB() {
        // 方法B将在独立的事务中执行
        // 即使methodA回滚，methodB的提交也不会被影响
    }
}

@Configuration
@EnableAsync
public class ThreadPoolConfig {
    /**
     * 核心业务线程池
     */
    @Bean("businessExecutor")
    public ThreadPoolTaskExecutor businessExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        // 核心线程数：线程池维护的最小线程数量
        executor.setCorePoolSize(10);
        // 最大线程数：线程池允许的最大线程数量
        executor.setMaxPoolSize(50);
        // 队列容量：当线程数达到核心线程数时，新任务会进入队列等待
        executor.setQueueCapacity(100);
        // 线程名前缀：方便日志追踪
        executor.setThreadNamePrefix("business-thread-");
        // 拒绝策略：当线程池和队列都满了时的处理策略
        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        // 线程空闲时间：非核心线程空闲存活时间
        executor.setKeepAliveSeconds(60);
        // 等待所有任务完成后关闭线程池
        executor.setWaitForTasksToCompleteOnShutdown(true);
        // 等待任务完成的超时时间
        executor.setAwaitTerminationSeconds(60);
        // 初始化线程池
        executor.initialize();
        return executor;
    }
    /**
     * 事务处理专用线程池
     */
    @Bean("transactionExecutor")
    public ThreadPoolTaskExecutor transactionExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(5);
        executor.setMaxPoolSize(20);
        executor.setQueueCapacity(50);
        executor.setThreadNamePrefix("transaction-thread-");
        // 使用自定义拒绝策略
        executor.setRejectedExecutionHandler(new RejectedExecutionHandler() {
            @Override
            public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
                // 记录日志
                log.warn("Transaction task rejected: {}", r.toString());
                // 尝试重新执行
                if (!executor.isShutdown()) {
                    try {
                        executor.getQueue().put(r);
                    } catch (InterruptedException e) {
                        Thread.currentThread().interrupt();
                    }
                }
            }
        });
        executor.initialize();
        return executor;
    }
}

@Service
public class AsyncService {
    @Async("businessExecutor")
    public CompletableFuture<String> asyncMethodWithReturn(String param) {
        log.info("Async method started with param: {}", param);
        try {
            Thread.sleep(1000); // 模拟耗时操作
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
        return CompletableFuture.completedFuture("Result for " + param);
    }
    @Async("transactionExecutor")
    public void asyncMethodWithoutReturn() {
        log.info("Async method without return value started");
        // 执行业务逻辑
    }
}

@Component
public class ThreadPoolMonitor {
    @Autowired
    @Qualifier("businessExecutor")
    private ThreadPoolTaskExecutor businessExecutor;
    @Scheduled(fixedRate = 30000) // 每30秒监控一次
    public void monitorThreadPool() {
        ThreadPoolExecutor executor = businessExecutor.getThreadPoolExecutor();
        log.info("====== Thread Pool Monitor ======");
        log.info("Active Threads: {}", executor.getActiveCount());
        log.info("Pool Size: {}", executor.getPoolSize());
        log.info("Core Pool Size: {}", executor.getCorePoolSize());
        log.info("Maximum Pool Size: {}", executor.getMaximumPoolSize());
        log.info("Queue Size: {}", executor.getQueue().size());
        log.info("Completed Tasks: {}", executor.getCompletedTaskCount());
        log.info("Total Tasks: {}", executor.getTaskCount());
        log.info("================================");
        // 如果队列使用率过高，可以动态调整
        double queueUsage = (double) executor.getQueue().size() / executor.getQueue().remainingCapacity();
        if (queueUsage > 0.8) {
            log.warn("Thread pool queue usage is high: {}%", queueUsage * 100);
        }
    }
}

@Service
public class ProblematicService {
    @Transactional
    public void mainMethod() {
        // 主线程事务开始
        log.info("Main thread transaction active: {}", 
                 TransactionSynchronizationManager.isActualTransactionActive());
        // 创建子线程
        new Thread(() -> {
            // 子线程中无法访问主线程的事务上下文
            log.info("Child thread transaction active: {}", 
                     TransactionSynchronizationManager.isActualTransactionActive());
            // 这里会抛出异常：没有活动的事务
            // 尝试数据库操作会失败
        }).start();
    }
}

@Service
public class TransactionPropagationService {
    @Transactional
    public void processWithContextPropagation() {
        // 获取当前事务的属性
        TransactionAttribute transactionAttribute = 
            TransactionAspectSupport.currentTransactionStatus().getTransactionAttribute();
        // 获取当前事务的隔离级别、超时时间等属性
        Integer isolationLevel = TransactionSynchronizationManager.getCurrentTransactionIsolationLevel();
        Integer timeout = transactionAttribute.getTimeout();
        boolean readOnly = TransactionSynchronizationManager.isCurrentTransactionReadOnly();
        // 将事务属性传递给子线程
        TransactionContext context = new TransactionContext();
        context.setIsolationLevel(isolationLevel);
        context.setTimeout(timeout);
        context.setReadOnly(readOnly);
        // 在子线程中重新创建事务
        CompletableFuture.runAsync(() -> {
            executeInNewTransaction(context, () -> {
                // 子线程的业务逻辑
                log.info("Child thread executing with transaction");
            });
        });
    }
    @Data
    private static class TransactionContext {
        private Integer isolationLevel;
        private Integer timeout;
        private Boolean readOnly;
    }
}

@Component
public class TransactionContextHolder {
    // 注意：InheritableThreadLocal有内存泄漏风险，需要谨慎使用
    private static final InheritableThreadLocal<Map<String, Object>> context = 
        new InheritableThreadLocal<>() {
            @Override
            protected Map<String, Object> childValue(Map<String, Object> parentValue) {
                // 深度拷贝，避免父子线程共享同一对象
                return parentValue != null ? new HashMap<>(parentValue) : null;
            }
        };
    public static void set(String key, Object value) {
        Map<String, Object> map = context.get();
        if (map == null) {
            map = new HashMap<>();
            context.set(map);
        }
        map.put(key, value);
    }
    public static Object get(String key) {
        Map<String, Object> map = context.get();
        return map != null ? map.get(key) : null;
    }
    public static void clear() {
        context.remove();
    }
}

@Service
public class ProgrammaticTransactionService {
    @Autowired
    private PlatformTransactionManager transactionManager;
    @Autowired
    private JdbcTemplate jdbcTemplate;
    public void processWithMultipleThreads(List<Task> tasks) {
        // 主线程事务
        DefaultTransactionDefinition definition = new DefaultTransactionDefinition();
        definition.setPropagationBehavior(TransactionDefinition.PROPAGATION_REQUIRED);
        definition.setIsolationLevel(TransactionDefinition.ISOLATION_READ_COMMITTED);
        TransactionStatus mainStatus = transactionManager.getTransaction(definition);
        try {
            // 主线程业务逻辑
            executeMainLogic();
            // 创建子线程任务
            List<CompletableFuture<Void>> futures = new ArrayList<>();
            for (Task task : tasks) {
                CompletableFuture<Void> future = CompletableFuture.runAsync(() -> {
                    // 每个子线程有自己的事务
                    DefaultTransactionDefinition childDefinition = new DefaultTransactionDefinition();
                    childDefinition.setPropagationBehavior(TransactionDefinition.PROPAGATION_REQUIRES_NEW);
                    childDefinition.setIsolationLevel(TransactionDefinition.ISOLATION_READ_COMMITTED);
                    TransactionStatus childStatus = transactionManager.getTransaction(childDefinition);
                    try {
                        executeChildLogic(task);
                        transactionManager.commit(childStatus);
                    } catch (Exception e) {
                        transactionManager.rollback(childStatus);
                        throw new RuntimeException("Child thread transaction failed", e);
                    }
                });
                futures.add(future);
            }
            // 等待所有子线程完成
            CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).join();
            // 提交主事务
            transactionManager.commit(mainStatus);
        } catch (Exception e) {
            // 回滚主事务
            transactionManager.rollback(mainStatus);
            throw new RuntimeException("Main transaction failed", e);
        }
    }
    private void executeMainLogic() {
        // 主线程业务逻辑实现
        jdbcTemplate.update("INSERT INTO main_table (data) VALUES (?)", "main data");
    }
    private void executeChildLogic(Task task) {
        // 子线程业务逻辑实现
        jdbcTemplate.update("INSERT INTO child_table (task_id, data) VALUES (?, ?)", 
                           task.getId(), task.getData());
    }
}

@Service
public class CoordinatedTransactionService {
    @Autowired
    private DataSource dataSource;
    @Autowired
    private PlatformTransactionManager transactionManager;
    /**
     * 策略：主线程等待所有子线程事务成功后才提交
     */
    public void coordinatedStrategy1(List<BusinessTask> tasks) {
        // 用于收集子线程执行结果
        List<CompletableFuture<Boolean>> futures = new ArrayList<>();
        // 创建CountDownLatch用于等待所有子线程完成
        CountDownLatch latch = new CountDownLatch(tasks.size());
        // 创建共享异常收集器
        AtomicReference<Exception> sharedException = new AtomicReference<>();
        // 启动主事务
        DefaultTransactionDefinition def = new DefaultTransactionDefinition();
        TransactionStatus mainStatus = transactionManager.getTransaction(def);
        try {
            // 执行主线程逻辑
            executeMainBusiness();
            // 启动子线程
            for (BusinessTask task : tasks) {
                CompletableFuture<Boolean> future = CompletableFuture.supplyAsync(() -> {
                    try {
                        // 每个子线程使用独立的事务
                        return executeChildTransaction(task);
                    } catch (Exception e) {
                        sharedException.set(e);
                        return false;
                    } finally {
                        latch.countDown();
                    }
                });
                futures.add(future);
            }
            // 等待所有子线程完成
            boolean completed = latch.await(30, TimeUnit.SECONDS);
            if (!completed) {
                throw new TimeoutException("Child threads timeout");
            }
            // 检查是否有子线程失败
            if (sharedException.get() != null) {
                throw new RuntimeException("Child thread failed", sharedException.get());
            }
            // 检查所有子线程结果
            for (CompletableFuture<Boolean> future : futures) {
                if (!future.get()) {
                    throw new RuntimeException("At least one child thread failed");
                }
            }
            // 提交主事务
            transactionManager.commit(mainStatus);
        } catch (Exception e) {
            // 回滚主事务
            transactionManager.rollback(mainStatus);
            throw new RuntimeException("Coordinated transaction failed", e);
        }
    }
    private boolean executeChildTransaction(BusinessTask task) {
        // 子线程使用独立的事务
        DefaultTransactionDefinition def = new DefaultTransactionDefinition();
        def.setPropagationBehavior(TransactionDefinition.PROPAGATION_REQUIRES_NEW);
        TransactionStatus status = transactionManager.getTransaction(def);
        try {
            // 执行子线程业务逻辑
            processTask(task);
            transactionManager.commit(status);
            return true;
        } catch (Exception e) {
            transactionManager.rollback(status);
            log.error("Child transaction failed for task: {}", task.getId(), e);
            return false;
        }
    }
}

@Service
public class TwoPhaseCommitService {
    @Autowired
    private DataSource dataSource;
    /**
     * 简化的两阶段提交实现
     */
    public void twoPhaseCommitStrategy(List<Runnable> tasks) {
        // 第一阶段：准备阶段
        List<CompletableFuture<Boolean>> prepareFutures = new ArrayList<>();
        List<TransactionStatus> childStatuses = Collections.synchronizedList(new ArrayList<>());
        try {
            // 主事务开始
            Connection mainConn = dataSource.getConnection();
            mainConn.setAutoCommit(false);
            try {
                // 主线程准备
                prepareMainPhase(mainConn);
                // 子线程准备
                for (Runnable task : tasks) {
                    CompletableFuture<Boolean> future = CompletableFuture.supplyAsync(() -> {
                        try {
                            Connection childConn = dataSource.getConnection();
                            childConn.setAutoCommit(false);
                            // 执行准备操作
                            boolean prepared = prepareChildPhase(childConn, task);
                            if (prepared) {
                                // 保存连接和状态，用于第二阶段
                                childStatuses.add(new TransactionStatus(childConn));
                            }
                            return prepared;
                        } catch (SQLException e) {
                            log.error("Child prepare phase failed", e);
                            return false;
                        }
                    });
                    prepareFutures.add(future);
                }
                // 等待所有准备阶段完成
                CompletableFuture<Void> allPrepare = CompletableFuture.allOf(
                    prepareFutures.toArray(new CompletableFuture[0])
                );
                allPrepare.get(10, TimeUnit.SECONDS);
                // 检查所有子线程是否准备成功
                boolean allPrepared = prepareFutures.stream()
                    .allMatch(f -> {
                        try {
                            return f.get();
                        } catch (Exception e) {
                            return false;
                        }
                    });
                if (!allPrepared) {
                    // 有任何失败，执行回滚
                    rollbackAll(mainConn, childStatuses);
                    throw new RuntimeException("Prepare phase failed");
                }
                // 第二阶段：提交阶段
                commitAll(mainConn, childStatuses);
            } catch (Exception e) {
                mainConn.rollback();
                throw new RuntimeException("Two-phase commit failed", e);
            } finally {
                mainConn.close();
            }
        } catch (SQLException e) {
            throw new RuntimeException("Database connection error", e);
        }
    }
    private void commitAll(Connection mainConn, List<TransactionStatus> childStatuses) throws SQLException {
        try {
            // 先提交所有子事务
            for (TransactionStatus status : childStatuses) {
                status.getConnection().commit();
                status.getConnection().close();
            }
            // 最后提交主事务
            mainConn.commit();
        } catch (SQLException e) {
            // 提交失败，尝试回滚所有
            try {
                mainConn.rollback();
            } catch (SQLException ex) {
                log.error("Failed to rollback main connection", ex);
            }
            throw e;
        }
    }
    @Data
    private static class TransactionStatus {
        private final Connection connection;
        private final long threadId = Thread.currentThread().getId();
        private final LocalDateTime createTime = LocalDateTime.now();
    }
}

@Service
@Slf4j
public class CompensationTransactionService {
    @Autowired
    private JdbcTemplate jdbcTemplate;
    /**
     * 补偿事务模式：记录所有操作，失败时执行补偿
     */
    public void compensationStrategy(List<BusinessOperation> operations) {
        // 用于记录需要补偿的操作
        List<CompensationAction> compensationActions = Collections.synchronizedList(new ArrayList<>());
        // 主事务开始
        DefaultTransactionDefinition mainDef = new DefaultTransactionDefinition();
        PlatformTransactionManager transactionManager = 
            new DataSourceTransactionManager(Objects.requireNonNull(jdbcTemplate.getDataSource()));
        TransactionStatus mainStatus = transactionManager.getTransaction(mainDef);
        try {
            // 执行主操作
            CompensationAction mainAction = executeMainOperation();
            compensationActions.add(mainAction);
            // 并行执行子操作
            List<CompletableFuture<CompensationAction>> futures = operations.stream()
                .map(op -> CompletableFuture.supplyAsync(() -> {
                    try {
                        return executeChildOperation(op);
                    } catch (Exception e) {
                        throw new CompletionException(e);
                    }
                }))
                .collect(Collectors.toList());
            // 等待所有子操作完成
            List<CompensationAction> childActions = futures.stream()
                .map(f -> {
                    try {
                        return f.get();
                    } catch (Exception e) {
                        throw new RuntimeException("Child operation failed", e);
                    }
                })
                .collect(Collectors.toList());
            compensationActions.addAll(childActions);
            // 所有操作成功，提交主事务
            transactionManager.commit(mainStatus);
            // 记录成功日志
            logCompensationSuccess(compensationActions);
        } catch (Exception e) {
            // 回滚主事务
            transactionManager.rollback(mainStatus);
            // 执行补偿操作
            executeCompensations(compensationActions);
            throw new RuntimeException("Transaction failed, compensation executed", e);
        }
    }
    private CompensationAction executeMainOperation() {
        // 执行业务操作，并返回补偿动作
        String operationId = UUID.randomUUID().toString();
        try {
            // 业务逻辑
            jdbcTemplate.update("INSERT INTO main_operations (id, data) VALUES (?, ?)", 
                               operationId, "main data");
            // 返回补偿动作
            return CompensationAction.builder()
                .operationId(operationId)
                .operationType("INSERT_MAIN")
                .compensationSql("DELETE FROM main_operations WHERE id = ?")
                .compensationParams(new Object[]{operationId})
                .build();
        } catch (Exception e) {
            throw new RuntimeException("Main operation failed", e);
        }
    }
    private void executeCompensations(List<CompensationAction> actions) {
        // 按照操作的反向顺序执行补偿
        Collections.reverse(actions);
        for (CompensationAction action : actions) {
            try {
                jdbcTemplate.update(action.getCompensationSql(), action.getCompensationParams());
                log.info("Compensation executed for operation: {}", action.getOperationId());
            } catch (Exception e) {
                log.error("Failed to execute compensation for operation: {}", 
                         action.getOperationId(), e);
                // 继续执行其他补偿，不中断
            }
        }
    }
    @Data
    @Builder
    private static class CompensationAction {
        private String operationId;
        private String operationType;
        private String compensationSql;
        private Object[] compensationParams;
        private LocalDateTime operationTime;
    }
}

@Service
public class TransactionSynchronizationService {
    @Autowired
    private DataSource dataSource;
    /**
     * 使用TransactionSynchronization协调多线程事务
     */
    @Transactional
    public void processWithSynchronization(List<SubTask> subTasks) {
        // 注册事务同步器
        TransactionSynchronizationManager.registerSynchronization(
            new CustomTransactionSynchronization(subTasks)
        );
        // 主线程业务逻辑
        executeMainBusiness();
        // 注意：子线程操作将在事务提交前执行
        // TransactionSynchronization.beforeCommit()中启动子线程
    }
    private class CustomTransactionSynchronization implements TransactionSynchronization {
        private final List<SubTask> subTasks;
        private final ExecutorService executorService;
        private final List<Future<?>> futures;
        public CustomTransactionSynchronization(List<SubTask> subTasks) {
            this.subTasks = subTasks;
            this.executorService = Executors.newFixedThreadPool(subTasks.size());
            this.futures = new ArrayList<>();
        }
        @Override
        public void beforeCommit(boolean readOnly) {
            log.info("TransactionSynchronization.beforeCommit called");
            // 在事务提交前启动子线程
            for (SubTask task : subTasks) {
                Future<?> future = executorService.submit(() -> {
                    try {
                        // 每个子线程使用独立连接和事务
                        executeSubTaskInNewTransaction(task);
                    } catch (Exception e) {
                        log.error("Subtask execution failed", e);
                        throw new RuntimeException(e);
                    }
                });
                futures.add(future);
            }
            // 等待所有子线程完成
            for (Future<?> future : futures) {
                try {
                    future.get(10, TimeUnit.SECONDS);
                } catch (Exception e) {
                    throw new RuntimeException("Failed to complete subtasks", e);
                }
            }
            executorService.shutdown();
        }
        @Override
        public void afterCompletion(int status) {
            log.info("TransactionSynchronization.afterCompletion called with status: {}", 
                     status == STATUS_COMMITTED ? "COMMITTED" : "ROLLED_BACK");
            if (status == STATUS_ROLLED_BACK) {
                // 事务回滚，需要清理子线程可能已经提交的操作
                log.warn("Main transaction rolled back, but child transactions may have been committed");
                // 这里可以实现补偿逻辑
            }
            cleanup();
        }
        private void cleanup() {
            if (!executorService.isShutdown()) {
                executorService.shutdownNow();
            }
        }
    }
}

@Component
@Slf4j
public class TransactionEventListenerService {
    @Autowired
    private ApplicationEventPublisher eventPublisher;
    @TransactionalEventListener(phase = TransactionPhase.BEFORE_COMMIT)
    public void handleBeforeCommit(TransactionEvent event) {
        log.info("Before commit event received");
        // 在事务提交前执行操作
        prepareForCommit();
    }
    @TransactionalEventListener(phase = TransactionPhase.AFTER_COMPLETION)
    public void handleAfterCompletion(TransactionCompletionEvent event) {
        log.info("Transaction completed with status: {}", 
                 event.getTransactionResult() == TransactionResult.COMMITTED ? "COMMITTED" : "ROLLED_BACK");
        if (event.getTransactionResult() == TransactionResult.COMMITTED) {
            // 事务提交后执行异步操作
            executePostCommitOperations(event.getBusinessData());
        } else {
            // 事务回滚后的清理操作
            executeRollbackCleanup(event.getBusinessData());
        }
    }
    @Async
    public void executePostCommitOperations(BusinessData data) {
        // 异步执行提交后的操作
        log.info("Executing post-commit operations asynchronously");
        // 这里可以启动子线程进行后续处理
    }
    @Data
    public static class TransactionEvent {
        private final String transactionId;
        private final LocalDateTime eventTime;
        private final BusinessData businessData;
    }
    @Data
    public static class TransactionCompletionEvent extends TransactionEvent {
        private final TransactionResult transactionResult;
    }
    public enum TransactionResult {
        COMMITTED,
        ROLLED_BACK
    }
}

@Service
@Slf4j
public class SeataDistributedTransactionService {
    @Autowired
    private UserService userService;
    @Autowired
    private OrderService orderService;
    @Autowired
    private InventoryService inventoryService;
    /**
     * 使用Seata AT模式处理多线程分布式事务
     * 注意：Seata默认不支持多线程，需要特殊处理
     */
    @GlobalTransactional(timeoutMills = 300000, name = "multi-thread-purchase")
    public void purchaseWithMultipleThreads(PurchaseRequest request) {
        // 获取全局事务ID
        String xid = RootContext.getXID();
        log.info("Global transaction started, xid: {}", xid);
        // 用于收集子线程执行结果
        List<CompletableFuture<Boolean>> futures = new ArrayList<>();
        try {
            // 任务1：扣减库存（异步执行）
            CompletableFuture<Boolean> inventoryFuture = CompletableFuture.supplyAsync(() -> {
                // 传播全局事务ID到子线程
                RootContext.bind(xid);
                try {
                    return inventoryService.deduct(request.getProductId(), request.getQuantity());
                } finally {
                    RootContext.unbind();
                }
            });
            futures.add(inventoryFuture);
            // 任务2：创建订单（异步执行）
            CompletableFuture<Boolean> orderFuture = CompletableFuture.supplyAsync(() -> {
                RootContext.bind(xid);
                try {
                    return orderService.createOrder(request);
                } finally {
                    RootContext.unbind();
                }
            });
            futures.add(orderFuture);
            // 任务3：更新用户信息（主线程执行）
            boolean userUpdated = userService.updatePurchaseInfo(request.getUserId(), request.getAmount());
            if (!userUpdated) {
                throw new RuntimeException("Failed to update user info");
            }
            // 等待所有异步任务完成
            CompletableFuture<Void> allFutures = CompletableFuture.allOf(
                futures.toArray(new CompletableFuture[0])
            );
            allFutures.get(30, TimeUnit.SECONDS);
            // 检查所有任务结果
            for (CompletableFuture<Boolean> future : futures) {
                if (!future.get()) {
                    throw new RuntimeException("One of the async tasks failed");
                }
            }
            log.info("All distributed transactions completed successfully");
        } catch (Exception e) {
            log.error("Distributed transaction failed", e);
            // Seata会自动回滚所有分支事务
            throw new RuntimeException("Purchase failed", e);
        } finally {
            // 清理上下文
            RootContext.unbind();
        }
    }
}
// Seata配置类
@Configuration
public class SeataConfig {
    @Bean
    public GlobalTransactionScanner globalTransactionScanner() {
        return new GlobalTransactionScanner("multi-thread-app", "my_test_tx_group");
    }
    /**
     * 自定义DataSourceProxy以支持多线程
     */
    @Bean
    public DataSource dataSource(DataSource druidDataSource) {
        return new DataSourceProxy(druidDataSource);
    }
}

@Service
@Slf4j
public class MQBasedTransactionService {
    @Autowired
    private RabbitTemplate rabbitTemplate;
    @Autowired
    private JdbcTemplate jdbcTemplate;
    /**
     * 基于消息队列的最终一致性方案
     */
    @Transactional
    public void processWithMQ(List<SubTask> tasks) {
        // 1. 主事务操作
        executeMainTransaction();
        // 2. 发送准备消息（不投递）
        List<String> messageIds = new ArrayList<>();
        for (SubTask task : tasks) {
            String messageId = sendPrepareMessage(task);
            messageIds.add(messageId);
        }
        // 3. 本地记录消息状态
        saveMessageStatus(messageIds, MessageStatus.PREPARED);
        // 4. 提交主事务（消息仍未投递）
        // 事务提交后，下面的代码才会执行
        // 5. 事务提交后，确认投递消息
        TransactionSynchronizationManager.registerSynchronization(
            new TransactionSynchronization() {
                @Override
                public void afterCommit() {
                    // 确认投递所有消息
                    for (String messageId : messageIds) {
                        confirmMessageDelivery(messageId);
                        updateMessageStatus(messageId, MessageStatus.CONFIRMED);
                    }
                    // 异步执行子任务
                    executeSubTasksAsync(tasks);
                }
                @Override
                public void afterCompletion(int status) {
                    if (status == STATUS_ROLLED_BACK) {
                        // 取消所有消息
                        for (String messageId : messageIds) {
                            cancelMessage(messageId);
                            updateMessageStatus(messageId, MessageStatus.CANCELLED);
                        }
                    }
                }
            }
        );
    }
    private void executeSubTasksAsync(List<SubTask> tasks) {
        ExecutorService executor = Executors.newFixedThreadPool(tasks.size());
        List<CompletableFuture<Void>> futures = tasks.stream()
            .map(task -> CompletableFuture.runAsync(() -> {
                try {
                    // 每个子线程处理自己的任务
                    processSubTask(task);
                } catch (Exception e) {
                    log.error("Subtask processing failed", e);
                    // 发送补偿消息
                    sendCompensationMessage(task);
                }
            }, executor))
            .collect(Collectors.toList());
        // 等待所有任务完成
        CompletableFuture.allOf(futures.toArray(new CompletableFuture[0]))
            .thenRun(() -> {
                executor.shutdown();
                log.info("All subtasks completed");
            })
            .exceptionally(ex -> {
                log.error("Failed to complete all subtasks", ex);
                executor.shutdownNow();
                return null;
            });
    }
    private enum MessageStatus {
        PREPARED,
        CONFIRMED,
        CANCELLED,
        COMPLETED
    }
}

# application.yml 线程池配置
spring:
  task:
    execution:
      pool:
        # 公共线程池配置
        common:
          core-size: 10
          max-size: 50
          queue-capacity: 1000
          keep-alive: 60s
          thread-name-prefix: "common-task-"
        # 事务处理专用线程池
        transaction:
          core-size: 5
          max-size: 20
          queue-capacity: 500
          keep-alive: 30s
          thread-name-prefix: "tx-task-"
          allow-core-thread-timeout: true
        # IO密集型任务线程池
        io-intensive:
          core-size: 20
          max-size: 100
          queue-capacity: 2000
          keep-alive: 120s
          thread-name-prefix: "io-task-"
# 线程池监控配置
management:
  endpoints:
    web:
      exposure:
        include: "health,info,metrics,threadpool"
  metrics:
    export:
      prometheus:
        enabled: true

@Configuration
@EnableTransactionManagement
public class TransactionOptimizationConfig {
    /**
     * 事务管理器配置优化
     */
    @Bean
    public PlatformTransactionManager transactionManager(DataSource dataSource) {
        DataSourceTransactionManager transactionManager = new DataSourceTransactionManager(dataSource);
        // 优化配置
        transactionManager.setNestedTransactionAllowed(true); // 允许嵌套事务
        transactionManager.setValidateExistingTransaction(true); // 验证已有事务
        transactionManager.setGlobalRollbackOnParticipationFailure(false); // 优化参与失败时的回滚行为
        return transactionManager;
    }
    /**
     * 事务模板配置
     */
    @Bean
    public TransactionTemplate transactionTemplate(PlatformTransactionManager transactionManager) {
        TransactionTemplate template = new TransactionTemplate(transactionManager);
        // 设置默认事务属性
        template.setPropagationBehavior(TransactionDefinition.PROPAGATION_REQUIRED);
        template.setIsolationLevel(TransactionDefinition.ISOLATION_READ_COMMITTED);
        template.setTimeout(30); // 30秒超时
        // 只读事务优化
        template.setReadOnly(false);
        return template;
    }
    /**
     * 事务拦截器优化
     */
    @Bean
    public TransactionInterceptor transactionInterceptor(PlatformTransactionManager transactionManager) {
        TransactionInterceptor interceptor = new TransactionInterceptor();
        interceptor.setTransactionManager(transactionManager);
        // 配置事务属性源
        NameMatchTransactionAttributeSource source = new NameMatchTransactionAttributeSource();
        // 查询方法使用只读事务
        RuleBasedTransactionAttribute readOnlyAttr = new RuleBasedTransactionAttribute();
        readOnlyAttr.setReadOnly(true);
        readOnlyAttr.setPropagationBehavior(TransactionDefinition.PROPAGATION_SUPPORTS);
        // 写操作使用读写事务
        RuleBasedTransactionAttribute writeAttr = new RuleBasedTransactionAttribute();
        writeAttr.setPropagationBehavior(TransactionDefinition.PROPAGATION_REQUIRED);
        writeAttr.setIsolationLevel(TransactionDefinition.ISOLATION_READ_COMMITTED);
        writeAttr.setTimeout(30);
        // 方法名模式匹配
        source.addTransactionalMethod("get*", readOnlyAttr);
        source.addTransactionalMethod("find*", readOnlyAttr);
        source.addTransactionalMethod("query*", readOnlyAttr);
        source.addTransactionalMethod("save*", writeAttr);
        source.addTransactionalMethod("update*", writeAttr);
        source.addTransactionalMethod("delete*", writeAttr);
        source.addTransactionalMethod("process*", writeAttr);
        interceptor.setTransactionAttributeSource(source);
        return interceptor;
    }
}

@Component
@Slf4j
public class TransactionMonitor {
    @Autowired
    private MeterRegistry meterRegistry;
    @Autowired
    private PlatformTransactionManager transactionManager;
    private final Map<String, AtomicInteger> transactionCounters = new ConcurrentHashMap<>();
    private final Map<String, AtomicLong> transactionDurations = new ConcurrentHashMap<>();
    /**
     * 事务监控AOP
     */
    @Aspect
    @Component
    public static class TransactionMonitoringAspect {
        private final ThreadLocal<Long> startTime = new ThreadLocal<>();
        private final TransactionMonitor monitor;
        public TransactionMonitoringAspect(TransactionMonitor monitor) {
            this.monitor = monitor;
        }
        @Around("@annotation(org.springframework.transaction.annotation.Transactional)")
        public Object monitorTransaction(ProceedingJoinPoint joinPoint) throws Throwable {
            String methodName = joinPoint.getSignature().toShortString();
            String transactionName = extractTransactionName(methodName);
            // 记录开始时间
            startTime.set(System.currentTimeMillis());
            try {
                // 增加事务计数器
                monitor.incrementTransactionCounter(transactionName);
                // 执行原方法
                Object result = joinPoint.proceed();
                // 记录成功
                monitor.recordTransactionSuccess(transactionName, 
                    System.currentTimeMillis() - startTime.get());
                return result;
            } catch (Exception e) {
                // 记录失败
                monitor.recordTransactionFailure(transactionName, 
                    System.currentTimeMillis() - startTime.get(), e);
                throw e;
            } finally {
                startTime.remove();
            }
        }
        private String extractTransactionName(String methodName) {
            // 简化的方法名提取逻辑
            return methodName.replaceAll(".*\\.", "").replaceAll("\\(.*\\)", "");
        }
    }
    public void incrementTransactionCounter(String transactionName) {
        transactionCounters
            .computeIfAbsent(transactionName, k -> new AtomicInteger(0))
            .incrementAndGet();
        // 发布到监控系统
        meterRegistry.counter("transactions.total", "name", transactionName).increment();
    }
    public void recordTransactionSuccess(String transactionName, long duration) {
        transactionDurations
            .computeIfAbsent(transactionName, k -> new AtomicLong(0))
            .addAndGet(duration);
        // 发布到监控系统
        meterRegistry.timer("transactions.duration", "name", transactionName, "status", "success")
            .record(duration, TimeUnit.MILLISECONDS);
        // 检查性能阈值
        if (duration > 1000) { // 超过1秒告警
            log.warn("Slow transaction detected: {} took {}ms", transactionName, duration);
        }
    }
    /**
     * 生成监控报告
     */
    @Scheduled(fixedDelay = 60000) // 每分钟生成一次报告
    public void generateMonitoringReport() {
        Map<String, Object> report = new HashMap<>();
        transactionCounters.forEach((name, counter) -> {
            long count = counter.getAndSet(0);
            long totalDuration = transactionDurations.getOrDefault(name, new AtomicLong(0))
                .getAndSet(0);
            long avgDuration = count > 0 ? totalDuration / count : 0;
            report.put(name, Map.of(
                "count", count,
                "avgDuration", avgDuration,
                "tps", count / 60.0
            ));
        });
        log.info("Transaction monitoring report: {}", report);
        // 发送到监控系统
        sendToMonitoringSystem(report);
    }
}

@Service
public class TransactionRollbackSolution {
    @Transactional(rollbackFor = Exception.class)
    public void processWithRollbackControl(List<Task> tasks) {
        // 使用CompletableFuture收集异常
        List<CompletableFuture<Void>> futures = new ArrayList<>();
        CompletableFuture<Throwable> errorFuture = new CompletableFuture<>();
        for (Task task : tasks) {
            CompletableFuture<Void> future = CompletableFuture.runAsync(() -> {
                try {
                    executeTask(task);
                } catch (Exception e) {
                    // 将异常传递给错误Future
                    errorFuture.complete(e);
                    throw new CompletionException(e);
                }
            });
            futures.add(future);
        }
        try {
            // 等待所有任务完成或发生错误
            CompletableFuture<Void> allFutures = CompletableFuture.allOf(
                futures.toArray(new CompletableFuture[0])
            );
            // 设置超时时间
            allFutures.get(30, TimeUnit.SECONDS);
            // 检查是否有错误发生
            if (errorFuture.isDone()) {
                throw new RuntimeException("Child thread failed", errorFuture.get());
            }
        } catch (TimeoutException e) {
            throw new RuntimeException("Operation timeout", e);
        } catch (Exception e) {
            // 确保事务回滚
            TransactionAspectSupport.currentTransactionStatus().setRollbackOnly();
            throw new RuntimeException("Process failed", e);
        }
    }
    /**
     * 另一种解决方案：使用TransactionCallback
     */
    public void processWithTransactionCallback(List<Task> tasks) {
        transactionTemplate.execute(new TransactionCallbackWithoutResult() {
            @Override
            protected void doInTransactionWithoutResult(TransactionStatus status) {
                try {
                    // 并行执行任务
                    List<CompletableFuture<Void>> futures = tasks.stream()
                        .map(task -> CompletableFuture.runAsync(() -> {
                            // 每个子线程使用独立事务
                            executeInNewTransaction(task);
                        }))
                        .collect(Collectors.toList());
                    // 等待所有完成
                    CompletableFuture.allOf(futures.toArray(new CompletableFuture[0]))
                        .exceptionally(ex -> {
                            // 标记事务为回滚
                            status.setRollbackOnly();
                            return null;
                        })
                        .join();
                } catch (Exception e) {
                    status.setRollbackOnly();
                    throw e;
                }
            }
        });
    }
}

@Service
public class ConnectionLeakSolution {
    @Autowired
    private DataSource dataSource;
    /**
     * 使用Connection包装器确保资源释放
     */
    public void processWithConnectionManagement(List<Task> tasks) {
        // 使用try-with-resources确保连接关闭
        try (ConnectionHolder connectionHolder = new ConnectionHolder(dataSource)) {
            List<CompletableFuture<Void>> futures = tasks.stream()
                .map(task -> CompletableFuture.runAsync(() -> {
                    // 每个线程使用独立的连接
                    try (Connection connection = dataSource.getConnection()) {
                        connection.setAutoCommit(false);
                        try {
                            executeTaskWithConnection(task, connection);
                            connection.commit();
                        } catch (Exception e) {
                            connection.rollback();
                            throw new RuntimeException("Task failed", e);
                        }
                    } catch (SQLException e) {
                        throw new RuntimeException("Connection error", e);
                    }
                }))
                .collect(Collectors.toList());
            // 等待所有任务完成
            CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).join();
        } catch (Exception e) {
            throw new RuntimeException("Process failed", e);
        }
    }
    /**
     * 连接持有器，确保连接正确关闭
     */
    private static class ConnectionHolder implements AutoCloseable {
        private final List<Connection> connections = new ArrayList<>();
        private final DataSource dataSource;
        public ConnectionHolder(DataSource dataSource) {
            this.dataSource = dataSource;
        }
        public Connection getConnection() throws SQLException {
            Connection connection = dataSource.getConnection();
            connections.add(connection);
            return connection;
        }
        @Override
        public void close() {
            for (Connection connection : connections) {
                try {
                    if (!connection.isClosed()) {
                        connection.close();
                    }
                } catch (SQLException e) {
                    log.error("Failed to close connection", e);
                }
            }
        }
    }
    /**
     * 连接池监控
     */
    @Component
    @Slf4j
    public static class ConnectionPoolMonitor {
        @Autowired
        private DataSource dataSource;
        @Scheduled(fixedRate = 30000)
        public void monitorConnectionPool() {
            if (dataSource instanceof HikariDataSource) {
                HikariDataSource hikariDataSource = (HikariDataSource) dataSource;
                log.info("Connection pool status: " +
                         "Active: {}, " +
                         "Idle: {}, " +
                         "Total: {}, " +
                         "Waiting: {}",
                         hikariDataSource.getHikariPoolMXBean().getActiveConnections(),
                         hikariDataSource.getHikariPoolMXBean().getIdleConnections(),
                         hikariDataSource.getHikariPoolMXBean().getTotalConnections(),
                         hikariDataSource.getHikariPoolMXBean().getThreadsAwaitingConnection());
                // 连接泄漏检测
                if (hikariDataSource.getHikariPoolMXBean().getActiveConnections() > 
                    hikariDataSource.getMaximumPoolSize() * 0.8) {
                    log.warn("Connection pool usage is high, possible connection leak");
                }
            }
        }
    }
}

/**
 * 推荐的架构模式
 */
@Component
public class TransactionArchitecturePattern {
    /**
     * 模式1：主从事务模式
     * 主线程负责协调，子线程执行具体任务
     */
    public void masterSlavePattern(List<Task> tasks) {
        // 1. 主线程开启事务，记录任务状态
        recordTaskStart(tasks);
        // 2. 子线程并行处理（各自独立事务）
        List<CompletableFuture<Result>> futures = processTasksInParallel(tasks);
        // 3. 收集结果，更新状态
        processResults(futures);
        // 4. 主线程提交事务
    }
    /**
     * 模式2：补偿事务模式
     * 适用于需要最终一致性的场景
     */
    public void compensationPattern(BusinessOperation operation) {
        // 1. 执行主操作
        OperationResult result = executeMainOperation(operation);
        // 2. 记录操作日志（用于补偿）
        recordOperationLog(operation, result);
        // 3. 异步执行后续操作
        executeAsyncFollowUp(operation, result);
        // 4. 提供补偿接口
        registerCompensationCallback(operation);
    }
    /**
     * 模式3：批量处理模式
     * 适用于大批量数据处理
     */
    public void batchProcessingPattern(List<DataItem> items) {
        // 1. 分批处理
        List<List<DataItem>> batches = partitionItems(items, 100);
        // 2. 并行处理每个批次
        batches.parallelStream().forEach(batch -> {
            // 每个批次独立事务
            processBatchInTransaction(batch);
        });
        // 3. 汇总结果
        summarizeResults();
    }
}

public abstract class TransactionTemplatePattern {
    @Autowired
    protected PlatformTransactionManager transactionManager;
    /**
     * 执行带事务的异步任务
     */
    protected <T> CompletableFuture<T> executeAsyncInTransaction(
            Supplier<T> task, 
            TransactionDefinition definition) {
        return CompletableFuture.supplyAsync(() -> {
            TransactionStatus status = transactionManager.getTransaction(definition);
            try {
                T result = task.get();
                transactionManager.commit(status);
                return result;
            } catch (Exception e) {
                transactionManager.rollback(status);
                throw new CompletionException(e);
            }
        });
    }
    /**
     * 执行带重试的事务
     */
    protected <T> T executeWithRetry(
            Callable<T> task, 
            int maxRetries, 
            long backoffDelay) {
        int retryCount = 0;
        while (retryCount <= maxRetries) {
            try {
                return transactionTemplate.execute(status -> {
                    try {
                        return task.call();
                    } catch (Exception e) {
                        throw new RuntimeException(e);
                    }
                });
            } catch (Exception e) {
                retryCount++;
                if (retryCount > maxRetries) {
                    throw e;
                }
                try {
                    Thread.sleep(backoffDelay * retryCount);
                } catch (InterruptedException ie) {
                    Thread.currentThread().interrupt();
                    throw new RuntimeException(ie);
                }
            }
        }
        throw new IllegalStateException("Should not reach here");
    }
}

@Configuration
@Profile({"dev", "test"})
public class DevThreadPoolConfig {
    @Bean
    public ThreadPoolTaskExecutor taskExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(5);
        executor.setMaxPoolSize(10);
        executor.setQueueCapacity(50);
        return executor;
    }
}
@Configuration
@Profile("prod")
public class ProdThreadPoolConfig {
    @Bean
    public ThreadPoolTaskExecutor taskExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(20);
        executor.setMaxPoolSize(100);
        executor.setQueueCapacity(1000);
        executor.setAllowCoreThreadTimeOut(true);
        executor.setKeepAliveSeconds(120);
        return executor;
    }
}