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");
}

// 简化的事务传播决策逻辑
public class PropagationDecisionEngine {
    public TransactionStatus handlePropagation(
            TransactionDefinition definition, 
            PlatformTransactionManager tm) {
        boolean existingTransaction = tm.hasExistingTransaction();
        switch (definition.getPropagationBehavior()) {
            case PROPAGATION_REQUIRED:
                if (existingTransaction) {
                    // 加入现有事务
                    return handleExistingTransaction(definition, tm, existingTransaction);
                } else {
                    // 创建新事务
                    return createNewTransaction(definition, tm);
                }
            case PROPAGATION_REQUIRES_NEW:
                // 总是挂起现有事务并创建新事务
                if (existingTransaction) {
                    SuspendedResourcesHolder suspendedResources = suspend(existingTransaction);
                    try {
                        return createNewTransaction(definition, tm);
                    } catch (Throwable ex) {
                        resume(suspendedResources);
                        throw ex;
                    }
                } else {
                    return createNewTransaction(definition, tm);
                }
            // ... 其他传播行为的处理逻辑
        }
    }
}

@Service
public class OrderProcessingService {
    @Transactional(propagation = Propagation.REQUIRED)
    public void processCompleteOrder(Order order) {
        // 场景1：从无事务方法调用
        // 此时会创建新事务
        // 更新订单状态
        order.setStatus("PROCESSING");
        orderRepository.save(order);
        // 调用库存服务（同一事务）
        inventoryService.deductStock(order.getProductId(), order.getQuantity());
        // 如果库存服务使用REQUIRED，会加入此事务
        // 调用支付服务（同一事务）
        paymentService.createPayment(order);
        // 如果支付服务失败，订单和库存操作都会回滚
    }
}
// 调用方示例
public void externalCall() {
    // 无事务上下文
    orderProcessingService.processCompleteOrder(order);
    // 整个过程在一个事务中执行
}
@Transactional
public void transactionalCall() {
    // 已有事务上下文
    orderProcessingService.processCompleteOrder(order);
    // 加入调用方的事务
}

@Service
public class ProductQueryService {
    @Transactional(propagation = Propagation.SUPPORTS)
    public ProductStatistics getProductStatistics(Long productId) {
        // 场景分析：
        // 情况1：从@Transactional方法调用 -> 在事务中执行
        // 情况2：从普通方法调用 -> 无事务执行
        // 查询1：获取产品基本信息
        Product product = productRepository.findById(productId);
        // 由于可能无事务，这里存在时间差问题
        // 在两个查询之间，数据可能被其他事务修改
        // 查询2：获取产品统计信息
        ProductStats stats = statsRepository.findByProductId(productId);
        // 组合结果
        return new ProductStatistics(product, stats);
    }
    // 改进方案：使用数据库一致性视图或添加同步锁
    @Transactional(propagation = Propagation.SUPPORTS)
    @Lock(LockModeType.PESSIMISTIC_READ)
    public ProductStatistics getConsistentStatistics(Long productId) {
        // 通过锁机制保证一致性
    }
}

@Service
public class FinancialService {
    @Transactional(propagation = Propagation.MANDATORY)
    public void transferFunds(TransferRequest request) {
        // 此方法必须被事务性方法调用
        // 确保资金转移的原子性
        // 扣减转出账户
        accountService.deduct(request.getFromAccount(), request.getAmount());
        // 增加转入账户
        accountService.add(request.getToAccount(), request.getAmount());
        // 记录交易流水
        transactionService.recordTransaction(request);
        // 三个操作必须同时成功或同时失败
    }
}
// 正确用法
@Service
public class BankingService {
    @Transactional(propagation = Propagation.REQUIRED)
    public TransferResult executeTransfer(TransferRequest request) {
        // 验证业务规则
        validateTransfer(request);
        // 执行资金转移（在事务中）
        financialService.transferFunds(request);
        // 发送通知（可能使用REQUIRES_NEW）
        notificationService.sendTransferNotification(request);
        return new TransferResult("SUCCESS");
    }
}
// 错误用法
@Service
public class InvalidBankingService {
    public TransferResult invalidTransfer(TransferRequest request) {
        // 错误：直接调用MANDATORY方法
        // 将抛出：No existing transaction found for transaction marked with propagation 'mandatory'
        financialService.transferFunds(request);
        return new TransferResult("SUCCESS");
    }
}

