Spring进阶篇(7)-TransactionSynchronizationManager(事务监听)

JAVA && Spring && SpringBoot2.x — 学习目录

TransactionSynchronizationManager是事务同步管理器。我们可以自定义实现TransactionSynchronization类,来监听Spring的事务操作。可以在事务提交之后,回调TransactionSynchronization类的方法。

1. TransactionSynchronizationManager在源码中的使用

在SpringCache的自定义CacheManager中。装饰Cache对象使其支持事务操作。即只有在事务提交成功之后,才会进行缓存。

源码位置:org.springframework.cache.transaction.TransactionAwareCacheDecorator#put

    @Override
    public void put(final Object key, @Nullable final Object value) {
        if (TransactionSynchronizationManager.isSynchronizationActive()) {
            TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronizationAdapter() {
                @Override
                public void afterCommit() {
                    TransactionAwareCacheDecorator.this.targetCache.put(key, value);
                }
            });
        }
        else {
            this.targetCache.put(key, value);
        }
    }

【JDBC中的connection详解】中,我们知道connection是线程不安全的,即需要为每一个数据库操作都获取一个Connection对象。事务操作可以看做是一个整体,必须使用同一个Connection进行操作。故在Spring中使用LocalThread(线程上下文)将Connection对象和线程绑定。

在Spring中的org.springframework.transaction.support.TransactionSynchronizationManager类中,便是使用ThreadLocal来为不同的事务线程提供独立的资源副本,并且同时维护这些事务的配置属性和运行状态。

2. Connection与TransactionSynchronizationManager关系

1. 请求事务方法时,调用dobegin()将事务信息保存到TransactionSynchronizationManager中:在该方法中主要流程是在数据库连接池中获取一个Connection对象,然后将Connection对象放入到ThreadLocal中。实际上该事务方法的信息均由TransactionSynchronizationManager类管理。

源码:org.springframework.jdbc.datasource.DataSourceTransactionManager#doBegin

   protected void doBegin(Object transaction, TransactionDefinition definition) {
        DataSourceTransactionObject txObject = (DataSourceTransactionObject) transaction;
        Connection con = null;

        try {
            if (!txObject.hasConnectionHolder() ||
                    txObject.getConnectionHolder().isSynchronizedWithTransaction()) {
                    //在可见的数据源(连接池)中获取Connection对象
                Connection newCon = obtainDataSource().getConnection();
                if (logger.isDebugEnabled()) {
                    logger.debug("Acquired Connection [" + newCon + "] for JDBC transaction");
                }
                        
                txObject.setConnectionHolder(new ConnectionHolder(newCon), true);
            }
            
            ...
          //关闭Connection对象的自动提交
           if (con.getAutoCommit()) {
                txObject.setMustRestoreAutoCommit(true);
                if (logger.isDebugEnabled()) {
                    logger.debug("Switching JDBC Connection [" + con + "] to manual commit");
                }
                con.setAutoCommit(false);
            }

            //将Connection对象绑定到Thread中
            if (txObject.isNewConnectionHolder()) {
                TransactionSynchronizationManager.bindResource(obtainDataSource(), txObject.getConnectionHolder());
            }
        }

        catch (Throwable ex) {
            if (txObject.isNewConnectionHolder()) {
                DataSourceUtils.releaseConnection(con, obtainDataSource());
                txObject.setConnectionHolder(null, false);
            }
            throw new CannotCreateTransactionException("Could not open JDBC Connection for transaction", ex);
        }
    }
public abstract class TransactionSynchronizationManager {
    private static final ThreadLocal<Map<Object, Object>> resources =
            new NamedThreadLocal<>("Transactional resources");

    public static void bindResource(Object key, Object value) throws IllegalStateException {
        Object actualKey = TransactionSynchronizationUtils.unwrapResourceIfNecessary(key);
        Assert.notNull(value, "Value must not be null");
       
        Map<Object, Object> map = resources.get();
        // set ThreadLocal Map if none found
        if (map == null) {
            map = new HashMap<>();
            resources.set(map);
        }
        //将Connection对象绑定到resources 上。
        Object oldValue = map.put(actualKey, value);
        ...
    }
}
  1. 执行sql语句时,实际上通过org.mybatis.spring.transaction.SpringManagedTransaction类直接获取Connection对象。

源码:org.mybatis.spring.transaction.SpringManagedTransaction#openConnection

  private void openConnection() throws SQLException {
    //在ThreadLocal中获取Connection对象
    this.connection = DataSourceUtils.getConnection(this.dataSource);
    this.autoCommit = this.connection.getAutoCommit();
    //在ThreadLocal中获取是否开启事务
    this.isConnectionTransactional = DataSourceUtils.isConnectionTransactional(this.connection, this.dataSource);
   ...
  }

