Spring源码事务管理：事务级操作监听回调

Posted 2021-09-19 boonya

目录

核心源码类

定义事务管理的生命周期

事务生命周期适配器

本地化线程隔离机制

实现事务管理

---

spring事务管理是基于本地化线程进行处理的，将事务提交线程本地化做到事务的线程级别的隔离。

核心源码类

• TransactionSynchronization

• TransactionSynchronizationAdapter

• TransactionSynchronizationManager

定义事务管理的生命周期

核心生命周期：事务提交前、提交后、以及事务完成等过程。

public interface TransactionSynchronization extends Flushable {
    int STATUS_COMMITTED = 0;
    int STATUS_ROLLED_BACK = 1;
    int STATUS_UNKNOWN = 2;

    void suspend();

    void resume();

    void flush();

    void beforeCommit(boolean var1);

    void beforeCompletion();

    void afterCommit();

    void afterCompletion(int var1);
}

事务生命周期适配器

该类是TransactionSynchronization和Orderd接口的实现。

public abstract class TransactionSynchronizationAdapter implements TransactionSynchronization, Ordered {
    public TransactionSynchronizationAdapter() {
    }

    public int getOrder() {
        return 2147483647;
    }

    public void suspend() {
    }

    public void resume() {
    }

    public void flush() {
    }

    public void beforeCommit(boolean readOnly) {
    }

    public void beforeCompletion() {
    }

    public void afterCommit() {
    }

    public void afterCompletion(int status) {
    }
}

本地化线程隔离机制

在TransactionSynchronizationManager类中定义了一系列本地化变量。

 private static final Log logger = LogFactory.getLog(TransactionSynchronizationManager.class);
    private static final ThreadLocal<Map<Object, Object>> resources = new NamedThreadLocal("Transactional resources");
    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");
    private static final ThreadLocal<Boolean> actualTransactionActive = new NamedThreadLocal("Actual transaction active");

实现事务管理

 

public abstract class TransactionSynchronizationManager {
    private static final Log logger = LogFactory.getLog(TransactionSynchronizationManager.class);
    private static final ThreadLocal<Map<Object, Object>> resources = new NamedThreadLocal("Transactional resources");
    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");
    private static final ThreadLocal<Boolean> actualTransactionActive = new NamedThreadLocal("Actual transaction active");

    public TransactionSynchronizationManager() {
    }

    public static Map<Object, Object> getResourceMap() {
        Map<Object, Object> map = (Map)resources.get();
        return map != null ? Collections.unmodifiableMap(map) : Collections.emptyMap();
    }

    public static boolean hasResource(Object key) {
        Object actualKey = TransactionSynchronizationUtils.unwrapResourceIfNecessary(key);
        Object value = doGetResource(actualKey);
        return value != null;
    }

    public static Object getResource(Object 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;
    }

    private static Object doGetResource(Object actualKey) {
        Map<Object, Object> map = (Map)resources.get();
        if (map == null) {
            return null;
        } else {
            Object value = map.get(actualKey);
            if (value instanceof ResourceHolder && ((ResourceHolder)value).isVoid()) {
                map.remove(actualKey);
                if (map.isEmpty()) {
                    resources.remove();
                }

                value = null;
            }

            return value;
        }
    }

    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 = (Map)resources.get();
        if (map == null) {
            map = new HashMap();
            resources.set(map);
        }

        Object oldValue = ((Map)map).put(actualKey, value);
        if (oldValue instanceof ResourceHolder && ((ResourceHolder)oldValue).isVoid()) {
            oldValue = null;
        }

