Spring 事务事件控制 解决业务异步操作解耦 @TransactionalEventListener 和 TransactionSynchronizationManager TransactionSynchronizationAdapter

背景：在业务中，经常需要在执行数据库操作后（事务提交完成），发送消息或事件来异步调用其他组件执行相应的业务操作。
比如：用户注册成功后，发送激活码或激活邮件，如果用户保存后就执行异步操作发送激活码或激活邮件，但是前面用户保存后发生异常，数据库进行回滚，用户实际没有注册成功，但用户收到激活码或激活邮件。此时，我们就迫切要求数据库事务完成后再执行异步操作。

@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