源码:org.springframework.jdbc.datasource.DataSourceUtils#doGetConnection

    public static Connection doGetConnection(DataSource dataSource) throws SQLException {
        Assert.notNull(dataSource, "No DataSource specified");
        //在doBegin()方法中,已经将创建的Connection对象放入到TransactionSynchronizationManager中
        ConnectionHolder conHolder = (ConnectionHolder) TransactionSynchronizationManager.getResource(dataSource);
        if (conHolder != null && (conHolder.hasConnection() || conHolder.isSynchronizedWithTransaction())) {
            conHolder.requested();
            if (!conHolder.hasConnection()) {
                logger.debug("Fetching resumed JDBC Connection from DataSource");
                conHolder.setConnection(fetchConnection(dataSource));
            }
            //直接返回Thread存储的Connection对象。
            return conHolder.getConnection();
        }

        ...

        return con;
    }

3. TransactionSynchronizationManager的结构


public abstract class TransactionSynchronizationManager {

     //线程上下文中保存着【线程池对象:ConnectionHolder】的Map对象。线程可以通过该属性获取到同一个Connection对象。
    private static final ThreadLocal<Map<Object, Object>> resources = new NamedThreadLocal<>("Transactional resources");

    //事务同步器,是Spring交由程序员进行扩展的代码,每个线程可以注册N个事务同步器。
    private static final ThreadLocal<Set<TransactionSynchronization>> synchronizations = new NamedThreadLocal<>("Transaction synchronizations");
    
    // 事务的名称  
    private static final ThreadLocal<String> currentTransactionName = new NamedThreadLocal<>("Current transaction name");
    // 事务是否是只读  
    private static final ThreadLocal<Boolean> currentTransactionReadOnly = new NamedThreadLocal<>("Current transaction read-only status");
    // 事务的隔离级别
    private static final ThreadLocal<Integer> currentTransactionIsolationLevel = new NamedThreadLocal<>("Current transaction isolation level");
    // 事务是否开启   actual:真实的
    private static final ThreadLocal<Boolean> actualTransactionActive = new NamedThreadLocal<>("Actual transaction active");
}

org.springframework.transaction.interceptor.TransactionInterceptor#invoke中,对事务方法进行拦截处理。在createTransactionIfNecessary创建TransactionInfo对象时,会调用AbstractPlatformTransactionManager#prepareSynchronization方法初始化事务同步器。

    protected void prepareSynchronization(DefaultTransactionStatus status, TransactionDefinition definition) {
        if (status.isNewSynchronization()) {
            TransactionSynchronizationManager.setActualTransactionActive(status.hasTransaction());
            TransactionSynchronizationManager.setCurrentTransactionIsolationLevel(
                    definition.getIsolationLevel() != TransactionDefinition.ISOLATION_DEFAULT ?
                            definition.getIsolationLevel() : null);
            TransactionSynchronizationManager.setCurrentTransactionReadOnly(definition.isReadOnly());
            TransactionSynchronizationManager.setCurrentTransactionName(definition.getName());
            //初始化事务同步器
            TransactionSynchronizationManager.initSynchronization();
        }
    }

4. TransactionSynchronization

这个类是程序员对事务同步的扩展点:用于事务同步回调的接口。

public interface TransactionSynchronization extends Flushable {

    /** Completion status in case of proper commit. */
    int STATUS_COMMITTED = 0;

    /** Completion status in case of proper rollback. */
    int STATUS_ROLLED_BACK = 1;

    /** Completion status in case of heuristic mixed completion or system errors. */
    int STATUS_UNKNOWN = 2;


    /**
     * 事务挂起
     * Supposed to unbind resources from TransactionSynchronizationManager if managing any.
     * @see TransactionSynchronizationManager#unbindResource
     */
    default void suspend() {
    }

    /**
     * 事务恢复
     * Supposed to rebind resources to TransactionSynchronizationManager if managing any.
     * @see TransactionSynchronizationManager#bindResource
     */
    default void resume() {
    }

    /**
     * 将基础会话刷新到数据存储区(如果适用),比如Hibernate/JPA的Session
     * @see org.springframework.transaction.TransactionStatus#flush()
     */
    @Override
    default void flush() {
    }

    /**
     * 在事务提交前触发,此处若发生异常,会导致回滚。
     * @see #beforeCompletion
     */
    default void beforeCommit(boolean readOnly) {
    }

    /**
     * 在beforeCommit之后,commit/rollback之前执行。即使异常,也不会回滚。
     * @see #beforeCommit
     * @see #afterCompletion
     */
    default void beforeCompletion() {
    }

    /**
     * 事务提交后执行。
     */
    default void afterCommit() {
    }

    /**
     * 事务提交/回滚执行
     */
    default void afterCompletion(int status) {
    }

一般而言,我们在TransactionSynchronization使用最多的是afterCommitafterCompletion方法。可以在事务执行完毕之后,直接调用afterCommit()方法进行异步通知。

我们在doCommit()方法中提交事务后,在cleanupAfterCompletion对connection进行重置,即我们依旧可以在afterCommit()回调中对数据库进行操作。