@Service
public class AuditLogService {
    @Transactional(propagation = Propagation.REQUIRES_NEW)
    public void logCriticalOperation(CriticalOperation operation) {
        // 场景1：从无事务方法调用 -> 创建新事务
        // 场景2：从事务方法调用 -> 挂起原事务，创建新事务
        // 记录操作日志
        auditLogRepository.save(operation);
        // 写入操作轨迹
        operationTraceRepository.saveTrace(operation);
        // 新事务独立提交
        // 即使外部事务回滚，审计日志仍然保留
    }
}
// 复杂场景：嵌套的REQUIRES_NEW
@Service  
public class ComplexBusinessService {
    @Transactional(propagation = Propagation.REQUIRED)
    public void complexBusinessProcess() {
        // 主事务开始
        // 步骤1：核心业务（主事务）
        coreBusinessService.execute();
        try {
            // 步骤2：审计日志（独立事务）
            // 挂起主事务，开启新事务
            auditLogService.logCriticalOperation(operation);
            // 新事务提交，恢复主事务
        } catch (AuditException e) {
            // 审计失败不影响主业务
            log.error("审计记录失败", e);
        }
        try {
            // 步骤3：发送通知（另一个独立事务）
            // 再次挂起主事务，开启另一个新事务
            notificationService.sendAsyncNotification(notification);
        } catch (NotificationException e) {
            // 通知失败也不影响主业务
            log.error("通知发送失败", e);
        }
        // 步骤4：继续主事务逻辑
        additionalBusinessService.process();
        // 主事务提交
    }
}

@Service
public class DataExportService {
    @Transactional(propagation = Propagation.NOT_SUPPORTED)
    public ExportFile exportLargeDataset(ExportCriteria criteria) {
        // 大量数据查询，避免事务锁定
        // 查询1：基础数据（无事务，立即提交）
        List<BaseData> baseData = baseRepository.findByCriteria(criteria);
        // 复杂计算过程
        List<ProcessedData> processed = processData(baseData);
        // 查询2：关联数据（仍无事务）
        List<RelatedData> related = relatedRepository.findByProcessedData(processed);
        // 生成文件
        ExportFile file = generateExportFile(processed, related);
        // 注意：如果在此过程中发生异常
        // 已经执行的SQL不会回滚（因为无事务）
        return file;
    }
}
// 与SUPPORTS的区别
@Service
public class ComparisonService {
    // SUPPORTS vs NOT_SUPPORTED 对比演示
    public void comparePropagation() {
        // 在事务中调用
        @Transactional
        public void callInTransaction() {
            // 调用SUPPORTS方法：在事务中执行
            supportsMethod();  // 参与事务
            // 调用NOT_SUPPORTED方法：挂起事务，非事务执行
            notSupportedMethod();  // 不参与事务，每条SQL独立提交
        }
        // 不在事务中调用
        public void callWithoutTransaction() {
            // 调用SUPPORTS方法：非事务执行
            supportsMethod();  // 非事务执行
            // 调用NOT_SUPPORTED方法：非事务执行
            notSupportedMethod();  // 非事务执行
            // 两者效果相同
        }
    }
}

@Service
public class CacheManagementService {
    @Transactional(propagation = Propagation.NEVER)
    public void refreshLocalCache(String cacheKey) {
        // 此方法严禁在事务中调用
        // 原因：缓存操作应该快速完成，避免长时间占用连接
        // 清除旧缓存
        cacheManager.evict(cacheKey);
        // 查询最新数据
        Object freshData = dataService.getFreshData(cacheKey);
        // 更新缓存
        cacheManager.put(cacheKey, freshData);
        // 整个过程应该快速完成
        // 如果放在事务中，可能因为事务等待而阻塞
    }
    // 正确调用模式
    public void updateDataWithCacheRefresh(Data newData) {
        // 第一步：更新数据库（在事务中）
        dataService.updateInTransaction(newData);
        // 第二步：提交事务后刷新缓存（确保数据已持久化）
        // 这里必须在事务外调用
        refreshLocalCache("data:" + newData.getId());
    }
    // 错误调用模式
    @Transactional
    public void wrongUpdateData(Data newData) {
        dataRepository.save(newData);
        // 错误：在事务中调用NEVER方法
        // 将抛出异常！
        refreshLocalCache("data:" + newData.getId());
    }
}

