深入 Spring 36：响应式事务、取消窗口与重新订阅

没有收到完成信号，能否认定订单被回滚

客户端取消请求后，应用没有收到 onComplete，日志也没有打印“下单成功”。这只能描述订阅者看到的信号，不能直接描述数据库状态。取消可能发生在 SQL 之前、SQL 执行之后、事务提交之前，也可能发生在已经提交但响应尚未到达时。不同窗口的业务结果不同。

响应式事务还增加了另一个容易遗漏的边界：Publisher 对象不等于一次事务执行。同一个冷 Publisher 被订阅两次，可以启动两次事务并执行两次 INSERT。第二次执行各自拥有正常提交结果，仍可能违反订单只能创建一次的业务要求。

实验固定 Spring Framework 6.2.11、Reactor 3.7.11、R2DBC PostgreSQL 1.0.7.RELEASE、R2DBC Pool 1.0.2.RELEASE、PostgreSQL 18.0 和 JDK 21。Chapter36 使用真实 R2DBC 驱动写库，以独立 pgJDBC 连接观察数据；池容量设为一，以验证资源归还。没有用内存数据库或假事务管理器代替提交、回滚和连接释放。

订阅驱动的事务及取消窗口

事务资源依附于订阅上下文

普通 JDBC 事务常通过线程关联当前连接。响应式序列可能在不同线程上继续，因而不能把固定线程当成事务身份。TransactionalOperator 为订阅建立事务上下文，R2dbcTransactionManager 获取连接并关联资源，DatabaseClient 经 Spring 的连接工具参与该事务。R2dbcTransactionManager 固定源码

这里的 Context 不是一个自动传播所有环境变量的容器。它是响应式订阅链中的上下文机制。只有加入相同受管理链的数据库工作，才可以按相同资源键查找事务连接。如果业务另行直接获取连接、单独 subscribe()，或者调用不支持该机制的阻塞 API，就需要重新分析资源归属。

实验在事务中调用响应式 TransactionSynchronizationManager.forCurrentTransaction()，验证事务处于活动状态。随后经过 publishOn(boundedElastic())，再次查找管理器，事务仍然活动，线程名称却已经改变。断言同时检查状态和实际线程切换，避免“根本没切线程”产生假阳性。

这能证明当前链上的调度切换没有丢失事务上下文，不能证明任意线程池回调都天然继承它。直接提交给不相关执行器的 Runnable 没有因此加入 R2DBC 事务；把 JDBC 连接对象塞入 Context 也不会使 JDBC 驱动变成非阻塞驱动。

成功与错误如何进入清理分支

固定版本的 TransactionalOperatorImpl.execute 使用 Flux.usingWhen 管理事务资源。资源取得后调用业务回调，正常完成路径执行 commit，错误路径执行 rollbackOnException，取消路径执行 rollback。上下文通过 contextWrite 接入。这一控制结构解释了为什么事务开始、业务信号和资源清理需要放在同一条订阅链中。TransactionalOperatorImpl 固定源码

正常组构造 Mono.defer(...insert...) 并应用 tx.transactional。尚未订阅时，业务执行计数为零，独立连接也查不到数据。第一次 block 是实验主线程的订阅入口；它等待受管理链完成，之后独立连接看到一条提交记录。这里允许主测试线程阻塞，以便建立确定的观察顺序；不能把同一用法照搬到 Netty 事件循环。

错误组先执行 INSERT，随后发出业务异常 IllegalStateException("business-error")。入口观察到该异常，独立查询得到零条记录，事务监听器记录成功完成的 rollback。异常断言校验消息，避免把连接失败或 SQL 拼写错误误当成预期业务失败。

一个重要的作用域问题是错误恢复操作符放在哪里。若 onErrorResume 位于事务范围内部，并将错误转成正常结果，事务边界可能只看到正常完成，从而提交。若恢复位于事务范围之外，事务管理器先看到错误并执行回滚，再由外层生成替代结果。不能仅凭链上出现 onErrorResume 推断最终事务结果。编程式事务与取消说明

取消实验必须固定发生窗口

取消组先查询 pg_backend_pid()，记录当前数据库会话，再执行 INSERT。INSERT 的 rowsUpdated 完成后才进入 Mono.never()，此时通知测试线程的屏障。屏障表示驱动已经报告 INSERT 完成，但业务流仍未正常完成。

主测试线程在该位置做两个独立检查：JDBC 会话看不到该行；R2DBC 池显示唯一连接仍被占用。随后调用订阅的 dispose()，等待内部 doOnCancel 记录取消已向上游传播。事务边界因取消进入回滚和清理路径。

订阅 -> 取连接 -> BEGIN -> INSERT 完成 -> Mono.never
                                         |
                              独立连接看不到行，池 acquired=1
                                         |
                                      dispose
                                         |
                               ROLLBACK -> 连接归还

