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