Springトランザクション伝播の実現原理の概要

24896 ワード

本文はみんなと一緒にSpring事務の関連するソースコードを分析して、紙面は比較的に長くて、コードの断片は比較的に多くて、コンピュータを使って読むことを提案します
本文の目標

Springトランザクション管理コアインタフェース

の理解

Springトランザクション管理のコアロジック

を理解する

トランザクションの伝播タイプとその実現原理

を理解する
バージョン#バージョン#
SpringBoot 2.3.3.RELEASE
トランザクションの伝播とは?
Springはトランザクション制御をカプセル化するほか、トランザクションの伝播という概念を抽象化している.トランザクションの伝播はリレーショナル・データベースで定義されたものではなく、Springがトランザクションをカプセル化する際に行った拡張機能であり、@Transactionalを通じてトランザクションの伝播を指定することができる.具体的なタイプは以下の通りである.
トランザクション伝播の動作タイプ
説明PROPAGATION_REQUIRED
現在トランザクションがない場合は、トランザクションを新規作成し、すでにトランザクションが存在する場合は、このトランザクションに追加します.Springのデフォルトのトランザクション伝播タイプPROPAGATION_SUPPORTS
現在のトランザクションをサポートし、現在トランザクションがない場合は非トランザクションで実行します.PROPAGATION_MANDATORY
現在のトランザクションを使用して、現在トランザクションがない場合は例外を放出します.PROPAGATION_REQUIRES_NEW
新しいトランザクションを作成します.現在トランザクションが存在する場合は、現在のトランザクションを一時停止します.PROPAGATION_NOT_SUPPORTED
非トランザクションで操作を実行し、現在トランザクションが存在する場合は、現在のトランザクションを保留します.PROPAGATION_NEVER
非トランザクションで実行され、現在トランザクションが存在する場合は例外が放出されます.PROPAGATION_NESTED
現在トランザクションが存在する場合は、ネストされたトランザクション内で実行されます.現在トランザクションがない場合は、PROPAGATION_と実行します.REQUIRD同様の操作.
くりを一つあげる
ネストされたトランザクションを例に

@Service
public class DemoServiceImpl implements DemoService {
    @Autowired
    private JdbcTemplate jdbcTemplate;

    @Autowired
    private DemoServiceImpl self;

