spring 事务介绍与源码分析

2021-02-19 本文已影响0人逗逗罗

事务

事务定义

事务（Transaction）是数据库区别于文件系统的重要特性之一。MySQL 事务主要用于处理操作量大，复杂度高的数据。
1. 在 MySQL 数据库中只有 Innodb 数据库引擎的数据库或表才支持事务；
2. 事务是一个最小的不可再分的工作单元，可以用来保持数据库的完整性，保证成批的 DML(insert, update, delete) 语句, 要么全执行要么全不执行

事务的ACID特征

原子性

一个事务（transaction）中的所有操作，要么全部完成，要么全部不完成，不会结束在中间某个环节。事务在执行过程中发生错误，会被回滚（Rollback）到事务开始前的状态，就像这个事务从来没有执行过一样。

一致性

在事务开始之前和事务结束以后，数据库的完整性没有被破坏。这表示写入的资料必须完全符合所有的预设规则，这包含资料的精确度、串联性以及后续数据库可以自发性地完成预定的工作。

隔离性

数据库允许多个并发事务同时对其数据进行读写和修改的能力，隔离性可以防止多个事务并发执行时由于交叉执行而导致数据的不一致。事务隔离分为不同级别，包括读未提交（Read uncommitted）、读提交（read committed）、可重复读（repeatable read）和串行化（Serializable）。

持久性

事务处理结束后，对数据的修改就是永久的，即便系统故障也不会丢失。

事务的隔离级别

查看事务级别：
当前会话： select @@tx_isolation;
全局： select @@global.tx_isolation;

设置事务隔离级别：
READ UNCOMMITTED
READ COMMITTED
REPEATABLE READ
SERIALIZABLE

当前会话： set session transaction isolatin level repeatable read;
全局： set global transaction isolation level repeatable read;

读未提交（Read uncommitted）

1. 事务A未提交的数据，事务B可以读取到
2. 该隔离级别会引起 “脏读”

读提交（Read committed）

1. 事务A提交的数据，事务B可以读取到
2. 该隔离级别会导致 “不可重复读”
3. Oracle 默认的隔离级别

可重复读（repeatable read）

1. 事务A提交的数据，事务B读取不到
2. 该隔离级别会导致 “幻读”，“幻读” 是针对 insert 操作来说的
3. Mysql 默认的隔离级别

事务A	操作	事务B	操作
---	begin;	---	---
---	select * from table where id = 3; 查询结果为空	---	begin;
---	---	---	insert into table (id, value) values (3, 5);
---	---	---	commit;
---	insert into table (id, value) values (3, 5); 插入语句会出现异常	---	---

串行化（Serializable）

1. 事务A和事务B，事务A在操作数据库时，事务B只能排队等待

事务隔离的实现原理

spring事务

简单应用实例

编程式事务

TransactionTemplate

// 使用模板方法设计模式对原始事务管理方式进行了封装。
transactionTemplate.execute(new TransactionCallbackWithoutResult() {
    @Override
    protected void doInTransactionWithoutResult(TransactionStatus status) {
        update(1, "zhangsan-tx");
        insert(2, "lisi-tx", "345", "comment11");
    }
});

TransactionManager

DataSourceTransactionManager transactionManager = new DataSourceTransactionManager();
transactionManager.setDataSource(dataSource());
TransactionDefinition transactionDefinition = new DefaultTransactionDefinition();
TransactionStatus transactionStatus = transactionManager.getTransaction(transactionDefinition);
try {
    update(1, "zhangsan-tx");
    insert(2, "lisi-tx", "345", "comment11");
    transactionManager.commit(transactionStatus);
} catch (Exception e) {
    System.out.println(e.getMessage());
    transactionManager.rollback(transactionStatus);
}

TransactionManager
spring本身不支持底层的事务实现，只是负责包装底层事务，应用底层支持什么样的事务策略，spring就支持什么样的事务策略，例如：
1. JDBC事务：DataSourceTransactionManager
  
  images
