场景
在业务中,经常需要在执行数据库操作后(事务提交完成),发送消息或事件来异步调用其他组件执行相应的业务操作。
比如: 用户注册成功后,发送激活码或激活邮件,如果用户保存后就执行异步操作发送激活码或激活邮件,但是前面用户保存后发生异常,数据库进行回滚,用户实际没有注册成功,但用户收到激活码或激活邮件。此时,我们就迫切要求数据库事务完成后再执行异步操作。
需求
在成功修改数据库后,向Rabbitmq发送一条消息
解决方案
1.使用TransactionSynchronizationManage 控制事务
![在这里插入图片描述](https://img-blog.csdnimg.cn/6a4ed7473e6449bd92b071c35ab926c2.png)
TransactionSynchronization 该类中有方法可以控制业务在提交事务后执行。
![在这里插入图片描述](https://img-blog.csdnimg.cn/2f62dde535c24910a3e498e61df25d9f.png)
service的代码
public boolean updateById(Hotel hotel) {
logger.info("----- into service -----");
//修改DB的数据
hotelMapper.updateById(hotel);
//事务提交后再执行
TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronization() {
@Override
public void afterCommit() {
//成功修改DB的数据后,向Rabbitmq发消息
rabbitTemplate.convertAndSend(MqConstants.HOTEL_EXCHANGE, MqConstants.HOTEL_INSERT_KEY, hotel.getId().toString());
logger.info("----- rabbitmq send message -----");
}
});
logger.info("--------- out service ----------");
//模拟异常
int n =0/0;
return false;
}
注意: 上面的代码将在事务提交后执行.如果在非事务context中将抛出java.lang.IllegalStateException: Transaction synchronization is not active。
![在这里插入图片描述](https://img-blog.csdnimg.cn/f8c35b70848e45e2a94705a6448a4283.png)
所以必须在service方法上添加 @Transactional注解。
2. 使用 TransactionSynchronizationAdapter 控制事务
配置类
import com.sun.istack.internal.NotNull;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;
import org.springframework.transaction.support.TransactionSynchronizationAdapter;
import org.springframework.transaction.support.TransactionSynchronizationManager;
import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.*;
@Component("afterCommitExecutor")
public class AfterCommitExecutor extends TransactionSynchronizationAdapter implements Executor {
private static final ThreadLocal<List<Runnable>> RUNNABLES = new ThreadLocal<List<Runnable>>();
private ThreadPoolExecutor threadPool;
private Logger logger = LoggerFactory.getLogger(AfterCommitExecutor.class);
@PostConstruct
public void init() {
logger.debug("初始化线程池。。。");
int availableProcessors = Runtime.getRuntime().availableProcessors();
if (0 >= availableProcessors) {
availableProcessors = 1;
}
int maxPoolSize = (availableProcessors > 5) ? availableProcessors * 2 : 5;
logger.debug("CPU Processors :%s MaxPoolSize:%s", availableProcessors, maxPoolSize);
threadPool = new ThreadPoolExecutor(
availableProcessors,
maxPoolSize,
60,
TimeUnit.SECONDS,
new LinkedBlockingQueue<Runnable>(maxPoolSize * 2),
Executors.defaultThreadFactory(),
new RejectedExecutionHandler() {
@Override
public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
logger.debug("Task:%s rejected", r.toString());
if (!executor.isShutdown()) {
executor.getQueue().poll();
executor.execute(r);
}
}
}
);
}
@PreDestroy
public void destroy() {
logger.debug("销毁线程池。。。");
if (null != threadPool && !threadPool.isShutdown()) {
threadPool.shutdown();
}
}
@Override
public void execute(@NotNull Runnable runnable) {
if (!TransactionSynchronizationManager.isSynchronizationActive()) {
runnable.run();
return;
}
List<Runnable> threadRunnables = RUNNABLES.get();
if (threadRunnables == null) {
threadRunnables = new ArrayList<Runnable>();
RUNNABLES.set(threadRunnables);
TransactionSynchronizationManager.registerSynchronization(this);
}
threadRunnables.add(runnable);
}
@Override
public void afterCommit() {
logger.debug("事务提交完成处理 ... ");
List<Runnable> threadRunnables = RUNNABLES.get();
for (int i = 0; i < threadRunnables.size(); i++) {
Runnable runnable = threadRunnables.get(i);
try {
threadPool.execute(runnable);
} catch (RuntimeException e) {
logger.error("", e);
}
}
}
@Override
public void afterCompletion(int status) {
logger.debug("事务处理完毕 .... ");
RUNNABLES.remove();
}
}
service代码
@Transactional
public boolean updateById(Hotel hotel) {
logger.info("----- into service -----");
//修改DB的数据
hotelMapper.updateById(hotel);
// 使用AfterCommitExecutor
afterCommitExecutor.execute(new Runnable() {
@Override
public void run() {
rabbitTemplate.convertAndSend(MqConstants.HOTEL_EXCHANGE, MqConstants.HOTEL_INSERT_KEY, hotel.getId().toString());
logger.info("----- rabbitmq send message -----");
}
});
logger.info("--------- out service ----------");
int n =0/0;
return false;
}