doOnCancel 被触发不等于异步 rollback 已经完成。测试没有在取消信号出现的瞬间就断言连接归还，而是通过同一个容量为一的池执行后续查询。后续查询成功说明前一个租借已释放；查询结束后还检查 acquired=0 和 pending=0，独立 JDBC 会话再确认被取消的记录仍为零。

这个窗口能证明“INSERT 已执行、事务尚未正常完成时取消”的回滚行为。它没有模拟提交包已经被 PostgreSQL 接收、提交确认却丢失的网络窗口。对后一类故障，仅靠客户端异常无法判定交易是否发生，仍然需要业务查询、唯一键或补偿流程。客户端断连也不能无条件等同于实验中的确定取消位置。

重新订阅重新执行，幂等性仍属于业务

成功组保留最初的冷 Publisher，再订阅一次。执行计数变成二，普通订单表中同一个业务键出现两条记录。每次事务都成功提交，但“只创建一个订单”的业务约束已经失败。

第二张实验表以业务键作为主键，写入使用 PostgreSQL ON CONFLICT DO NOTHING。对同一 Publisher 连续订阅两次，第一次受影响行数为一，第二次为零。数据库约束阻止重复行，应用还必须决定如何返回第一次的业务结果。零行更新本身不是完整的幂等接口协议，尤其不能解决同一幂等键对应不同请求内容的问题。

retry 在错误后重新订阅，因而需要沿着这个模型分析。一次错误可能出现在业务副作用之前，也可能发生在结果已持久化之后。给数据库调用增加 retry 并不能省略幂等设计；它只是改变执行次数和故障恢复路径。

缓存 Publisher 也不是任意事务的替代方案。缓存可能改变订阅次数、错误重放和结果存活时间，但不能代替跨进程、跨重启的唯一约束。事务负责一个执行范围内的提交与回滚，幂等性约束负责多个执行范围之间的重复效果，两者的对象不同。

连接释放也是验收结果

成功、错误、取消、重新订阅和幂等五组之后，都通过容量为一的池执行 select 1，再检查连接与等待者计数。这样既有可继续服务的行为证据，也有池状态观察，避免只以没有异常结束作为资源安全的证明。

事务监听器在此程序中观察到五次提交、两次回滚。五次提交分别来自两次普通成功订阅、一次跨线程上下文探针、两次幂等写入；两次回滚来自业务错误和受控取消。监听器事件只用于关联执行路径，独立数据库查询才用于判定业务行是否存在。

程序 finally 中等待池异步关闭，并关闭使用过的共享调度器。最后检查池已 disposed。测试只清空它拥有的 spring_ch36_orders 和 spring_ch36_idempotent 两张实验表；不停止 PostgreSQL 服务，也不修改其他章节的表。

复现实验

先按工程 db/README.md 启动 PostgreSQL。默认地址是 127.0.0.1:55432/spring_lab，用户 spring_lab，本地实验无密码。设置 JAVA_HOME 为 JDK 21 后，在工程根目录运行：

1
2
3

./mvnw -f reactive-lab/pom.xml -q compile dependency:build-classpath \
  -Dmdep.outputFile=target/classpath.txt
python3 reactive-lab/verify.py 36

Java 入口同时接受 spring.lab.r2dbc.url 与 spring.lab.jdbc.url 系统属性；更换数据库地址时，两者必须指向同一数据库。只更改一边会让独立观察失去对照意义。

本次真实运行通过 23 条断言，日志在 evidence/36/local-20261002/run.txt，退出码为 0。manifest.json 保存命令、JDK 与源码哈希。实验覆盖固定取消窗口、成功、业务错误、上下文切换和重复订阅；提交确认丢失、服务端崩溃恢复及 R2DBC 驱动自身上游测试均未在本实验运行。

预测题与改动练习

在错误组中将 onErrorResume 分别放到 as(tx::transactional) 前后，预测订单行数，然后由独立连接验证。检查事务监听器的 commit/rollback，解释外层最终都返回成功值时，数据库结果为什么仍可能不同。

把取消屏障移到 INSERT 之前，保留相同池检查。预期数据库仍没有行，但证据只能覆盖“业务写入尚未开始”的窗口。将这两次运行的日志并列，避免把结果相同误认为取消发生阶段相同。

给普通表增加业务唯一键，再把 ON CONFLICT DO NOTHING 改成遇到冲突返回已有订单。对相同键、不同订单金额发起两次订阅，明确接口应返回冲突还是沿用首个结果。这个选择需要业务契约，不能由事务传播级别决定。

参考资料

响应式编程式事务：文档路径滚动更新，源码与实验固定 6.2.11。
TransactionalOperatorImpl：正常、错误和取消清理分支。
R2dbcTransactionManagerTests：上游测试用于源码对照，本文未运行 Spring 整体测试集。
R2DBC PostgreSQL 1.0.7：驱动支持与连接选项。

下载入口：深入 Spring（00）：从手动组装到可验证的容器实验。