Spring 事务事件控制解决业务异步操作解耦 TransactionSynchronizationManager Transaction

Dear 丶 2023-09-25 18:45 113阅读 0赞

## 场景 ##

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

## 需求 ##

在成功修改数据库后，向Rabbitmq发送一条消息

## 解决方案 ##

### 1.使用TransactionSynchronizationManage 控制事务 ###

![在这里插入图片描述][6a4ed7473e6449bd92b071c35ab926c2.png]  
TransactionSynchronization 该类中有方法可以控制业务在提交事务后执行。  
![在这里插入图片描述][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。  
![在这里插入图片描述][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;
        }

[6a4ed7473e6449bd92b071c35ab926c2.png]: https://img-blog.csdnimg.cn/6a4ed7473e6449bd92b071c35ab926c2.png
[2f62dde535c24910a3e498e61df25d9f.png]: https://img-blog.csdnimg.cn/2f62dde535c24910a3e498e61df25d9f.png
[f8c35b70848e45e2a94705a6448a4283.png]: https://img-blog.csdnimg.cn/f8c35b70848e45e2a94705a6448a4283.png