在事务还没有执行完消息就已经发出去了, 导致后续的一些数据或逻辑上的问题产生。

场景如下&＃xff1a;

• 异步-记录日志&＃xff1a;当事务提交后&＃xff0c;再记录日志。

• 发送mq消息&＃xff1a;只有业务数据都存入表后&＃xff0c;再发mq消息。

方案1. 利用TransactionSynchronizationManager的registerSynchronization()方法注册TransactionSynchronization实现类

我们只需要在执行的事务方法中, 添加如下代码, 就可以完成在事务提交后的逻辑处理了

// TransactionSynchronizationAdapter是TransactionSynchronization的默认实现
TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronizationAdapter() {
&＃64;Override
public void afterCommit() {
// 事务提交后需要执行的业务逻辑: 发消息, 日志...
}
});


原理&＃xff1a;

Spring 事务的扩展 – TransactionSynchronization

事务操作的时候它的当前线程还保存了 TransactionSynchronization 对象。而这个对象伴随着 Spring 对 事务处理的各个生命周期都会有相应的扩展

public interface TransactionSynchronization extends Flushable {
/** 事务提交状态 */
int STATUS_COMMITTED &＃61; 0;
/** 事务回滚状态 */
int STATUS_ROLLED_BACK &＃61; 1;
/**系统异常状态 */
int STATUS_UNKNOWN &＃61; 2;
void suspend();
void resume();
void flush();
// 事务提交之前
void beforeCommit(boolean readOnly);
// 事务成功或者事务回滚之前
void beforeCompletion();
// 事务成功提交之后
void afterCommit();
// 操作完成之后(包含事务成功或者事务回滚)
void afterCompletion(int status);
}

事务的事务扩展项目中的应用场景是当订单成功之后&＃xff0c;发送一条消息到 MQ 当中去。由于事务是和数据库连接相绑定的&＃xff0c;如果把发送消息和数据库操作放在一个事务里面。当发送消息时间过长时会占用数据库连接&＃xff0c;所以就要把数据库操作与发送消息到 MQ 解耦开来。可以利用 TransactionSynchronization#afterCommit 的这个方法&＃xff0c;当数据成功保存到数据库并且事务提交了就把消息发送到 MQ 里面。

&＃64;Transactional
public void finishOrder(Order order){
// 修改订单成功
updateOrderSuccess(order);
// 发送消息到 MQ
TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronizationAdapter(){
&＃64;Override
public void afterCommit() {
mqService.send(order);
}
});
}

当事务成功提交之后&＃xff0c;就会把消息发送给 MQ&＃xff0c;并且不会占用数据库连接资源。

方式二、Spring 事务扩展 – &＃64;TransactionalEventListener

在 Spring framework 4.2 之后还可以使用&＃64;TransactionalEventListener处理数据库事务提交成功后再执行操作。这种方式比 TransactionSynchronization 更加优雅。它的使用方式如下&＃xff1a;

&＃64;Autowired
private ApplicationEventPublisher publisher;
&＃64;Override
&＃64;Transactional(rollbackFor &＃61; Exception.class)
public void add(AdvanceChargeApplyAddInput input) {
this.save(advanceChargeApply);
// 发送事件
publisher.publishEvent(advanceChargeApply);
}
// 响应事件, 事务提交后执行
&＃64;TransactionalEventListener
public void handle(PayloadApplicationEvent event) {
System.out.println("TransactionalEventListener 事务提交后执行");
}

也可以自定义事件源和事件监听&＃xff1a;

&＃64;Transactional
public void finishOrder(Order order){
// 修改订单成功
updateOrderSuccess(order);

// 发布 Spring Event 事件
applicationEventPublisher.publishEvent(new MyAfterTransactionEvent(order));
}

&＃64;Slf4j
&＃64;Component
private static class MyTransactionListener {
&＃64;Autowired
private MqService mqService;
&＃64;TransactionalEventListener(phase &＃61; TransactionPhase.AFTER_COMMIT)
private void onHelloEvent(MyAfterTransactionEvent event) {
Order order &＃61; event.getOrder();
mqService.send(order);
}
}
// 定一个事件&＃xff0c;继承自ApplicationEvent
private static class MyAfterTransactionEvent extends ApplicationEvent {
private Order order;
public MyAfterTransactionEvent(Object source, Order order) {
super(source);
this.order &＃61; order;
}
public Order getOrder() {
return order;
}
}

它的实现原理是当 Spring Bean 的方法标注了通过 TransactionalEventListenerFactory#createApplicationListener创建 ApplicationListenerMethodTransactionalAdapter 然后在事件回调当中创建 TransactionSynchronization的实现类TransactionSynchronizationEventAdapter。并且通过 TransactionSynchronizationManager.registerSynchronization

把 TransactionSynchronizationEventAdapter 注册到当前线程当中。

TransactionSynchronizationEventAdapter

private static class TransactionSynchronizationEventAdapter extends TransactionSynchronizationAdapter {
private final ApplicationListenerMethodAdapter listener;
private final ApplicationEvent event;
private final TransactionPhase phase;
public TransactionSynchronizationEventAdapter(ApplicationListenerMethodAdapter listener,
ApplicationEvent event, TransactionPhase phase) {
this.listener &＃61; listener;
this.event &＃61; event;
this.phase &＃61; phase;
}
&＃64;Override
public int getOrder() {
return this.listener.getOrder();
}
&＃64;Override
public void beforeCommit(boolean readOnly) {
if (this.phase &＃61;&＃61; TransactionPhase.BEFORE_COMMIT) {
processEvent();
}
}
&＃64;Override
public void afterCompletion(int status) {
if (this.phase &＃61;&＃61; TransactionPhase.AFTER_COMMIT && status &＃61;&＃61; STATUS_COMMITTED) {
processEvent();
}
else if (this.phase &＃61;&＃61; TransactionPhase.AFTER_ROLLBACK && status &＃61;&＃61; STATUS_ROLLED_BACK) {
processEvent();
}
else if (this.phase &＃61;&＃61; TransactionPhase.AFTER_COMPLETION) {
processEvent();
}
}
protected void processEvent() {
this.listener.processEvent(this.event);
}
}