@TransactionalEventListener投递RabbitMQ消息时JPA事务下消息丢失问题

场景与基础实现 #

• 搭建匹配生产特征的模拟应用，实现事务内数据库数据保存、同流程发布事件并向RabbitMQ投递消息的完整业务逻辑
• 核心类与方法定义：
  • 事务入口方法createEmployeeInTrans，标注@Transactional注解，内部调用业务核心方法createEmployee
  • createEmployee方法逻辑：完成EmpEntity实体的数据库持久化操作后，发布携带EmpDTO载荷的ActivityEvent事件
  • 事件监听器onActivitySave：标注@TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT)，在事件触发后通过自定义kRabbitTemplate向指定交换机、路由键投递消息
• 自定义kRabbitTemplate配置：
  • 开启RabbitMQ channel层级事务
  • 绑定JSON格式消息转换器

问题现象 #

• 业务流程执行完成后，数据库数据可正常持久化落库
• RabbitMQ服务端未收到投递的消息，出现消息丢失问题

对AMQP事务机制的已知认知 #

• 若RabbitTemplate开启channel事务，调用convertAndSend方法时：
  • 无Spring/JPA事务上下文时，模板会在send方法内部自动完成事务提交
  • 存在Spring/JPA事务上下文时，isChannelLocallyTransacted判定为false，消息提交时机将跟随外层Spring事务的完成节奏

问题排查过程 #

1. 调用convertAndSend时检测到Spring事务处于激活状态，按预期逻辑消息应当跟随事务完成提交
2. RabbitTemplate在ConnectionFactoryUtils.bindResourceToTransaction绑定资源时，因TransactionSynchronizationManager.isSynchronizationActive()返回false，无法注册Rabbit事务同步回调，导致外层事务完成时无法触发RabbitMQ消息提交动作
3. 进一步定位发现：事务管理器在AbstractPlatformTransactionManager.triggerAfterCompletion逻辑中，因status.isNewSynchronization()判定为true，在数据库操作完成后提前清空了事务同步状态；该异常链路在不使用ApplicationEvent、直接在事务方法内调用convertAndSend时不会触发
4. 异常衍生影响：已绑定的RabbitMQ连接资源未被正常清理，会阻塞当前线程后续的消息投递操作

临时修复方案 #

• 在onActivitySave方法上添加@Transactional(propagation = Propagation.REQUIRES_NEW)注解，为监听器逻辑开启独立新事务后，消息可正常投递
• 当前为生产紧急问题，该临时修复方案的长期可靠性尚未验证

核心待确认问题 #

• 使用ApplicationEvent事件流程时，为何会触发triggerAfterCompletion中status.isNewSynchronization()=true的判定分支？
• 父方法事务按逻辑应当结束的场景下，为何在事务后触发的监听器中检测到TransactionSynchronizationManager.isActualTransactionActive()返回true？
• 当检测到事务处于激活状态时，是否应当执行事务同步状态的清空操作？
• Spring AMQP的bindResourceToTransaction逻辑是否默认假设：事务激活时必然存在可用的事务同步状态？若该假设成立，为何代码在检测到同步状态未激活时不主动完成初始化？
• 当前基于REQUIRES_NEW传播行为的修复方案会脱离父事务上下文，是否存在更合理的实现方案？

内容的提问来源于stack exchange，提问作者Kuldeep Kala