@Service
public class BatchImportService {
    @Transactional(propagation = Propagation.REQUIRED)
    public ImportResult importProducts(List<Product> products) {
        ImportResult result = new ImportResult();
        for (Product product : products) {
            try {
                // 每个产品在嵌套事务中导入
                importSingleProduct(product);
                result.addSuccess(product);
            } catch (DuplicateProductException e) {
                // 重复产品：记录但不中断导入
                result.addSkipped(product, "产品已存在");
                log.warn("跳过重复产品: {}", product.getCode());
            } catch (InvalidProductException e) {
                // 无效产品：仅回滚当前产品
                result.addFailed(product, "产品数据无效");
                log.error("产品数据无效: {}", product.getCode());
                // 嵌套事务回滚到保存点
                // 主事务继续执行下一个产品
            }
        }
        return result;
    }
    @Transactional(propagation = Propagation.NESTED)
    public void importSingleProduct(Product product) {
        // 步骤1：验证产品数据
        validateProduct(product);
        // 步骤2：检查重复
        checkDuplicate(product);
        // 步骤3：保存产品
        productRepository.save(product);
        // 步骤4：保存产品图片
        saveProductImages(product);
        // 步骤5：更新产品索引
        updateProductIndex(product);
        // 所有步骤在同一个嵌套事务中
        // 任何一步失败都会回滚整个嵌套事务
        // 但不会影响其他产品的导入
    }
}
// NESTED与REQUIRES_NEW的对比
@Service
public class PropagationComparison {
    public void compareNestedVsRequiresNew() {
        // 场景：处理订单项列表
        // 使用NESTED
        @Transactional
        public void processWithNested(List<OrderItem> items) {
            for (OrderItem item : items) {
                try {
                    // 嵌套事务：共享主事务连接
                    processItemNested(item);
                } catch (Exception e) {
                    // 仅回滚当前item
                    continue;
                }
            }
            // 主事务统一提交
        }
        // 使用REQUIRES_NEW  
        @Transactional
        public void processWithRequiresNew(List<OrderItem> items) {
            for (OrderItem item : items) {
                try {
                    // 独立事务：新建连接
                    processItemRequiresNew(item);
                } catch (Exception e) {
                    // 仅回滚当前item的独立事务
                    continue;
                }
            }
            // 主事务提交
        }
    }
    // 关键区别：
    // 1. 连接使用：NESTED共享连接，REQUIRES_NEW新建连接
    // 2. 锁范围：NESTED可能锁冲突，REQUIRES_NEW隔离更好
    // 3. 性能：NESTED更高效，REQUIRES_NEW开销大
}

graph TD
    A[开始选择传播行为] --> B{方法是否需要
事务原子性保证？}
    B -->|必须保证| C[需要事务保护路径]
    B -->|不需要保证| D[不需要事务保护路径]
    C --> E{是否允许独立提交？}
    E -->|允许，需要独立| F[REQUIRES_NEW]
    E -->|不允许，需一体| G{是否需要部分回滚能力？}
    G -->|需要，部分失败不影响整体| H{数据库是否支持保存点？}
    H -->|支持| I[NESTED]
    H -->|不支持| J[考虑业务重构]
    G -->|不需要| K{是否强制要求事务上下文？}
    K -->|强制，无事务应报错| L[MANDATORY]
    K -->|不强制| M[REQUIRED]
    D --> N{是否严格禁止事务？}
    N -->|严格禁止，有事务应报错| O[NEVER]
    N -->|不严格，可适应| P{是否优化读操作？}
    P -->|是，优先无事务查询| Q[SUPPORTS]
    P -->|否，避免事务影响| R[NOT_SUPPORTED]