        if (oldValue != null) {
            throw new IllegalStateException("Already value [" + oldValue + "] for key [" + actualKey + "] bound to thread [" + Thread.currentThread().getName() + "]");
        } else {
            if (logger.isTraceEnabled()) {
                logger.trace("Bound value [" + value + "] for key [" + actualKey + "] to thread [" + Thread.currentThread().getName() + "]");
            }

        }
    }

    public static Object unbindResource(Object key) throws IllegalStateException {
        Object actualKey = TransactionSynchronizationUtils.unwrapResourceIfNecessary(key);
        Object value = doUnbindResource(actualKey);
        if (value == null) {
            throw new IllegalStateException("No value for key [" + actualKey + "] bound to thread [" + Thread.currentThread().getName() + "]");
        } else {
            return value;
        }
    }

    public static Object unbindResourceIfPossible(Object key) {
        Object actualKey = TransactionSynchronizationUtils.unwrapResourceIfNecessary(key);
        return doUnbindResource(actualKey);
    }

    private static Object doUnbindResource(Object actualKey) {
        Map<Object, Object> map = (Map)resources.get();
        if (map == null) {
            return null;
        } else {
            Object value = map.remove(actualKey);
            if (map.isEmpty()) {
                resources.remove();
            }

            if (value instanceof ResourceHolder && ((ResourceHolder)value).isVoid()) {
                value = null;
            }

            if (value != null && logger.isTraceEnabled()) {
                logger.trace("Removed value [" + value + "] for key [" + actualKey + "] from thread [" + Thread.currentThread().getName() + "]");
            }

            return value;
        }
    }

    public static boolean isSynchronizationActive() {
        return synchronizations.get() != null;
    }

    public static void initSynchronization() throws IllegalStateException {
        if (isSynchronizationActive()) {
            throw new IllegalStateException("Cannot activate transaction synchronization - already active");
        } else {
            logger.trace("Initializing transaction synchronization");
            synchronizations.set(new LinkedHashSet());
        }
    }

    public static void registerSynchronization(TransactionSynchronization synchronization) throws IllegalStateException {
        Assert.notNull(synchronization, "TransactionSynchronization must not be null");
        if (!isSynchronizationActive()) {
            throw new IllegalStateException("Transaction synchronization is not active");
        } else {
            ((Set)synchronizations.get()).add(synchronization);
        }
    }

    public static List<TransactionSynchronization> getSynchronizations() throws IllegalStateException {
        Set<TransactionSynchronization> synchs = (Set)synchronizations.get();
        if (synchs == null) {
            throw new IllegalStateException("Transaction synchronization is not active");
        } else if (synchs.isEmpty()) {
            return Collections.emptyList();
        } else {
            List<TransactionSynchronization> sortedSynchs = new ArrayList(synchs);
            AnnotationAwareOrderComparator.sort(sortedSynchs);
            return Collections.unmodifiableList(sortedSynchs);
        }
    }

    public static void clearSynchronization() throws IllegalStateException {
        if (!isSynchronizationActive()) {
            throw new IllegalStateException("Cannot deactivate transaction synchronization - not active");
        } else {
            logger.trace("Clearing transaction synchronization");
            synchronizations.remove();
        }
    }

    public static void setCurrentTransactionName(String name) {
        currentTransactionName.set(name);
    }

    public static String getCurrentTransactionName() {
        return (String)currentTransactionName.get();
    }

    public static void setCurrentTransactionReadOnly(boolean readOnly) {
        currentTransactionReadOnly.set(readOnly ? Boolean.TRUE : null);
    }

    public static boolean isCurrentTransactionReadOnly() {
        return currentTransactionReadOnly.get() != null;
    }

    public static void setCurrentTransactionIsolationLevel(Integer isolationLevel) {
        currentTransactionIsolationLevel.set(isolationLevel);
    }

    public static Integer getCurrentTransactionIsolationLevel() {
        return (Integer)currentTransactionIsolationLevel.get();
    }

    public static void setActualTransactionActive(boolean active) {
        actualTransactionActive.set(active ? Boolean.TRUE : null);
    }

    public static boolean isActualTransactionActive() {
        return actualTransactionActive.get() != null;
    }

    public static void clear() {
        clearSynchronization();
        setCurrentTransactionName((String)null);
        setCurrentTransactionReadOnly(false);
        setCurrentTransactionIsolationLevel((Integer)null);
        setActualTransactionActive(false);
    }
}

更多参看：

https://blog.csdn.net/ly199108171231/article/details/92984574

https://www.jianshu.com/p/4b5eb29cc6d9