事务（十七）

如何在数据库事务提交成功后进行异步操作

问题

业务场景

业务需求上经常会有一些边缘操作，比如主流程操作A：用户报名课程操作入库，边缘操作B：发送邮件或短信通知。

业务要求

  • 操作A操作数据库失败后，事务回滚，那么操作B不能执行。

  • 操作A执行成功后，操作B也必须执行成功

如何实现

  • 普通的执行A，之后执行B，是可以满足要求1，对于要求2通常需要设计补偿的操作

  • 一般边缘的操作，通常会设置成为异步的，以提升性能，比如发送MQ，业务系统负责事务成功后消息发送成功，然后接收系统负责保证通知成功完成

本文内容

如何在spring事务提交之后进行异步操作，这些异步操作必须得在该事务成功提交后才执行，回滚则不执行。

要点

  • 如何在spring事务提交之后操作

  • 如何把操作异步化

实现方案

使用TransactionSynchronizationManager在事务提交之后操作

 1 public void insert(TechBook techBook){
 2         bookMapper.insert(techBook);
 3        // send after tx commit but is async
 4         TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronizationAdapter() {
 5             @Override
 6             public void afterCommit() {
 7                 System.out.println("send email after transaction commit...");
 8             }
 9         }
10        );
11         ThreadLocalRandom random = ThreadLocalRandom.current();
12         if(random.nextInt() % 2 ==0){
13             throw new RuntimeException("test email transaction");
14         }
15         System.out.println("service end");
16     }

该方法就可以实现在事务提交之后进行操作

操作异步化

使用mq或线程池来进行异步，比如使用线程池：

 1 private final ExecutorService executorService = Executors.newFixedThreadPool(5);
 2     public void insert(TechBook techBook){
 3         bookMapper.insert(techBook);
 4  
 5 //        send after tx commit but is async
 6         TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronizationAdapter() {
 7             @Override
 8             public void afterCommit() {
 9                 executorService.submit(new Runnable() {
10                     @Override
11                     public void run() {
12                         System.out.println("send email after transaction commit...");
13                         try {
14                             Thread.sleep(10*1000);
15                         } catch (InterruptedException e) {
16                             e.printStackTrace();
17                         }
18                         System.out.println("complete send email after transaction commit...");
19                     }
20                 });
21             }
22         }
23         );
24  
25 //        async work but tx not work, execute even when tx is rollback
26 //        asyncService.executeAfterTxComplete();
27  
28         ThreadLocalRandom random = ThreadLocalRandom.current();
29         if(random.nextInt() % 2 ==0){
30             throw new RuntimeException("test email transaction");
31         }
32         System.out.println("service end");
33     }

封装以上两步

对于第二步来说，如果这类方法比较多的话，则写起来重复性太多，因而，抽象出来如下：

1.定义一个接口：

public interface AfterCommitExecutor extends Executor {

}

2.创建第一步定义的接口的实现类，并且spring通过@Component注入：

/**
 * 事务提交后，异步线程处理类
 */
@Component
public class AfterCommitExecutorImpl extends TransactionSynchronizationAdapter implements AfterCommitExecutor {
    private static final Logger LOGGER = LoggerFactory.getLogger(AfterCommitExecutorImpl.class);
    private static final ThreadLocal<List<Runnable>> RUNNABLES = new ThreadLocal<>();

    // 注入线程池对象，这个对象是系统线程池配置类定义的
    @Resource
    private TaskExecutor taskExecutor;

    @Override
    public void execute(Runnable runnable) {
        LOGGER.info("Submitting new runnable {} to run after commit", runnable);
        if (!TransactionSynchronizationManager.isSynchronizationActive()) {
            LOGGER.info("Transaction synchronization is NOT ACTIVE. Executing right now runnable {}", runnable);
            runnable.run();
            return;
        }
        List<Runnable> threadRunnables = RUNNABLES.get();
        if (threadRunnables == null) {
            threadRunnables = new ArrayList<>();
            RUNNABLES.set(threadRunnables);
            TransactionSynchronizationManager.registerSynchronization(this);
        }
        threadRunnables.add(runnable);
    }
 