// 电商订单处理的最佳实践
@Service
public class ECommerceBestPractices {
    // 1. 下单主流程 - REQUIRED（默认选择）
    @Transactional(propagation = Propagation.REQUIRED)
    public Order createOrder(OrderRequest request) {
        // 创建订单（必须与后续操作原子性）
        Order order = orderRepository.save(new Order(request));
        // 扣减库存（同一事务）
        inventoryService.deductStock(request.getItems());
        // 生成支付单（同一事务）
        Payment payment = paymentService.createPayment(order);
        // 所有操作成功才提交，任何一个失败都回滚
        return order;
    }
    // 2. 支付回调处理 - REQUIRES_NEW
    @Transactional(propagation = Propagation.REQUIRES_NEW)
    public void handlePaymentSuccess(PaymentCallback callback) {
        // 支付成功必须立即记录，不受后续操作影响
        paymentRepository.updatePaymentStatus(callback.getPaymentId(), "SUCCESS");
        // 更新订单状态（可能失败，但不影响支付状态）
        try {
            orderService.updateOrderStatus(callback.getOrderId(), "PAID");
        } catch (OrderException e) {
            log.error("订单状态更新失败，支付已成功", e);
        }
    }
    // 3. 查询订单历史 - SUPPORTS
    @Transactional(propagation = Propagation.SUPPORTS, readOnly = true)
    public Page<Order> queryOrderHistory(Long userId, Pageable pageable) {
        // 查询操作，有事务则用，无事务也可
        // readOnly=true优化查询性能
        return orderRepository.findByUserId(userId, pageable);
    }
    // 4. 批量订单导出 - NOT_SUPPORTED
    @Transactional(propagation = Propagation.NOT_SUPPORTED)
    public ExportResult exportOrders(Date startDate, Date endDate) {
        // 大数据量导出，避免事务开销
        List<Order> orders = orderRepository.findByDateRange(startDate, endDate);
        return generateExport(orders); // 可能处理数十万条数据
    }
    // 5. 订单数据校验 - NESTED（如果数据库支持）
    @Transactional(propagation = Propagation.REQUIRED)
    public ValidationResult validateOrders(List<Order> orders) {
        ValidationResult result = new ValidationResult();
        for (Order order : orders) {
            try {
                // 每个订单独立校验，失败不影响其他
                validateSingleOrder(order);
                result.addValid(order);
            } catch (ValidationException e) {
                // 仅回滚当前订单的校验操作
                result.addInvalid(order, e.getMessage());
            }
        }
        return result;
    }
    @Transactional(propagation = Propagation.NESTED)
    private void validateSingleOrder(Order order) {
        // 复杂的校验逻辑
        validateBusinessRules(order);
        validateInventory(order);
        validateCustomerCredit(order);
    }
}

// 金融系统对事务的严格要求
@Service
public class FinancialTransactionService {
    // 1. 资金转账 - MANDATORY（强制事务）
    @Transactional(propagation = Propagation.MANDATORY)
    public void transfer(TransferCommand command) {
        // 必须确保在事务中执行
        // 保证原子性：要么全转，要么全不转
        // 扣减转出账户
        accountService.debit(command.getFromAccount(), command.getAmount());
        // 增加转入账户
        accountService.credit(command.getToAccount(), command.getAmount());
        // 记录交易流水
        transactionLogService.logTransfer(command);
    }
    // 2. 日终批量处理 - 复杂传播组合
    @Transactional(propagation = Propagation.REQUIRED)
    public BatchResult dailyBatchProcessing() {
        BatchResult result = new BatchResult();
        // 阶段1：数据准备（主事务）
        prepareDailyData();
        // 阶段2：利息计算（每个账户独立事务）
        List<Account> accounts = accountRepository.findAllActive();
        for (Account account : accounts) {
            try {
                calculateInterestForAccount(account);
                result.addSuccess(account);
            } catch (CalculationException e) {
                // 单个账户计算失败不影响整体
                result.addFailed(account, e.getMessage());
            }
        }
        // 阶段3：审计日志（独立事务，必须记录）
        auditDailyBatch(result);
        // 阶段4：生成报告（无事务，避免锁）
        generateDailyReport(result);
        return result;
    }
    @Transactional(propagation = Propagation.REQUIRES_NEW)
    private void calculateInterestForAccount(Account account) {
        // 每个账户独立计算和更新
        // 避免长时间锁定账户表
    }
    @Transactional(propagation = Propagation.REQUIRES_NEW)
    private void auditDailyBatch(BatchResult result) {
        // 审计记录必须保存，即使批量处理部分失败
    }
    @Transactional(propagation = Propagation.NOT_SUPPORTED)
    private void generateDailyReport(BatchResult result) {
        // 复杂报表生成，避免事务开销
    }
}

