背景：在业务中，经常需要在执行数据库操作后（事务提交完成），发送消息或事件来异步调用其他组件执行相应的业务操作。
比如：用户注册成功后，发送激活码或激活邮件，如果用户保存后就执行异步操作发送激活码或激活邮件，但是前面用户保存后发生异常，数据库进行回滚，用户实际没有注册成功，但用户收到激活码或激活邮件。此时，我们就迫切要求数据库事务完成后再执行异步操作。
@Autowired
private UserDao userDao;
@Autowired
private JmsProducer jmsProducer;
public User saveUser(User user) {
// 保存用户
userDao.save(user);
// 发送激活码或激活邮件
jmsProducer.sendEmail(user.getId());
}
// -------------------------------------
public void sendEmail(int userId) {
/*
* 获取待接收邮件的用户。（如果上面的保存用户方法还未提交事务，则实际数据还未插入到数据库中，此时会返回null）
*/
User user = userDao.get(userId);
// 可能抛NullPointException
String email = user.getEmail();
mailService.send(email);
}
解决方案
1、Spring 4.2之后,使用注解@TransactionalEventListener
/**
* 业务Service
*/
@Service
@Transactional
public class FooService {
@Autowired
private ApplicationEventPublisher applicationEventPublisher;
public User saveUser(User user) {
userDao.save(user);
// 注册事件
applicationEventPublisher.publishEvent(new SavedUserEvent(user.getId()));
}
}
// -------------------------------------
/**
* 保存用户事件
*/
public class SavedUserEvent {
private int userId;
public SavedUserEvent(int userId) {
this.userId = userId;
}
// getter and setter
}
// ---------------------------------
/**
* 事件侦听，处理对应事件
*/
@Component
public class FooEventListener() {
@Autowired
private UserDao userDao;
@Autowired
private MailService mailService;
@TransactionalEventListener
public sendEmail(SavedUserEvent savedUserEvent) {
User user = userDao.get(userId);
String email = user.getEmail();
mailService.send(email);
}
}
2.使用TransactionSynchronizationManager 和 TransactionSynchronizationAdapter
@Autowired
private UserDao userDao;
@Autowired
private JmsProducer jmsProducer;
public User saveUser(User user) {
// 保存用户
userDao.save(user);
final int userId = user.getId();
// 事务提交后调用
TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronizationAdapter() {
@Override
public void afterCommit() {
jmsProducer.sendEmail(userId);
}
});
}
注意：上面的代码将在事务提交后执行.如果在非事务context中将抛出java.lang.IllegalStateException: Transaction synchronization is not active。
改进后代码：
@Autowired
private UserDao userDao;
@Autowired
private JmsProducer jmsProducer;
public User saveUser(User user) {
// 保存用户
userDao.save(user);
final int userId = user.getId();
// 兼容无论是否有事务
if(TransactionSynchronizationManager.isActualTransactionActive()) {
TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronizationAdapter() {
@Override
public void afterCommit() {
jmsProducer.sendEmail(userId);
}
});
} else {
jmsProducer.sendEmail(userId);
}
}
3.在上面2的基础上扩展TransactionSynchronizationAdapter
@Component("afterCommitExecutor")
public class AfterCommitExecutor extends TransactionSynchronizationAdapter implements Executor {
private static final ThreadLocal<List<Runnable>> RUNNABLES = new ThreadLocal<List<Runnable>>();
private ThreadPoolExecutor threadPool;
@PostConstruct
public void init() {
Logs.debug("初始化线程池。。。");
int availableProcessors = Runtime.getRuntime().availableProcessors();
if (0 >= availableProcessors) {
availableProcessors = 1;
}
int maxPoolSize = (availableProcessors > 5) ? availableProcessors * 2 : 5;
Logs.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) {
Logs.debug("Task:%s rejected", r.toString());
if (!executor.isShutdown()) {
executor.getQueue().poll();
executor.execute(r);
}
}
}
);
}
@PreDestroy
public void destroy() {
Logs.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() {
Logs.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) {
Logs.error("", e);
}
}
}
@Override
public void afterCompletion(int status) {
Logs.debug("事务处理完毕 .... ");
RUNNABLES.remove();
}
}
业务层使用：
@Autowired
private UserDao userDao;
@Autowired
private JmsProducer jmsProducer;
@Autowired
private AfterCommitExecutor afterCommitExecutor;
public User saveUser(User user) {
// 保存用户
userDao.save(user);
final int userId = user.getId();
// 使用AfterCommitExecutor
afterCommitExecutor.execute(new Runnable() {
@Override
public void run() {
jmsProducer.sendEmail(userId);
}
});
}
参考文章
Spring的TransactionEventListener
Spring Event 事件中的事务控制
Java Code Examples org.springframework.transaction.support.TransactionSynchronizationAdapter
Spring Events | Baeldung
Thinking about IT: Transaction synchronization callbacks in Spring Framework