    @Override
    public void afterCommit() {
        List<Runnable> threadRunnables = RUNNABLES.get();
        LOGGER.info("Transaction successfully committed, executing {} runnables", threadRunnables.size());
        for (int i = 0; i < threadRunnables.size(); i++) {
            Runnable runnable = threadRunnables.get(i);
            LOGGER.info("Executing runnable {}", runnable);
            try {
                taskExecutor.execute(runnable);
            } catch (RuntimeException e) {
                LOGGER.error("Failed to execute runnable " + runnable, e);
            }
        }
    }
 
    @Override
    public void afterCompletion(int status) {
        LOGGER.info("Transaction completed with status {}", status == STATUS_COMMITTED ? "COMMITTED" : "ROLLED_BACK");
        RUNNABLES.remove();
    }
 
}

线程池配置类定义TaskExecutor :

    @Bean
    public TaskExecutor taskExecutor() {
        ThreadPoolTaskExecutor taskExecutor = new ThreadPoolTaskExecutor();
        //线程池大小
        taskExecutor.setCorePoolSize(10);
        //线程池最大线程数
        taskExecutor.setMaxPoolSize(1000);
        //最大等待任务数
        taskExecutor.setQueueCapacity(1000);
        taskExecutor.setKeepAliveSeconds(3600);
        taskExecutor.initialize();
        return taskExecutor;
    }

3.在需要事务提交后运行异步操作的地方，这么使用：

        // 注入刚才定义的接口实现类
        @Autowired
    private AfterCommitExecutor afterCommitExecutor;

        //在事务方法中调用如下：
        // 上传pdf
        afterCommitExecutor.execute(() -> {
            FileParam fileParam = bulidFile(taskReformCertificate.getId(), IExportWord.REFORM_CERTIFICATE_2006_PATH, OssClientUtils.FILE_NAME_CERTIFICATE_NOTICE, IExportWord.PDF);
            String toOssPath = OssClientUtils.getReformFilePath(taskReformCertificate.getReformId(), OssClientUtils.FILE_NAME_CERTIFICATE_NOTICE + "." + IExportWord.PDF);
            String ossBucket = gunsProperties.getOssBucket();
            log.info("上传送达回证pdf --start--, ossBucket={}, toOssPath={},", ossBucket, toOssPath);
            OssClientUtils.uploadPdf2Oss(fileParam, ossBucket, toOssPath);
            log.info("上传送达回证pdf --end--");
        });

这样，上传pdf的操作就会在所在方法中的事务提交后，异步去执行上传pdf的操作，也就能读取刚才事务提交的数据库数据了。

关于Spring的Async

springboot使用Async可以参考：https://www.cnblogs.com/huanzi-qch/p/11231041.html

spring为了方便应用使用线程池进行异步化，默认提供了@Async注解，可以整个app使用该线程池，而且只要一个@Async注解在方法上面即可，省去重复的submit操作。关于async要注意的几点：

1、async的配置

<context:component-scan base-package="com.yami" />
   <!--配置@Async注解使用的线程池，这里的id随便命名，最后在task:annotation-driven executor= 指定上就可以-->
    <task:executor id="myExecutor" pool-size="5"/>
    <task:annotation-driven executor="myExecutor" />
<context:component-scan base-package="com.yami.web.controller"/>
<mvc:annotation-driven/>

这个必须配置在root context里头，而且web context不能扫描controller层外的注解，否则会覆盖掉。

2、async的调用问题

async方法的调用，不能由同类方法内部调用，否则拦截不生效，这是spring默认的拦截问题，必须在其他类里头调用另一个类中带有async的注解方法，才能起到异步效果。

3、事务问题

async方法如果也开始事务的话，要注意事务传播以及事务开销的问题。而且在async方法里头使用如上的TransactionSynchronizationManager.registerSynchronization不起作用，值得注意。

参考：https://blog.csdn.net/varyall/article/details/78923592

带着疑问去思考，然后串联，进而归纳总结，不断追问自己，进行自我辩证，像侦查嫌疑案件一样看待技术问题，漆黑的街道，你我一起寻找线索，你就是技术界大侦探福尔摩斯
【推广】 免费学中医，健康全家人
原文地址：https://www.cnblogs.com/cainiao-Shun666/p/14785960.html