基于 Spring 事务同步机制的事务后置动作收集器 Starter 实践 基于 Spring 事务同步机制的事务后置动作收集器 Starter 实践一、涉及的技术知识点1.1 Spring 事务同步机制知识点说明TransactionSynchronizationSpring 提供的事务同步回调接口在事务提交/回滚后触发TransactionSynchronizationAdapter适配器类选择性覆写需要的回调方法TransactionSynchronizationManager管理当前线程的事务同步器注册afterCommit()事务提交成功后调用afterCompletion(status)事务完成后调用无论提交还是回滚status 标识结果事务状态常量STATUS_COMMITTED0、STATUS_ROLLED_BACK1、STATUS_UNKNOWN21.2 设计模式模式应用观察者模式收集器作为事务的观察者监听提交/回滚事件命令模式AfterTransactionSyncAction接口封装延迟执行的操作收集器模式在事务内累积多个动作事务结束后批量执行策略模式提交动作和回滚动作分开注册、分开执行1.3 解决的核心问题在Transactional方法内有些操作必须在事务提交后才能执行发 MQ 消息事务未提交就发消息消费者可能查到旧数据释放分布式锁事务未提交就解锁其他线程读到未提交的中间状态调外部接口事务回滚了但外部接口已调用无法撤销缓存更新事务未提交就更新缓存缓存与数据库不一致1.4 Java 核心知识点说明函数式接口AfterTransactionSyncAction只有一个方法支持 LambdaLinkedList保持注册顺序执行方法引用lock::unlock作为AfterTransactionSyncAction传入注博客https://blog.csdn.net/badao_liumang_qizhi二、包结构xxx.xxx.lib.transaction ├── AfterTransactionSyncAction.java // 提交后动作接口 ├── AfterTransactionRollbackSyncAction.java // 回滚后动作接口 └── AfterTransactionActionCollector.java // 核心收集器注册到事务同步器注意无spring.factories不是自动配置库。使用时在业务代码中直接new并注册。三、核心实现流程Transactional 方法内 │ ├─→ new AfterTransactionActionCollector() │ ├─→ collector.addCommitSyncAction(() - sendMqMessage()) // 注册提交后动作 ├─→ collector.addCommitSyncAction(lock::unlock) // 注册释放锁动作 ├─→ collector.addRollbackSyncAction(() - compensate()) // 注册回滚补偿动作 │ ├─→ TransactionSynchronizationManager │ .registerSynchronization(collector) // 注册到Spring事务管理器 │ ├─→ 业务逻辑执行数据库操作等 │ └─→ 事务结束 │ ├── 提交成功 → afterCommit() │ └── 按顺序执行所有 commitActionssendMqMessage() → lock.unlock() │ ├── 回滚 → afterCompletion(STATUS_ROLLED_BACK1) │ └── 按顺序执行所有 rollbackActionscompensate() │ └── afterCompletion() 最终清理 └── commitActions.clear() rollbackActions.clear()四、通用示例代码4.1 AfterTransactionSyncAction提交后动作接口packagecom.example.transaction;/** * 事务提交后执行的动作接口. * 函数式接口支持Lambda表达式. */FunctionalInterfacepublicinterfaceAfterTransactionSyncAction{voidexecute();}4.2 AfterTransactionRollbackSyncAction回滚后动作接口packagecom.example.transaction;/** * 事务回滚后执行的动作接口. * 用于补偿操作如撤销已发的外部通知等. */FunctionalInterfacepublicinterfaceAfterTransactionRollbackSyncAction{voidexecute();}4.3 AfterTransactionActionCollector核心收集器packagecom.example.transaction;importjava.util.LinkedList;importjava.util.List;importorg.slf4j.Logger;importorg.slf4j.LoggerFactory;importorg.springframework.transaction.support.TransactionSynchronizationAdapter;/** * 事务后置动作收集器. * * 在 Transactional 方法内使用收集需要在事务提交/回滚后执行的动作. * * 核心价值 * 1. 保证 MQ 消息在事务提交后才发送消费者能查到已提交数据 * 2. 保证分布式锁在事务提交后才释放其他线程读到一致数据 * 3. 支持事务回滚时执行补偿逻辑 * * 使用方式 * 1. new 一个 collector * 2. addCommitSyncAction() 注册提交后动作 * 3. addRollbackSyncAction() 注册回滚后动作 * 4. TransactionSynchronizationManager.registerSynchronization(collector) 注册 */publicclassAfterTransactionActionCollectorextendsTransactionSynchronizationAdapter{privatestaticfinalintSTATUS_COMMITTED0;privatestaticfinalintSTATUS_ROLLED_BACK1;privatestaticfinalintSTATUS_UNKNOWN2;privatefinalListAfterTransactionSyncActioncommitActionsnewLinkedList();privatefinalListAfterTransactionRollbackSyncActionrollbackActionsnewLinkedList();privatefinalLoggerloggerLoggerFactory.getLogger(getClass());/** * 添加事务提交后执行的动作等同于 addCommitSyncAction. */publicvoidaddSyncAction(AfterTransactionSyncActionaction){logger.debug(add action to thread local: {},action);commitActions.add(action);}/** * 添加事务提交后执行的动作. */publicvoidaddCommitSyncAction(AfterTransactionSyncActionaction){logger.debug(add commit sync action: {},action);commitActions.add(action);}/** * 添加事务回滚后执行的动作. */publicvoidaddRollbackSyncAction(AfterTransactionRollbackSyncActionaction){logger.debug(add rollback sync action: {},action);rollbackActions.add(action);}/** * 事务提交成功后回调. * 按注册顺序依次执行所有 commitActions. */OverridepublicvoidafterCommit(){if(!commitActions.isEmpty()){for(AfterTransactionSyncActionaction:commitActions){logger.debug(begin to execute after commit action: {},action);action.execute();}}else{logger.debug(no commitActions to be executed);}}/** * 事务完成后回调无论提交还是回滚. * 如果是回滚status1执行所有 rollbackActions. * 最后清理所有已注册的动作. */OverridepublicvoidafterCompletion(intstatus){logger.debug(after transaction with status {}, commit action size: {}, rollback action size: {},status,commitActions.size(),rollbackActions.size());if(statusSTATUS_ROLLED_BACK!rollbackActions.isEmpty()){logger.debug(the transaction rolled back! begin to execute roll back actions);for(AfterTransactionRollbackSyncActionaction:rollbackActions){logger.debug(begin to execute after rollback action: {},action);action.execute();}}// 清理防止内存泄漏commitActions.clear();rollbackActions.clear();}}4.4 pom.xmlStarter 侧projectgroupIdcom.example/groupIdartifactIdexample-transaction-action-starter/artifactIdversion1.0.0/versionpackagingjar/packagingdependenciesdependencygroupIdorg.springframework/groupIdartifactIdspring-tx/artifactIdscopeprovided/scope/dependencydependencygroupIdorg.slf4j/groupIdartifactIdslf4j-api/artifactIdscopeprovided/scope/dependency/dependencies/project五、引入方使用5.1 添加依赖dependencygroupIdcom.example/groupIdartifactIdexample-transaction-action-starter/artifactIdversion1.0.0/version/dependency5.2 使用示例1事务提交后发 MQServicepublicclassOrderService{ResourceprivateOrderMqSenderorderMqSender;Transactional(rollbackForException.class)publicvoidcreateOrder(OrderDtoorderDto){// 1. 创建收集器并注册到事务管理器AfterTransactionActionCollectorcollectornewAfterTransactionActionCollector();TransactionSynchronizationManager.registerSynchronization(collector);// 2. 执行数据库操作OrderorderorderRepository.save(convertToEntity(orderDto));// 3. 注册事务提交后发MQLambda 写法collector.addCommitSyncAction(()-{orderMqSender.sendOrderCreatedMessage(order.getId());});// 4. 其他业务逻辑...// 事务提交后MQ消息才会发出如果事务回滚消息不会发}}5.3 使用示例2事务提交后释放分布式锁ServicepublicclassStockService{ResourceprivateDistributedLockProviderdistributedLockProvider;Transactional(rollbackForException.class)publicvoiddeductStock(IntegeritemId,Integerqty){// 1. 获取锁StringlockKeystock_deduct_itemId;DistributedLocklockdistributedLockProvider.getLock(lockKey);lock.tryLock(TimeUnit.MINUTES,1);// 2. 注册事务提交后释放锁AfterTransactionActionCollectorcollectornewAfterTransactionActionCollector();TransactionSynchronizationManager.registerSynchronization(collector);collector.addCommitSyncAction(lock::unlock);// 方法引用// 3. 执行库存扣减stockRepository.deduct(itemId,qty);// 事务提交后锁才释放 → 其他线程读到的一定是已提交的数据// 事务回滚 → afterCompletion 中 clear 动作列表但锁不会主动释放需要等待过期}}5.4 使用示例3提交和回滚分别处理ServicepublicclassPaymentService{Transactional(rollbackForException.class)publicvoidprocessPayment(PaymentDtodto){AfterTransactionActionCollectorcollectornewAfterTransactionActionCollector();TransactionSynchronizationManager.registerSynchronization(collector);// 调用第三方支付已扣款StringpaymentIdthirdPartyPayService.charge(dto.getAmount());// 保存支付记录paymentRepository.save(newPayment(paymentId,dto));// 事务提交后通知下游发货collector.addCommitSyncAction(()-{deliveryService.triggerDelivery(dto.getOrderId());});// 事务回滚后调第三方退款补偿collector.addRollbackSyncAction(()-{thirdPartyPayService.refund(paymentId);});}}5.5 使用示例4结合分布式锁完整场景ServicepublicclassBatchDeliveryService{Transactional(rollbackForException.class)publicvoidbatchConfirmDelivery(DeliveryParamsDtoparams){// 获取锁Stringkeybatch_delivery_params.getMemberId();DistributedLocklockdistributedLockProvider.getLock(key,TimeUnit.MINUTES,10);lock.tryLock(TimeUnit.MINUTES,8);// 注册事务后释放锁AfterTransactionActionCollectorcollectornewAfterTransactionActionCollector();TransactionSynchronizationManager.registerSynchronization(collector);collector.addCommitSyncAction(lock::unlock);// 执行批量发货业务// ... 写入发货单、写入扩展表is_fixed_delivery_date、写出库单 ...// 事务提交后发MQ通知collector.addCommitSyncAction(()-{deliveryMqSender.sendToYc(params.getDeliveryRecordCode());});}}六、关键设计总结设计要点实现方式收益事务与副作用解耦副作用延迟到事务提交后执行MQ 消费者读到已提交数据外部调用不会因回滚而产生脏数据提交/回滚分离commitActions rollbackActions 两个列表不同事务结果执行不同逻辑支持补偿有序执行LinkedList 保持注册顺序多个动作按业务预期顺序执行自动清理afterCompletion 中 clear 两个列表防止内存泄漏零配置无 spring.factories直接 new 使用极简无侵入函数式接口支持 Lambda 和方法引用代码简洁优雅最小依赖仅依赖 spring-tx slf4j任何 Spring 事务项目都能用