    private void processCommit(DefaultTransactionStatus status) throws TransactionException {
        try {
            //提交事务
            doCommit(status);
            ...
            try {
             //回调所有事务同步器的afterCommit方法。
                triggerAfterCommit(status);
            }
            finally {
            //回调所有事务同步器的afterCompletion方法。
                triggerAfterCompletion(status, TransactionSynchronization.STATUS_COMMITTED);
            }

        }
        finally {
           //清除TransactionSynchronizationManager的ThreadLocal绑定的数据。
           //解除Thread绑定的resources资源。
           //将Commit设置为自动提交。
          //清理ConnectionHolder资源。
            cleanupAfterCompletion(status);
        }
    }

文章参考

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

自定义springcache实现事务提交后处理缓存

Spring事务监听机制---使用@TransactionalEventListener处理数据库事务提交成功后再执行操作(附:Spring4.2新特性讲解)

最后编辑于
©著作权归作者所有,转载或内容合作请联系作者
  • 序言:七十年代末,一起剥皮案震惊了整个滨河市,随后出现的几起案子,更是在滨河造成了极大的恐慌,老刑警刘岩,带你破解...
    沈念sama阅读 204,530评论 6 478
  • 序言:滨河连续发生了三起死亡事件,死亡现场离奇诡异,居然都是意外死亡,警方通过查阅死者的电脑和手机,发现死者居然都...
    沈念sama阅读 86,403评论 2 381
  • 文/潘晓璐 我一进店门,熙熙楼的掌柜王于贵愁眉苦脸地迎上来,“玉大人,你说我怎么就摊上这事。” “怎么了?”我有些...
    开封第一讲书人阅读 151,120评论 0 337
  • 文/不坏的土叔 我叫张陵,是天一观的道长。 经常有香客问我,道长,这世上最难降的妖魔是什么? 我笑而不...
    开封第一讲书人阅读 54,770评论 1 277
  • 正文 为了忘掉前任,我火速办了婚礼,结果婚礼上,老公的妹妹穿的比我还像新娘。我一直安慰自己,他们只是感情好,可当我...
    茶点故事阅读 63,758评论 5 367
  • 文/花漫 我一把揭开白布。 她就那样静静地躺着,像睡着了一般。 火红的嫁衣衬着肌肤如雪。 梳的纹丝不乱的头发上,一...
    开封第一讲书人阅读 48,649评论 1 281
  • 那天,我揣着相机与录音,去河边找鬼。 笑死,一个胖子当着我的面吹牛,可吹牛的内容都是我干的。 我是一名探鬼主播,决...
    沈念sama阅读 38,021评论 3 398
  • 文/苍兰香墨 我猛地睁开眼,长吁一口气:“原来是场噩梦啊……” “哼!你这毒妇竟也来了?” 一声冷哼从身侧响起,我...
    开封第一讲书人阅读 36,675评论 0 258
  • 序言:老挝万荣一对情侣失踪,失踪者是张志新(化名)和其女友刘颖,没想到半个月后,有当地人在树林里发现了一具尸体,经...
    沈念sama阅读 40,931评论 1 299
  • 正文 独居荒郊野岭守林人离奇死亡,尸身上长有42处带血的脓包…… 初始之章·张勋 以下内容为张勋视角 年9月15日...
    茶点故事阅读 35,659评论 2 321
  • 正文 我和宋清朗相恋三年,在试婚纱的时候发现自己被绿了。 大学时的朋友给我发了我未婚夫和他白月光在一起吃饭的照片。...
    茶点故事阅读 37,751评论 1 330
  • 序言:一个原本活蹦乱跳的男人离奇死亡,死状恐怖,灵堂内的尸体忽然破棺而出,到底是诈尸还是另有隐情,我是刑警宁泽,带...
    沈念sama阅读 33,410评论 4 321
  • 正文 年R本政府宣布,位于F岛的核电站,受9级特大地震影响,放射性物质发生泄漏。R本人自食恶果不足惜,却给世界环境...
    茶点故事阅读 39,004评论 3 307
  • 文/蒙蒙 一、第九天 我趴在偏房一处隐蔽的房顶上张望。 院中可真热闹,春花似锦、人声如沸。这庄子的主人今日做“春日...
    开封第一讲书人阅读 29,969评论 0 19
  • 文/苍兰香墨 我抬头看了看天上的太阳。三九已至,却和暖如春,着一层夹袄步出监牢的瞬间,已是汗流浃背。 一阵脚步声响...
    开封第一讲书人阅读 31,203评论 1 260
  • 我被黑心中介骗来泰国打工, 没想到刚下飞机就差点儿被人妖公主榨干…… 1. 我叫王不留,地道东北人。 一个月前我还...
    沈念sama阅读 45,042评论 2 350
  • 正文 我出身青楼,却偏偏与公主长得像,于是被迫代替她去往敌国和亲。 传闻我的和亲对象是个残疾皇子,可洞房花烛夜当晚...
    茶点故事阅读 42,493评论 2 343