    @Transactional
    @Override
    public void insertDB() {
        String sql = "INSERT INTO sys_user(`id`, `username`) VALUES (?, ?)";
        jdbcTemplate.update(sql, uuid(), "taven");

        try {
            //         ，           
            self.nested();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    @Transactional(propagation = Propagation.NESTED)
    @Override
    public void nested() {
        String sql = "INSERT INTO sys_user(`id`, `username`) VALUES (?, ?)";
        jdbcTemplate.update(sql, uuid(), "nested");

        throw new RuntimeException("rollback nested");

    }

    private String uuid() {
        return UUID.randomUUID().toString();
    }
}

上記のコードでは、nested()メソッドはトランザクション伝播タイプがネストされていることをマークし、nested()で例外が投げ出されるとnested()メソッドのsqlのみがロールバックされ、insertDB()メソッドで実行されたsqlには影響しません.
注意:サービスが内部メソッドを呼び出す場合、this呼び出しを直接使用すると、トランザクションは有効になりません.したがって、this呼び出しを使用することは、外部のエージェントクラスをスキップすることに相当するため、AOPは有効ではなく、トランザクションは使用できません.
考える
SpringトランザクションはAOPによって実現されることはよく知られていますが、私たち自身がAOP制御トランザクションを書いたらどうすればいいのでしょうか.

//    
public Object invokeWithinTransaction() {
    //     
    connection.beginTransaction();
    try {
        //       
        Object result = invoke();
        //     
        connection.commit();
        return result;
    } catch(Exception e) {
        //        
        connection.rollback();
        throw e;
    } 
    
}

その上で、もし私たちが自分でやったら、事務の伝播はどのように実現するかを考えてみましょう.PROPAGATION_REQUIREDを例にとると、これは簡単なようです.現在トランザクションがあるかどうかを判断します(すでに存在するトランザクション・オブジェクトをThreadLocalで格納することも考えられます).トランザクションがある場合は、新しいトランザクションは開かれません.逆に、トランザクションがない場合は、新しいトランザクションを作成します.
トランザクションが現在のスライスによって開かれている場合は、トランザクションをコミット/ロールバックし、逆に処理しません.
では、トランザクション伝播で説明されている現在のトランザクションの保留(一時停止)と埋め込みトランザクションはどのように実現されますか?
ソース入手
トランザクション伝播に関連するソースコードを読むには、Springトランザクション管理のコアインタフェースとクラスについて説明します.

TransactionDefinition

このインタフェースは、トランザクションのすべてのプロパティ(独立性レベル、伝播タイプ、タイムアウト時間など)を定義し、日常開発でよく使用されている@Transactionalは、実際にはTransactionDefinitionに変換されます.

TransactionStatus

トランザクションのステータス、最も一般的な実装DefaultTransactionStatusを例にとると、このクラスには現在のトランザクション・オブジェクト、savepoint、現在保留中のトランザクション、完了するかどうか、ロールバックのみを格納するなど

TransactionManager

これは空のインタフェースで、直接彼のinterfaceを継承するPlatformTransactionManager(私たちが普段使っているのはこれで、デフォルトの実装クラスDataSourceTransactionManager)とReactiveTransactionManager(応答トランザクションマネージャ、本稿の重点ではないので、多くは言いません)があります.
上記の2つのインタフェースから見ると、TransactionManagerの主な役割は

TransactionDefinitionでトランザクションを開き、TransactionStatus

に戻ります.

TransactionStatusを介してトランザクションをコミット、ロールバックする(実際にトランザクションを開くConnectionは通常TransactionStatusに格納される)

public interface PlatformTransactionManager extends TransactionManager {
    
    TransactionStatus getTransaction(@Nullable TransactionDefinition definition)
            throws TransactionException;

    
    void commit(TransactionStatus status) throws TransactionException;

    
    void rollback(TransactionStatus status) throws TransactionException;

}

TransactionInterceptor

トランザクション・ブロッカー、トランザクションAOPのコア・クラス(レスポンス・トランザクション、プログラミング・トランザクション、および一般的な標準トランザクションをサポート)
次に、トランザクションロジックのエントリTransactionInterceptorから着手し、Springトランザクション管理のコアロジックとトランザクション伝播の実現を見てみましょう.
TransactionInterceptor
TransactionInterceptorはMethodInvocation(これはAOPを実現する方法の1つである)を実現し、そのコアロジックは親TransactionAspectSupportにあり、方法位置:TransactionInterceptor::invokeWithinTransaction

    protected Object invokeWithinTransaction(Method method, @Nullable Class> targetClass,
            final InvocationCallback invocation) throws Throwable {
        // If the transaction attribute is null, the method is non-transactional.
        TransactionAttributeSource tas = getTransactionAttributeSource();
        //         TransactionAttribute extends TransactionDefinition
        final TransactionAttribute txAttr = (tas != null ? tas.getTransactionAttribute(method, targetClass) : null);
        //                     
        //         Spring           TransactionManager
        final TransactionManager tm = determineTransactionManager(txAttr);

        //             ...
        
        PlatformTransactionManager ptm = asPlatformTransactionManager(tm);
        final String joinpointIdentification = methodIdentification(method, targetClass, txAttr);

        if (txAttr == null || !(ptm instanceof CallbackPreferringPlatformTransactionManager)) {
            // Standard transaction demarcation with getTransaction and commit/rollback calls.
            TransactionInfo txInfo = createTransactionIfNecessary(ptm, txAttr, joinpointIdentification);

            Object retVal;
            try {
                // This is an around advice: Invoke the next interceptor in the chain.
                // This will normally result in a target object being invoked.
                //             ，          ，          
                retVal = invocation.proceedWithInvocation();
            }
            catch (Throwable ex) {
                // target invocation exception
                //               （      ）
                completeTransactionAfterThrowing(txInfo, ex);
                throw ex;
            }
            finally {
                cleanupTransactionInfo(txInfo);
            }

            if (retVal != null && vavrPresent && VavrDelegate.isVavrTry(retVal)) {
                // Set rollback-only in case of Vavr failure matching our rollback rules...
                TransactionStatus status = txInfo.getTransactionStatus();
                if (status != null && txAttr != null) {
                    retVal = VavrDelegate.evaluateTryFailure(retVal, txAttr, status);
                }
            }
            //              
            commitTransactionAfterReturning(txInfo);
            return retVal;
        }

        //             ...
    }

ここにはコードがたくさんあります.注釈の位置によって、コアロジックを整理することができます.

現在のトランザクション属性を取得し、@Transactionalで定義できる注釈トランザクションを例にとるトランザクションマネージャ

createTransactionIfNecessary、トランザクション

を作成する必要があるかどうかを判断します.

invocation.proceedWithInvocationはブロッキングチェーンを実行し、最終的にはターゲットメソッド

に実行する.

completeTransactionAfterThrowing異常が投げ出された後、このトランザクションを完了し、コミットまたはロールバックし、この異常

を投げ出す.

commitTransactionAfterReturningメソッドの名前から見ると、このメソッドはトランザクションをコミットします.

しかし、ソースコードに深く入ると、この方法にはロールバックロジックも含まれており、具体的な動作は現在のTransactionStatusのいくつかの状態に基づいて決定されます(つまり、現在のTransactionStatusを設定することで、トランザクションロールバックを制御することができます.必ずしも例外を放出することはできません).詳細はAbstractPlatformTransactionManager::commitを参照してください.
続けて、createTransactionIfNecessaryが何をしたか見てみましょう
TransactionAspectSupport::createTransactionIfNecessary

    protected TransactionInfo createTransactionIfNecessary(@Nullable PlatformTransactionManager tm,
            @Nullable TransactionAttribute txAttr, final String joinpointIdentification) {

        // If no name specified, apply method identification as transaction name.
        if (txAttr != null && txAttr.getName() == null) {
            txAttr = new DelegatingTransactionAttribute(txAttr) {
                @Override
                public String getName() {
                    return joinpointIdentification;
                }
            };
        }

        TransactionStatus status = null;
        if (txAttr != null) {
            if (tm != null) {
                //            
                status = tm.getTransaction(txAttr);
            }
            else {
                if (logger.isDebugEnabled()) {
                    logger.debug("Skipping transactional joinpoint [" + joinpointIdentification +
                            "] because no transaction manager has been configured");
                }
            }
        }
        
        return prepareTransactionInfo(tm, txAttr, joinpointIdentification, status);
    }

createTransactionIfNecessaryのコアロジック

PlatformTransactionManager(トランザクションマネージャ)を使用してトランザクション

をオープンします.

TransactionAspectSupport.currentTransactionStatus()事務情報の準備、これは具体的に何をしたのか、後で

についてお話しします.prepareTransactionInfoを引き続き見ると、この方法はPlatformTransactionManager::getTransactionを実現するのは1つしかない.

    public final TransactionStatus getTransaction(@Nullable TransactionDefinition definition)
            throws TransactionException {

        // Use defaults if no transaction definition given.
        TransactionDefinition def = (definition != null ? definition : TransactionDefinition.withDefaults());

        //       ，       AbstractPlatformTransactionManager        
        Object transaction = doGetTransaction();
        boolean debugEnabled = logger.isDebugEnabled();

        //         
        if (isExistingTransaction(transaction)) {
            // Existing transaction found -> check propagation behavior to find out how to behave.
            return handleExistingTransaction(def, transaction, debugEnabled);
        }

        // Check definition settings for new transaction.
        if (def.getTimeout() < TransactionDefinition.TIMEOUT_DEFAULT) {
            throw new InvalidTimeoutException("Invalid transaction timeout", def.getTimeout());
        }

        //     PROPAGATION_MANDATORY,          
        // No existing transaction found -> check propagation behavior to find out how to proceed.
        if (def.getPropagationBehavior() == TransactionDefinition.PROPAGATION_MANDATORY) {
            throw new IllegalTransactionStateException(
                    "No existing transaction found for transaction marked with propagation 'mandatory'");
        }
        // PROPAGATION_REQUIRED, PROPAGATION_REQUIRES_NEW, PROPAGATION_NESTED           
        else if (def.getPropagationBehavior() == TransactionDefinition.PROPAGATION_REQUIRED ||
                def.getPropagationBehavior() == TransactionDefinition.PROPAGATION_REQUIRES_NEW ||
                def.getPropagationBehavior() == TransactionDefinition.PROPAGATION_NESTED) {
            SuspendedResourcesHolder suspendedResources = suspend(null);
            if (debugEnabled) {
                logger.debug("Creating new transaction with name [" + def.getName() + "]: " + def);
            }
            try {
                //     
                return startTransaction(def, transaction, debugEnabled, suspendedResources);
            }
            catch (RuntimeException | Error ex) {
                resume(null, suspendedResources);
                throw ex;
            }
        }
        else {
            // Create "empty" transaction: no actual transaction, but potentially synchronization.
            if (def.getIsolationLevel() != TransactionDefinition.ISOLATION_DEFAULT && logger.isWarnEnabled()) {
                logger.warn("Custom isolation level specified but no actual transaction initiated; " +
                        "isolation level will effectively be ignored: " + def);
            }
            boolean newSynchronization = (getTransactionSynchronization() == SYNCHRONIZATION_ALWAYS);
            return prepareTransactionStatus(def, null, true, newSynchronization, debugEnabled, null);
        }
    }

コードが多いので、注釈部分に注目すればいいです.

AbstractPlatformTransactionManager::getTransaction現在のトランザクション

を取得する.

トランザクションが存在する場合、doGetTransaction処理が呼び出されます.これは後で

に説明します.
次に、トランザクションの伝播に基づいてトランザクションを開くかどうかを決定します.

トランザクション伝播タイプがhandleExistingTransactionであり、トランザクションが存在しない場合、例外

が放出されます.

伝播タイプがPROPAGATION_MANDATORYであり、現在トランザクションが存在しない場合、PROPAGATION_REQUIRED, PROPAGATION_REQUIRES_NEW, PROPAGATION_NESTEDを呼び出してトランザクション

を作成する.

3 3、4を満たさない場合、たとえばstartTransactionでは、トランザクションの同期が実行されますが、真のトランザクション

は作成されません.
Spring
トランザクションの同期は前のブログで、転送ゲートについて話しました.
https://www.jianshu.com/p/788...
Spring現在のトランザクションの管理方法
次に、上記のPROPAGATION_NOT_SUPPORTED、doGetTransactionについて説明します.この2つの方法は、異なるTransactionManagerによって独自に実現されます.
SpringBootのデフォルトのTransactionManager、DataSourceTransactionManagerを例に挙げます.

    @Override
    protected Object doGetTransaction() {
        DataSourceTransactionObject txObject = new DataSourceTransactionObject();
        txObject.setSavepointAllowed(isNestedTransactionAllowed());
        ConnectionHolder conHolder =
                (ConnectionHolder) TransactionSynchronizationManager.getResource(obtainDataSource());
        txObject.setConnectionHolder(conHolder, false);
        return txObject;
    }

    @Override
    protected boolean isExistingTransaction(Object transaction) {
        DataSourceTransactionObject txObject = (DataSourceTransactionObject) transaction;
        return (txObject.hasConnectionHolder() && txObject.getConnectionHolder().isTransactionActive());
    }

handleExistingTransactionと併せて見ると、AbstractPlatformTransactionManager::getTransactionは実際に現在の接続を取得している.現在トランザクションが存在するかどうかを判断します.DataSourceTransactionObjectオブジェクトにconnectionが含まれているかどうか、およびconnectionがトランザクションを開いているかどうかを判断します.doGetTransactionが現在のconnectionを取得するロジックを引き続き見てみましょう
TransactionSynchronizationManager::getResource

    private static final ThreadLocal> resources =
            new NamedThreadLocal<>("Transactional resources");
    
    @Nullable
    // TransactionSynchronizationManager::getResource
    public static Object getResource(Object key) {
        // DataSourceTransactionManager       ，      key
        
        // TransactionSynchronizationUtils::unwrapResourceIfNecessary   key    ，         
        //          
        Object actualKey = TransactionSynchronizationUtils.unwrapResourceIfNecessary(key);
        Object value = doGetResource(actualKey);
        if (value != null && logger.isTraceEnabled()) {
            logger.trace("Retrieved value [" + value + "] for key [" + actualKey + "] bound to thread [" +
                    Thread.currentThread().getName() + "]");
        }
        return value;
    }

    /**
     * Actually check the value of the resource that is bound for the given key.
     */
    @Nullable
    private static Object doGetResource(Object actualKey) {
        Map

ここで、DataSourceTransactionManagerがスレッド間の接続をどのように管理しているかがわかります.ThreadLocalにはMapが格納されています.keyはデータソースオブジェクトで、valueはデータソースの現在のスレッドの接続です.
DataSourceTransactionManagerは、トランザクションを開始すると、TransactionSynchronizationManager.getResource(obtainDataSource())を呼び出して、指定したデータ・ソースのConnectionを現在のスレッドにバインドします.
AbstractPlatformTransactionManager::handleExistingTransaction
トランザクションが存在する場合、どのように処理するかを振り返ってみましょう.

    private TransactionStatus handleExistingTransaction(
            TransactionDefinition definition, Object transaction, boolean debugEnabled)
            throws TransactionException {

        //                       
        if (definition.getPropagationBehavior() == TransactionDefinition.PROPAGATION_NEVER) {
            throw new IllegalTransactionStateException(
                    "Existing transaction found for transaction marked with propagation 'never'");
        }

        // PROPAGATION_NOT_SUPPORTED       ，       ，        
        if (definition.getPropagationBehavior() == TransactionDefinition.PROPAGATION_NOT_SUPPORTED) {
            if (debugEnabled) {
                logger.debug("Suspending current transaction");
            }
            //       
            Object suspendedResources = suspend(transaction);
            boolean newSynchronization = (getTransactionSynchronization() == SYNCHRONIZATION_ALWAYS);
            //         TransactionStatus
            return prepareTransactionStatus(
                    definition, null, false, newSynchronization, debugEnabled, suspendedResources);
        }

        // PROPAGATION_REQUIRES_NEW       ，       ，      
        if (definition.getPropagationBehavior() == TransactionDefinition.PROPAGATION_REQUIRES_NEW) {
            if (debugEnabled) {
                logger.debug("Suspending current transaction, creating new transaction with name [" +
                        definition.getName() + "]");
            }
            SuspendedResourcesHolder suspendedResources = suspend(transaction);
            try {
                return startTransaction(definition, transaction, debugEnabled, suspendedResources);
            }
            catch (RuntimeException | Error beginEx) {
                resumeAfterBeginException(transaction, suspendedResources, beginEx);
                throw beginEx;
            }
        }

        // PROPAGATION_NESTED     ，          
        if (definition.getPropagationBehavior() == TransactionDefinition.PROPAGATION_NESTED) {
            if (!isNestedTransactionAllowed()) {
                throw new NestedTransactionNotSupportedException(
                        "Transaction manager does not allow nested transactions by default - " +
                        "specify 'nestedTransactionAllowed' property with value 'true'");
            }
            if (debugEnabled) {
                logger.debug("Creating nested transaction with name [" + definition.getName() + "]");
            }
            //  JTA         savePoint       
            // savePoint：                ，          
            if (useSavepointForNestedTransaction()) {
                // Create savepoint within existing Spring-managed transaction,
                // through the SavepointManager API implemented by TransactionStatus.
                // Usually uses JDBC 3.0 savepoints. Never activates Spring synchronization.
                DefaultTransactionStatus status =
                        prepareTransactionStatus(definition, transaction, false, false, debugEnabled, null);
                //      
                status.createAndHoldSavepoint();
                return status;
            }
            else {
                // Nested transaction through nested begin and commit/rollback calls.
                // Usually only for JTA: Spring synchronization might get activated here
                // in case of a pre-existing JTA transaction.
                return startTransaction(definition, transaction, debugEnabled, null);
            }
        }

        //                ，PROPAGATION_SUPPORTS    PROPAGATION_REQUIRED
        // Assumably PROPAGATION_SUPPORTS or PROPAGATION_REQUIRED.
        if (debugEnabled) {
            logger.debug("Participating in existing transaction");
        }
        
        //                          
        if (isValidateExistingTransaction()) {
            if (definition.getIsolationLevel() != TransactionDefinition.ISOLATION_DEFAULT) {
                Integer currentIsolationLevel = TransactionSynchronizationManager.getCurrentTransactionIsolationLevel();
                if (currentIsolationLevel == null || currentIsolationLevel != definition.getIsolationLevel()) {
                    Constants isoConstants = DefaultTransactionDefinition.constants;
                    throw new IllegalTransactionStateException("Participating transaction with definition [" +
                            definition + "] specifies isolation level which is incompatible with existing transaction: " +
                            (currentIsolationLevel != null ?
                                    isoConstants.toCode(currentIsolationLevel, DefaultTransactionDefinition.PREFIX_ISOLATION) :
                                    "(unknown)"));
                }
            }
            if (!definition.isReadOnly()) {
                if (TransactionSynchronizationManager.isCurrentTransactionReadOnly()) {
                    throw new IllegalTransactionStateException("Participating transaction with definition [" +
                            definition + "] is not marked as read-only but existing transaction is");
                }
            }
        }
        boolean newSynchronization = (getTransactionSynchronization() != SYNCHRONIZATION_NEVER);
        //     TransactionStatus，      
        return prepareTransactionStatus(definition, transaction, false, newSynchronization, debugEnabled, null);
    }

ここはコードが多いので、論理的に上記の注釈を見ればいいです.ここでやっと待ちに待った保留中のトランザクションとインライントランザクションが見えてきました.DataSourceTransactionManagerの実装を見てみましょう.

保留中のトランザクション:TransactionSynchronizationManager::bindResourceでデータ・ソースに基づいて現在の接続を取得し、resourceから削除します.その後、この接続はTransactionStatusオブジェクトの

に格納されます.

    // DataSourceTransactionManager::doSuspend
    @Override
    protected Object doSuspend(Object transaction) {
        DataSourceTransactionObject txObject = (DataSourceTransactionObject) transaction;
        txObject.setConnectionHolder(null);
        return TransactionSynchronizationManager.unbindResource(obtainDataSource());
    }

トランザクションのコミットまたはロールバック後、TransactionSynchronizationManager::unbindResourceが呼び出されると、TransactionStatusでキャッシュされたConnectionがresourceに再バインドされます.

埋め込みトランザクション:リレーショナル・データベースのsavePointによって実装され、コミットまたはロールバック時に、現在のトランザクションがsavePointである場合、savePointを解放するか、savePointにロールバックするかを判断します.具体的なロジックは、AbstractPlatformTransactionManager::cleanupAfterCompletionおよびAbstractPlatformTransactionManager::processRollback

を参照してください.
これで、トランザクションの伝播ソース分析が終了します.
prepareTransactionInfo
前に問題を残しました.prepareTransactionInfo法は何をしましたか.まずAbstractPlatformTransactionManager::processCommitの構造を見てみましょう.

    protected static final class TransactionInfo {

        @Nullable
        private final PlatformTransactionManager transactionManager;

        @Nullable
        private final TransactionAttribute transactionAttribute;

        private final String joinpointIdentification;

        @Nullable
        private TransactionStatus transactionStatus;

        @Nullable
        private TransactionInfo oldTransactionInfo;
        
        // ...
    }

Springでのクラスの役割は、オブジェクトを内部に渡すためです.ThreadLocalには最新のTransactionInfoが格納されており、現在のTransactionInfoでoldTransactionInfoを見つけることができます.トランザクションが作成されるたびに新しいTransactionInfo(実際のトランザクションが作成されるかどうかにかかわらず)がThreadLocalに格納されます.トランザクションが終了するたびに、現在のThreadLocalのTransactionInfoがoldTransactionInfoにリセットされ、Springトランザクションが論理的に無限にネストされるようにチェーンテーブルが形成されます.
もし収穫があると思ったら、私の公衆番号に注目して、あなたの称賛と関心は私に対する最大の支持です.

iOS 9をターゲットにしたBabel + webpackビルドはminifyに注意が必要

node.js + babel + webpack でCUIプログラミングしたかった