2. Hibernate事务：HibernateTransactionManager
3. JPA事务：JpaTransactionManager
4. Java原生API事务：JtaTransactionManager，分布式事务处理

        - TransactionStatus：事务具体运行状态
            ![images](https://img.haomeiwen.com/i8024044/8972551d870e7572?imageMogr2/auto-orient/strip%7CimageView2/2/w/1240)


        - TransactionDefinition：事务定义信息(隔离，传播，超时，只读)
            ![images](https://img.haomeiwen.com/i8024044/dfc40fe836faaa82?imageMogr2/auto-orient/strip%7CimageView2/2/w/1240)

+ 声明式事务
    - TransactionProxyFactoryBean
        为每个需要事务支持的 Bean 创建代理类
        ```java
        @Bean(name = "daoProxy")
        public TransactionProxyFactoryBean daoTemplate(DeclareDaoService declareDaoService) {
            TransactionProxyFactoryBean transactionProxyFactoryBean = new TransactionProxyFactoryBean();
            transactionProxyFactoryBean.setTransactionManager(transactionManager());
            transactionProxyFactoryBean.setTarget(declareDaoService);
            Properties properties = new Properties();
            properties.setProperty("*", "PROPAGATION_REQUIRED");
            transactionProxyFactoryBean.setTransactionAttributes(properties);
            return transactionProxyFactoryBean;
        }
        ```

    - TransactionInterceptor
        ```java
        @Bean("transactionInterceptor")
        public TransactionInterceptor transactionInterceptor() {
            TransactionInterceptor transactionInterceptor = new TransactionInterceptor();
            transactionInterceptor.setTransactionManager(transactionManager());
            Properties properties = new Properties();
            properties.setProperty("*", "PROPAGATION_REQUIRED");
            transactionInterceptor.setTransactionAttributes(properties);
            return transactionInterceptor;
        }

        @Bean
        public BeanNameAutoProxyCreator beanNameAutoProxyCreator(){
            BeanNameAutoProxyCreator beanNameAutoProxyCreator = new BeanNameAutoProxyCreator();
            beanNameAutoProxyCreator.setBeanNames("*Service");
            beanNameAutoProxyCreator.setInterceptorNames("transactionInterceptor");
            return beanNameAutoProxyCreator;
        }
        ```

    - xml 配置文件 tx:advice 
        ```xml
        <!-- 定义事务管理器（声明式的事务） -->  
        <bean id="transactionManager"
            class="org.springframework.orm.hibernate3.HibernateTransactionManager">
            <property name="sessionFactory" ref="sessionFactory" />
        </bean>

        <!-- 
            通过对源码的分析 tx 标签的解析流程：
            XmlBeanDefinitionReader
                DefaultBeanDefinitionDocumentReader
                    DefaultNamespaceHandlerResolver（META-INF/spring.handlers）
                        TxNamespaceHandler
                            TxAdviceBeanDefinitionParser
                            AnnotationDrivenBeanDefinitionParser
                            JtaTransactionManagerBeanDefinitionParser
            TxAdviceBeanDefinitionParser -> 最终注册的 BeanDefinition 为 TransactionInterceptor              
        -->
        <tx:advice id="txAdvice" transaction-manager="transactionManager">
            <tx:attributes>
                <tx:method name="*" propagation="REQUIRED" />
            </tx:attributes>
        </tx:advice>
        
        <aop:config>
            <aop:pointcut id="interceptorPointCuts"
                expression="execution(* *..*(..))" />
            <aop:advisor advice-ref="txAdvice"
                pointcut-ref="interceptorPointCuts" />        
        </aop:config> 
        ```

    - 注解的方式
        ```java
        @EnableTransactionManagement
        public class TxConfig {
        }

        @Service
        @Transactional
        public class DeclareCglibDaoService {
        }

        ```

spring事务的两种实现方式
- 基于java动态代码的方式
- cgLib的方式

源码分析

声明式事务基于 XML 配置的实现方式

<!-- 
    通过对源码的分析 tx 标签的解析流程：
    XmlBeanDefinitionReader
        DefaultBeanDefinitionDocumentReader
            DefaultNamespaceHandlerResolver（META-INF/spring.handlers）
                TxNamespaceHandler
                    TxAdviceBeanDefinitionParser
                    AnnotationDrivenBeanDefinitionParser
                    JtaTransactionManagerBeanDefinitionParser
    TxAdviceBeanDefinitionParser -> 最终注册的 BeanDefinition 为 TransactionInterceptor

    aop:config 标签解析
        1. 生成一个后置处理器 AspectJAwareAdvisorAutoProxyCreator
        2. 解析 advisor 标签： 生成 DefaultBeanFactoryPointcutAdvisor        
-->

TxAdviceBeanDefinitionParser
TxAdviceBeanDefinitionParser 最终会生成 TransactionInterceptor 的 BeanDefinition

class TxAdviceBeanDefinitionParser extends AbstractSingleBeanDefinitionParser {
    
    // ...
    @Override
    protected Class<?> getBeanClass(Element element) {
        return TransactionInterceptor.class;
    }

    // 根据 tx:attributes 子标签创建 TransactionAttributeSource 事务属性解析器
    protected void doParse(Element element, ParserContext parserContext, BeanDefinitionBuilder builder) {
        // 1. 解析 transactionManager 属性
        builder.addPropertyReference("transactionManager", TxNamespaceHandler.getTransactionManagerName(element));
        // 2. 获取 attributes 子标签
        List<Element> txAttributes = DomUtils.getChildElementsByTagName(element, ATTRIBUTES_ELEMENT);
        if (txAttributes.size() > 1) {
            parserContext.getReaderContext().error(
                    "Element <attributes> is allowed at most once inside element <advice>", element);
        }
        else if (txAttributes.size() == 1) {
            // Using attributes source.
            Element attributeSourceElement = txAttributes.get(0);
            // 3. 解析 attribute 子标签，注册属性解析器 NameMatchTransactionAttributeSource
            RootBeanDefinition attributeSourceDefinition = parseAttributeSource(attributeSourceElement, parserContext);
            builder.addPropertyValue("transactionAttributeSource", attributeSourceDefinition);
        }
        else {
            // Assume annotations source.
            // 4. 注入 @Transaction 属性解析器 AnnotationTransactionAttributeSource
            builder.addPropertyValue("transactionAttributeSource",
                    new RootBeanDefinition("org.springframework.transaction.annotation.AnnotationTransactionAttributeSource"));
        }
    }
    // ...
}

TransactionInterceptor

images

TransactionInterceptor 实现了 MethodInterceptor 接口重写了 invoke 方法，根据 Spring AOP 的源码分析可知，MethodInterceptor 实现了 Advice 接口最终会同 pointcut 一起生成一个 Advisor 在动态代理中用于方法的增强。

public class TransactionInterceptor extends TransactionAspectSupport implements MethodInterceptor, Serializable {
    // ...
    @Override
    @Nullable
    public Object invoke(MethodInvocation invocation) throws Throwable {
        // Work out the target class: may be {@code null}.
        // The TransactionAttributeSource should be passed the target class
        // as well as the method, which may be from an interface.
        Class<?> targetClass = (invocation.getThis() != null ? AopUtils.getTargetClass(invocation.getThis()) : null);

        // Adapt to TransactionAspectSupport's invokeWithinTransaction...
        return invokeWithinTransaction(invocation.getMethod(), targetClass, invocation::proceed);
    }
    // ...
    
    protected Object invokeWithinTransaction(Method method, @Nullable Class<?> targetClass,
    final InvocationCallback invocation) throws Throwable {

        // If the transaction attribute is null, the method is non-transactional.
        // 1. 获取事务属性解析器
        TransactionAttributeSource tas = getTransactionAttributeSource();
        final TransactionAttribute txAttr = (tas != null ? tas.getTransactionAttribute(method, targetClass) : null);
        // 2. 决定事务管理器
        final PlatformTransactionManager tm = determineTransactionManager(txAttr);
        // 3. 确定方法连接点的唯一标识
        final String joinpointIdentification = methodIdentification(method, targetClass, txAttr);

        if (txAttr == null || !(tm instanceof CallbackPreferringPlatformTransactionManager)) {
            // Standard transaction demarcation with getTransaction and commit/rollback calls.
            // 2. 根据情况决定创建事物
            TransactionInfo txInfo = createTransactionIfNecessary(tm, txAttr, joinpointIdentification);
            Object retVal = null;
            try {
                // This is an around advice: Invoke the next interceptor in the chain.
                // This will normally result in a target object being invoked.
                // 3. 调用目标方法和拦截器链
                retVal = invocation.proceedWithInvocation();
            }
            catch (Throwable ex) {
                // target invocation exception
                // 4. 异常回滚
                completeTransactionAfterThrowing(txInfo, ex);
                throw ex;
            }
            finally {
                // 5. 保证清理工作
                cleanupTransactionInfo(txInfo);
            }
            // 6. 事物提交操作
            commitTransactionAfterReturning(txInfo);
            return retVal;
        }

        else {
            final ThrowableHolder throwableHolder = new ThrowableHolder();

            // It's a CallbackPreferringPlatformTransactionManager: pass a TransactionCallback in.
            try {
                Object result = ((CallbackPreferringPlatformTransactionManager) tm).execute(txAttr, status -> {
                    TransactionInfo txInfo = prepareTransactionInfo(tm, txAttr, joinpointIdentification, status);
                    try {
                        return invocation.proceedWithInvocation();
                    }
                    catch (Throwable ex) {
                        if (txAttr.rollbackOn(ex)) {
                            // A RuntimeException: will lead to a rollback.
                            if (ex instanceof RuntimeException) {
                                throw (RuntimeException) ex;
                            }
                            else {
                                throw new ThrowableHolderException(ex);
                            }
                        }
                        else {
                            // A normal return value: will lead to a commit.
                            throwableHolder.throwable = ex;
                            return null;
                        }
                    }
                    finally {
                        cleanupTransactionInfo(txInfo);
                    }
                });

                // Check result state: It might indicate a Throwable to rethrow.
                if (throwableHolder.throwable != null) {
                    throw throwableHolder.throwable;
                }
                return result;
            }
            catch (ThrowableHolderException ex) {
                throw ex.getCause();
            }
            catch (TransactionSystemException ex2) {
                if (throwableHolder.throwable != null) {
                    logger.error("Application exception overridden by commit exception", throwableHolder.throwable);
                    ex2.initApplicationException(throwableHolder.throwable);
                }
                throw ex2;
            }
            catch (Throwable ex2) {
                if (throwableHolder.throwable != null) {
                    logger.error("Application exception overridden by commit exception", throwableHolder.throwable);
                }
                throw ex2;
            }
        }
    }
}

声明式事务基于注解的实现方式

EnableTransactionManagerment

引入 TransactionManagementConfigurationSelector

@Import(TransactionManagementConfigurationSelector.class)
public @interface EnableTransactionManagement {

    boolean proxyTargetClass() default false;

    AdviceMode mode() default AdviceMode.PROXY;

    int order() default Ordered.LOWEST_PRECEDENCE;

}

TransactionManagementConfigurationSelector
默认加载两个类：AutoProxyRegistrar 和 ProxyTransactionManagementConfiguration

public class TransactionManagementConfigurationSelector extends AdviceModeImportSelector<EnableTransactionManagement> {
    @Override
    protected String[] selectImports(AdviceMode adviceMode) {
        switch (adviceMode) {
            case PROXY:
                // 默认加载两个类：AutoProxyRegistrar 和  ProxyTransactionManagementConfiguration
                return new String[] {AutoProxyRegistrar.class.getName(),
                        ProxyTransactionManagementConfiguration.class.getName()};
            case ASPECTJ:
                return new String[] {determineTransactionAspectClass()};
            default:
                return null;
        }
    }

    private String determineTransactionAspectClass() {
        return (ClassUtils.isPresent("javax.transaction.Transactional", getClass().getClassLoader()) ?
                TransactionManagementConfigUtils.JTA_TRANSACTION_ASPECT_CONFIGURATION_CLASS_NAME :
                TransactionManagementConfigUtils.TRANSACTION_ASPECT_CONFIGURATION_CLASS_NAME);
    }

}

AutoProxyRegistrar
默认注册 InfrastructureAdvisorAutoProxyCreator

public class AutoProxyRegistrar implements ImportBeanDefinitionRegistrar {

    private final Log logger = LogFactory.getLog(getClass());
    @Override
    public void registerBeanDefinitions(AnnotationMetadata importingClassMetadata, BeanDefinitionRegistry registry) {
        boolean candidateFound = false;
        Set<String> annTypes = importingClassMetadata.getAnnotationTypes();
        for (String annType : annTypes) {
            AnnotationAttributes candidate = AnnotationConfigUtils.attributesFor(importingClassMetadata, annType);
            if (candidate == null) {
                continue;
            }
            Object mode = candidate.get("mode");
            Object proxyTargetClass = candidate.get("proxyTargetClass");
            if (mode != null && proxyTargetClass != null && AdviceMode.class == mode.getClass() &&
                    Boolean.class == proxyTargetClass.getClass()) {
                candidateFound = true;
                if (mode == AdviceMode.PROXY) {
                    // 默认加载 InfrastructureAdvisorAutoProxyCreator 类
                    AopConfigUtils.registerAutoProxyCreatorIfNecessary(registry);
                    if ((Boolean) proxyTargetClass) {
                        AopConfigUtils.forceAutoProxyCreatorToUseClassProxying(registry);
                        return;
                    }
                }
            }
        }
        if (!candidateFound && logger.isInfoEnabled()) {
            String name = getClass().getSimpleName();
            logger.info(String.format("%s was imported but no annotations were found " +
                    "having both 'mode' and 'proxyTargetClass' attributes of type " +
                    "AdviceMode and boolean respectively. This means that auto proxy " +
                    "creator registration and configuration may not have occurred as " +
                    "intended, and components may not be proxied as expected. Check to " +
                    "ensure that %s has been @Import'ed on the same class where these " +
                    "annotations are declared; otherwise remove the import of %s " +
                    "altogether.", name, name, name));
        }
    }

}

AbstractTransactionManagementConfiguration
1. //注册事务增强器BeanFactoryTransactionAttributeSourceAdvisor
2. //注册事务属性解析器 AnnotationTransactionAttributeSource，解析注解中的各种属性，比如propagation、isolation、timeout等
3. //注册了TransactionInterceptor事务拦截器：保存了事务属性信息、事务管理器

@Configuration(proxyBeanMethods = false)
@Role(BeanDefinition.ROLE_INFRASTRUCTURE)
public class ProxyTransactionManagementConfiguration extends AbstractTransactionManagementConfiguration {
    // 1. 注册事务增强器BeanFactoryTransactionAttributeSourceAdvisor
    @Bean(name = TransactionManagementConfigUtils.TRANSACTION_ADVISOR_BEAN_NAME)
    @Role(BeanDefinition.ROLE_INFRASTRUCTURE)
    public BeanFactoryTransactionAttributeSourceAdvisor transactionAdvisor(
            TransactionAttributeSource transactionAttributeSource, TransactionInterceptor transactionInterceptor) {

        BeanFactoryTransactionAttributeSourceAdvisor advisor = new BeanFactoryTransactionAttributeSourceAdvisor();
        advisor.setTransactionAttributeSource(transactionAttributeSource);
        advisor.setAdvice(transactionInterceptor);
        if (this.enableTx != null) {
            advisor.setOrder(this.enableTx.<Integer>getNumber("order"));
        }
        return advisor;
    }

    // 2. 注册事务属性解析器，解析注解中的各种属性，比如propagation、isolation、timeout等
    @Bean
    @Role(BeanDefinition.ROLE_INFRASTRUCTURE)
    public TransactionAttributeSource transactionAttributeSource() {
        return new AnnotationTransactionAttributeSource();
    }

    // 3. 注册了TransactionInterceptor事务拦截器：保存了事务属性信息、事务管理器
    @Bean
    @Role(BeanDefinition.ROLE_INFRASTRUCTURE)
    public TransactionInterceptor transactionInterceptor(TransactionAttributeSource transactionAttributeSource) {
        TransactionInterceptor interceptor = new TransactionInterceptor();
        interceptor.setTransactionAttributeSource(transactionAttributeSource);
        if (this.txManager != null) {
            interceptor.setTransactionManager(this.txManager);
        }
        return interceptor;
    }

}

AnnotationTransactionAttributeSource 事务注解解析器

public class AnnotationTransactionAttributeSource extends AbstractFallbackTransactionAttributeSource
        implements Serializable {

    public AnnotationTransactionAttributeSource(boolean publicMethodsOnly) {
        this.publicMethodsOnly = publicMethodsOnly;
        if (jta12Present || ejb3Present) {
            this.annotationParsers = new LinkedHashSet<>(4);
            // 创建 Transaction 注解解析
            this.annotationParsers.add(new SpringTransactionAnnotationParser());
            if (jta12Present) {
                this.annotationParsers.add(new JtaTransactionAnnotationParser());
            }
            if (ejb3Present) {
                this.annotationParsers.add(new Ejb3TransactionAnnotationParser());
            }
        }
        else {
            this.annotationParsers = Collections.singleton(new SpringTransactionAnnotationParser());
        }
    }
}

SpringTransactionAnnotationParser, @Transaction 注解解析器

public class SpringTransactionAnnotationParser implements TransactionAnnotationParser, Serializable {

    @Override
    public boolean isCandidateClass(Class<?> targetClass) {
        return AnnotationUtils.isCandidateClass(targetClass, Transactional.class);
    }

    @Override
    @Nullable
    public TransactionAttribute parseTransactionAnnotation(AnnotatedElement element) {
        AnnotationAttributes attributes = AnnotatedElementUtils.findMergedAnnotationAttributes(
                element, Transactional.class, false, false);
        if (attributes != null) {
            return parseTransactionAnnotation(attributes);
        }
        else {
            return null;
        }
    }

    public TransactionAttribute parseTransactionAnnotation(Transactional ann) {
        return parseTransactionAnnotation(AnnotationUtils.getAnnotationAttributes(ann, false, false));
    }

    protected TransactionAttribute parseTransactionAnnotation(AnnotationAttributes attributes) {
        RuleBasedTransactionAttribute rbta = new RuleBasedTransactionAttribute();

        Propagation propagation = attributes.getEnum("propagation");
        rbta.setPropagationBehavior(propagation.value());
        Isolation isolation = attributes.getEnum("isolation");
        rbta.setIsolationLevel(isolation.value());
        rbta.setTimeout(attributes.getNumber("timeout").intValue());
        rbta.setReadOnly(attributes.getBoolean("readOnly"));
        rbta.setQualifier(attributes.getString("value"));

        List<RollbackRuleAttribute> rollbackRules = new ArrayList<>();
        for (Class<?> rbRule : attributes.getClassArray("rollbackFor")) {
            rollbackRules.add(new RollbackRuleAttribute(rbRule));
        }
        for (String rbRule : attributes.getStringArray("rollbackForClassName")) {
            rollbackRules.add(new RollbackRuleAttribute(rbRule));
        }
        for (Class<?> rbRule : attributes.getClassArray("noRollbackFor")) {
            rollbackRules.add(new NoRollbackRuleAttribute(rbRule));
        }
        for (String rbRule : attributes.getStringArray("noRollbackForClassName")) {
            rollbackRules.add(new NoRollbackRuleAttribute(rbRule));
        }
        rbta.setRollbackRules(rollbackRules);

        return rbta;
    }

}

spring 事务介绍与源码分析

事务

spring事务

猜你喜欢

热点阅读