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

标签: spring async transaction | 发表时间:2016-01-01 17:02 | 作者:xixicat
出处:http://segmentfault.com/blogs

问题

业务场景

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

业务要求

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

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

如何实现

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

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

本文内容

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

要点

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

  • 如何把操作异步化

实现方案

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

  public void insert(TechBook techBook){
        bookMapper.insert(techBook);
       // send after tx commit but is async
        TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronizationAdapter() {
            @Override
            public void afterCommit() {
                System.out.println("send email after transaction commit...");
            }
        }
       );
        ThreadLocalRandom random = ThreadLocalRandom.current();
        if(random.nextInt() % 2 ==0){
            throw new RuntimeException("test email transaction");
        }
        System.out.println("service end");
    }

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

操作异步化

使用mq或线程池来进行异步,比如使用线程池:

  private final ExecutorService executorService = Executors.newFixedThreadPool(5);
    public void insert(TechBook techBook){
        bookMapper.insert(techBook);
 
//        send after tx commit but is async
        TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronizationAdapter() {
            @Override
            public void afterCommit() {
                executorService.submit(new Runnable() {
                    @Override
                    public void run() {
                        System.out.println("send email after transaction commit...");
                        try {
                            Thread.sleep(10*1000);
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                        System.out.println("complete send email after transaction commit...");
                    }
                });
            }
        }
        );
 
//        async work but tx not work, execute even when tx is rollback
//        asyncService.executeAfterTxComplete();
 
        ThreadLocalRandom random = ThreadLocalRandom.current();
        if(random.nextInt() % 2 ==0){
            throw new RuntimeException("test email transaction");
        }
        System.out.println("service end");
    }

封装以上两步

对于第二步来说,如果这类方法比较多的话,则写起来重复性太多,因而,抽象出来如下:
这里改造了 azagorneanu的代码:

  public interface AfterCommitExecutor extends Executor {
}
 
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 java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
 