// 连接使用对比分析
public class ConnectionUsageAnalysis {
    // REQUIRES_NEW的连接开销
    @Transactional(propagation = Propagation.REQUIRED)
    public void analyzeRequiresNewCost() {
        // 主事务获取连接Connection1
        for (int i = 0; i < 1000; i++) {
            // 每次调用都需要新连接
            independentOperation(i); // 获取Connection2, Connection3...
        }
        // 总共：1001个连接（如果连接池不够会阻塞或失败）
    }
    @Transactional(propagation = Propagation.REQUIRES_NEW)
    public void independentOperation(int index) {
        // 需要新数据库连接
    }
    // NESTED的优化方案
    @Transactional(propagation = Propagation.REQUIRED)
    public void analyzeNestedOptimization() {
        // 主事务获取连接Connection1
        for (int i = 0; i < 1000; i++) {
            nestedOperation(i); // 使用同一连接，创建保存点
        }
        // 总共：1个连接，1000个保存点
    }
    @Transactional(propagation = Propagation.NESTED)
    public void nestedOperation(int index) {
        // 使用主事务连接，创建保存点
    }
}

// 传播行为对锁的影响
@Service
public class LockingAnalysis {
    // 场景：库存扣减的锁竞争
    @Transactional(propagation = Propagation.REQUIRED)
    public void highContentionScenario() {
        // 锁住商品A的记录
        inventoryService.deductProductA(10);
        // 嵌套调用可能增加锁等待
        processRelatedOperations(); // 内部可能锁其他资源
    }
    // 优化：减少锁持有时间
    @Transactional(propagation = Propagation.REQUIRED)
    public void optimizedLocking() {
        // 快速操作，尽快释放锁
        inventoryService.deductProductA(10);
        // 后续非关键操作使用独立事务
        asyncLogOperation(); // REQUIRES_NEW，不持有主事务锁
    }
    @Transactional(propagation = Propagation.REQUIRES_NEW)
    public void asyncLogOperation() {
        // 独立事务，不竞争主事务锁
    }
}

// 自调用导致传播行为失效
@Service
public class SelfInvocationProblem {
    @Transactional(propagation = Propagation.REQUIRES_NEW)
    public void methodA() {
        // 这个事务会生效
        repository.save(entityA);
        // 自调用：事务失效！
        this.methodB(); // 不会创建新事务，而是加入当前事务
        // 正确方式：注入自身代理
        self.methodB(); // 通过代理调用
    }
    @Transactional(propagation = Propagation.REQUIRES_NEW)
    public void methodB() {
        // 期望独立事务，实际可能加入methodA的事务
        repository.save(entityB);
    }
    // 解决方案1：注入自身代理
    @Autowired
    private SelfInvocationProblem self;
    // 解决方案2：使用AspectJ模式
    @Configuration
    @EnableTransactionManagement(mode = AdviceMode.ASPECTJ)
    public class AspectJConfig {
    }
}