@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<List<Runnable>>();
    private ExecutorService threadPool = Executors.newFixedThreadPool(5);
 
    @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<Runnable>();
            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 {
                threadPool.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();
    }
 
}
public void insert(TechBook techBook){
        bookMapper.insert(techBook);
 
//        send after tx commit but is async
//        TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronizationAdapter() {
//            @Override
//            public void afterCommit() {
//                executorService.submit(new Runnable() {
//                    @Override
//                    public void run() {
//                        System.out.println("send email after transaction commit...");
//                        try {
//                            Thread.sleep(10*1000);
//                        } catch (InterruptedException e) {
//                            e.printStackTrace();
//                        }
//                        System.out.println("complete send email after transaction commit...");
//                    }
//                });
//            }
//        }
//        );
 
        //send after tx commit and is async
        afterCommitExecutor.execute(new Runnable() {
            @Override
            public void run() {
                try {
                    Thread.sleep(5*1000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                System.out.println("send email after transactioin commit");
            }
        });
 
//        async work but tx not work, execute even when tx is rollback
//        asyncService.executeAfterTxComplete();
 
        ThreadLocalRandom random = ThreadLocalRandom.current();
        if(random.nextInt() % 2 ==0){
            throw new RuntimeException("test email transaction");
        }
        System.out.println("service end");
    }

关于Spring的Async

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" />

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

  <context:component-scan base-package="com.yami.web.controller"/>
<mvc:annotation-driven/>

2、async的调用问题

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

3、事务问题

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

相关 [数据库 成功 异步] 推荐:

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

- - SegmentFault 最新的文章
业务需求上经常会有一些边缘操作,比如主流程操作A:用户报名课程操作入库,边缘操作B:发送邮件或短信通知. 操作A操作数据库失败后,事务回滚,那么操作B不能执行. 操作A执行成功后,操作B也必须执行成功. 普通的执行A,之后执行B,是可以满足要求1,对于要求2通常需要设计补偿的操作. 一般边缘的操作,通常会设置成为异步的,以提升性能,比如发送MQ,业务系统负责事务成功后消息发送成功,然后接收系统负责保证通知成功完成.

[tcp] 异步connect是否成功?

- 旺旺 - 斯巴达第二季
我原先的client端代码流程如下:  . 设为异步socket(fcntl). connect到远端(此时connect调用返回非0,但errno为EINPROGRESS,表示正在建立连接中). 捕获到EPOLLOUT事件,此时便认为connect已经成功,client端开始发消息. 这个过程通常能够运转,但是线上环境复杂多变,如果发生这种情况:server进程调用listen开始侦听后,被gdb或信号挂住了,此时异步connect会怎样.

1.2 HBase使用场景和成功案例 - 异步社区

- -
1.2 HBase使用场景和成功案例. 有时候了解软件产品的最好方法是看看它是怎么用的. 它可以解决什么问题和这些解决方案如何适用于大型应用架构,这些能够告诉你很多. 因为HBase有许多公开的产品部署案例,我们正好可以这么做. 本节将详细介绍一些成功使用HBase的使用场景. 不要自我限制,认为HBase只能在这些使用场景下使用.

数据库sharding

- - 数据库 - ITeye博客
当团队决定自行实现sharding的时候,DAO层可能是嵌入sharding逻辑的首选位置,因为在这个层面上,每一个DAO的方法都明确地知道需要访问的数据表以及查询参数,借助这些信息可以直接定位到目标shard上,而不必像框架那样需要对SQL进行解析然后再依据配置的规则进行路由. 另一个优势是不会受ORM框架的制约.

数据库索引

- - CSDN博客推荐文章
索引是由用户创建的、能够被修改和删除的、实际存储于数据库中的物理存在;创建索引的目的是使用户能够从整体内容直接查找到某个特定部分的内容. 一般来说,索引能够提高查询,但是会增加额外的空间消耗,并且降低删除、插入和修改速度. 1.聚集索引:表数据按照索引的顺序来存储的. 2.非聚集索引:表数据存储顺序与索引顺序无关.

数据库事务

- - 数据库 - ITeye博客
事务传播发生在类似以下情形:. 假设methodB的配置是:. 如果methodA在事务里,那么methodB也在这个事务中运行. 如果methodA不在事务里,那么methodB重新建立一个事务运行. 如果methodA在事务里,那么methodB也在这个事务中运行. 如果methodA不在是事务里,那么methodB在非事务中运行.

数据库优化

- - 数据库 - ITeye博客
程序运行效率,优化应用程序,在SP编写过程中应该注意以下几点: . a) SQL的使用规范: .   i.尽量避免大事务操作,慎用holdlock子句,提高系统并发能力.   ii.尽量避免反复访问同一张或几张表,尤其是数据量较大的表,可以考虑先根据条件提取数据到临时表中,然后再做连接.   iii.尽量避免使用游标,因为游标的效率较差,如果游标操作的数据超过1万行,那么就应该改写;如果使用了游标,就要尽量避免在游标循环中再进行表连接的操作.

数据库调优

- - 数据库 - ITeye博客
1、1、调整数据结构的设计. 这一部分在开发信息系统之前完成,程序员需要考虑是否使用ORACLE数据库的分区功能,对于经常访问的数据库表是否需要建立索引等. 这一部分也是在开发信息系统之前完成,程序员在这一步需要考虑应用程序使用什么样的体系结构,是使用传统的Client/Server两层体系结构,还是使用Browser/Web/Database的三层体系结构.

MySQL数据库的修复

- Xin - 博客园-首页原创精华区
找到mysql的安装目录的bin/myisamchk工具,在命令行中输入:. 然后myisamchk 工具会帮助你恢复数据表的索引. 好象也不用重新启动mysql,问题就解决了. 当你试图修复一个被破坏的表的问题时,有三种修复类型. 如果你得到一个错误信息指出一个临时文件不能建立,删除信息所指出的文件并再试一次--这通常是上一次修复操作遗留下来的.