// 异常处理对传播行为的影响
@Service
public class ExceptionPropagation {
    @Transactional(propagation = Propagation.REQUIRED)
    public void parentMethod() {
        try {
            // 调用REQUIRES_NEW方法
            childMethodWithRequiresNew();
        } catch (ChildException e) {
            // 子方法异常被捕获
            // 子事务已回滚，但父事务继续
            log.error("子方法失败", e);
        }
        // 父事务继续执行
        otherOperation(); // 仍然可以执行
    }
    @Transactional(propagation = Propagation.REQUIRES_NEW)
    public void childMethodWithRequiresNew() {
        repository.save(childEntity);
        throw new ChildException("子事务失败");
        // 新事务回滚，异常传播到父方法
    }
    // 嵌套事务的异常处理
    @Transactional(propagation = Propagation.REQUIRED)
    public void parentWithNested() {
        try {
            childMethodWithNested();
        } catch (ChildException e) {
            // 嵌套事务回滚到保存点
            // 父事务继续
            handleError(e);
        }
    }
    @Transactional(propagation = Propagation.NESTED)
    public void childMethodWithNested() {
        repository.save(childEntity);
        throw new ChildException("嵌套事务失败");
        // 回滚到保存点，异常传播
    }
}

# Spring Boot监控配置
management:
  endpoints:
    web:
      exposure:
        include: metrics,prometheus,transactions
  metrics:
    export:
      prometheus:
        enabled: true
    distribution:
      percentiles-histogram:
        http.server.requests: true
      sla:
        transaction.duration: 100ms,200ms,500ms,1s,2s
logging:
  level:
    org.springframework.transaction: DEBUG
    org.springframework.jdbc.datasource.DataSourceTransactionManager: TRACE
    org.springframework.orm.jpa.JpaTransactionManager: TRACE

// 传播行为的单元测试
@SpringBootTest
@Transactional // 测试类默认事务
class TransactionPropagationTest {
    @Autowired
    private TestService testService;
    @Autowired
    private PlatformTransactionManager transactionManager;
    @Test
    void testRequiredPropagation() {
        // 测试1：无事务时是否创建新事务
        TransactionStatus statusBefore = TransactionAspectSupport.currentTransactionStatus();
        assertNull(statusBefore); // 当前无事务
        testService.methodWithRequired();
        // 验证事务创建
    }
    @Test
    @Transactional // 测试已有事务
    void testRequiredWithExistingTransaction() {
        TransactionStatus statusBefore = TransactionAspectSupport.currentTransactionStatus();
        assertNotNull(statusBefore); // 当前已有事务
        // 调用方法应加入现有事务
        testService.methodWithRequired();
        // 验证事务ID相同
    }
    @Test
    void testRequiresNewCreatesNewTransaction() {
        // 模拟监控
        MeterRegistry meterRegistry = new SimpleMeterRegistry();
        // 执行REQUIRES_NEW方法
        testService.methodWithRequiresNew();
        // 验证连接使用次数
        Timer timer = meterRegistry.timer("transaction.requires_new.duration");
        assertTrue(timer.count() > 0);
    }
    @Test
    void testMandatoryThrowsWithoutTransaction() {
        // 验证无事务时抛出异常
        assertThrows(IllegalTransactionStateException.class, () -> {
            testService.methodWithMandatory();
        });
    }
}

// 分布式事务的SAGA模式实现
@Service
public class DistributedOrderSaga {
    @Transactional(propagation = Propagation.REQUIRED)
    public void createDistributedOrder(OrderRequest request) {
        // 本地事务开始
        Order order = createLocalOrder(request);
        try {
            // 步骤1：调用库存服务（远程）
            inventoryFeignClient.deduct(request.getItems());
            // 步骤2：调用支付服务（远程）
            paymentFeignClient.create(order);
            // 本地提交
            orderRepository.save(order);
        } catch (RemoteException e) {
            // 分布式事务失败，执行补偿
            compensate(order, request);
            throw new SagaException("分布式事务失败", e);
        }
    }
    // 补偿操作也需要事务
    @Transactional(propagation = Propagation.REQUIRES_NEW)
    private void compensate(Order order, OrderRequest request) {
        // 回滚本地订单
        orderRepository.delete(order);
        // 调用远程补偿（可能需要异步）
        compensationService.compensateInventory(request.getItems());
        compensationService.compensatePayment(order.getId());
    }
}