Spring 源码学习(九) Transaction 事务

create table test.user(
id int auto_increment	
primary key,
name varchar(20) null, age int(3) null)
engine=InnoDB charset=utf8;

public class JdbcUser {

	private Integer id;

	private String name;

	private Integer age;
	
	...(使用 ctrl + N 进行代码补全 setter 和 getter)
}

public class UserRowMapper implements RowMapper {


	@Override
	public Object mapRow(ResultSet rs, int rowNum) throws SQLException {
		JdbcUser user = new JdbcUser();
		user.setId(rs.getInt("id"));
		user.setName(rs.getString("name"));
		user.setAge(rs.getInt("age"));
		return user;
	}
}

public interface UserDao {

	/**
	 * 插入
	 * @param user	用户信息
	 */
	void insertUser(JdbcUser user);

	/**
	 * 根据 id 进行删除
	 * @param id	主键
	 */
	void deleteById(Integer id);

	/**
	 * 查询
	 * @return	全部
	 */
	List<JdbcUser> selectAll();
}

public class UserJdbcTemplate implements UserDao {

	private DataSource dataSource;

	private JdbcTemplate jdbcTemplate;


	@Override
	@Transactional(propagation = Propagation.REQUIRED)
	public void insertUser(JdbcUser user) {
		String sql = "insert into user (id, name, age) values (?, ?, ?)";
		jdbcTemplate.update(sql, user.getId(), user.getName(), user.getAge());
		System.out.println("Create record : " + user.toString());
	}

	@Override
	@Transactional
	public void deleteById(Integer id) {
		String sql = "delete from user where id = ?";
		jdbcTemplate.update(sql, id);
		System.out.println("Delete record, id = " + id);
		// 事务测试，抛出异常，让上面的插入操作回滚
		throw new RuntimeException("aa");
	}


	@Override
	public List<JdbcUser> selectAll() {
		String sql = "select * from user";
		List<JdbcUser> users = jdbcTemplate.query(sql, new UserRowMapper());
		return users;
	}



	public void setDataSource(DataSource dataSource) {
		// 使用 setter 注入参数时，同时初始化 jdbcTemplate
		this.dataSource = dataSource;
		this.jdbcTemplate = new JdbcTemplate(dataSource);
	}

}

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
	   xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
	   xmlns:tx="http://www.springframework.org/schema/tx"
	   xsi:schemaLocation="http://www.springframework.org/schema/beans
	   http://www.springframework.org/schema/beans/spring-beans.xsd
	   http://www.springframework.org/schema/tx
	   http://www.springframework.org/schema/tx/spring-tx.xsd">

	<!-- 数据源 MySQL -->
	<bean id="dataSource" class="org.springframework.jdbc.datasource.DriverManagerDataSource">
		<property name="driverClassName" value="com.mysql.jdbc.Driver"/>
		<property name="url" value="jdbc:mysql://localhost:3306/test?characterEncoding=utf8"/>
		<property name="username" value="root"/>
		<property name="password" value="12345678"/>
	</bean>

	<bean id="userJdbcTemplate" class="transaction.UserJdbcTemplate">
		<property name="dataSource" ref="dataSource"/>
	</bean>

	<!-- 开启事务，如果将这行去掉，将不会创建事务  -->
	<tx:annotation-driven/>
	<bean id="transactionManager" class="org.springframework.jdbc.datasource.DataSourceTransactionManager">
		<property name="dataSource" ref="dataSource"/>
	</bean>
</beans>

optional(project(":spring-jdbc"))  // for Quartz support
optional(project(":spring-tx"))  // for Quartz support
compile group: 'mysql', name: 'mysql-connector-java', version: '5.1.6'

public class TransactionBootstrap {

	public static void main(String[] args) {
			ApplicationContext context = new ClassPathXmlApplicationContext("transaction/transaction.xml");
			UserJdbcTemplate jdbcTemplate = (UserJdbcTemplate) context.getBean("userJdbcTemplate");
			System.out.println("--- Records Creation start ----");
			JdbcUser user = new JdbcUser(4, "test", 21);
			jdbcTemplate.insertUser(user);
	}
}

@Override
public void init() {
	registerBeanDefinitionParser("advice", new TxAdviceBeanDefinitionParser());
	// 使用 AnnotationDrivenBeanDefinitionParser 解析器，解析 annotation-driven 标签
	registerBeanDefinitionParser("annotation-driven", new AnnotationDrivenBeanDefinitionParser());
	registerBeanDefinitionParser("jta-transaction-manager", new JtaTransactionManagerBeanDefinitionParser());
}

public BeanDefinition parse(Element element, ParserContext parserContext) {
	registerTransactionalEventListenerFactory(parserContext);
	String mode = element.getAttribute("mode");
	// AspectJ 另外处理
	if ("aspectj".equals(mode)) {
		// mode="aspectj"
		registerTransactionAspect(element, parserContext);
		if (ClassUtils.isPresent("javax.transaction.Transactional", getClass().getClassLoader())) {
			registerJtaTransactionAspect(element, parserContext);
		}
	}
	else {
		// mode="proxy"
		AopAutoProxyConfigurer.configureAutoProxyCreator(element, parserContext);
	}
	return null;
}

<tx:annotation-driven mode="aspectj"/>

private static class AopAutoProxyConfigurer {
	public static void configureAutoProxyCreator(Element element, ParserContext parserContext) {
		// 注册 InfrastructureAdvisorAutoProxyCreator 自动创建代理器
		AopNamespaceUtils.registerAutoProxyCreatorIfNecessary(parserContext, element);
		// txAdvisorBeanName = org.springframework.transaction.config.internalTransactionAdvisor
		String txAdvisorBeanName = TransactionManagementConfigUtils.TRANSACTION_ADVISOR_BEAN_NAME;
		if (!parserContext.getRegistry().containsBeanDefinition(txAdvisorBeanName)) {
			Object eleSource = parserContext.extractSource(element);
			// Create the TransactionAttributeSource definition.
			// 创建 TransactionAttributeSource 的 bean
			RootBeanDefinition sourceDef = new RootBeanDefinition(
					"org.springframework.transaction.annotation.AnnotationTransactionAttributeSource");
			// 注册 bean，并使用 Spring 中的定义规则生成 beanName
			String sourceName = parserContext.getReaderContext().registerWithGeneratedName(sourceDef);
			// 创建 TransactionInterceptor 的 bean
			RootBeanDefinition interceptorDef = new RootBeanDefinition(TransactionInterceptor.class);
			interceptorDef.getPropertyValues().add("transactionAttributeSource", new RuntimeBeanReference(sourceName));
			String interceptorName = parserContext.getReaderContext().registerWithGeneratedName(interceptorDef);
			// 创建 TransactionAttributeSourceAdvisor 的 bean
			RootBeanDefinition advisorDef = new RootBeanDefinition(BeanFactoryTransactionAttributeSourceAdvisor.class);
			// 将 sourceName 的 bean 注入 advisor 的 transactionAttributeSource 属性中
			advisorDef.getPropertyValues().add("transactionAttributeSource", new RuntimeBeanReference(sourceName));
			// 将 interceptorName 的 bean 注入到 advisor 的 adviceBeanName 属性中
			advisorDef.getPropertyValues().add("adviceBeanName", interceptorName);
			if (element.hasAttribute("order")) {
				// 如果配置了 order 属性，则加入到 bean 中
				advisorDef.getPropertyValues().add("order", element.getAttribute("order"));
			}
			// 以 txAdvisorBeanName 名字注册 advisorDef
			parserContext.getRegistry().registerBeanDefinition(txAdvisorBeanName, advisorDef);
			// 创建 CompositeComponentDefinition
			CompositeComponentDefinition compositeDef = new CompositeComponentDefinition(element.getTagName(), eleSource);
			compositeDef.addNestedComponent(new BeanComponentDefinition(sourceDef, sourceName));
			compositeDef.addNestedComponent(new BeanComponentDefinition(interceptorDef, interceptorName));
			compositeDef.addNestedComponent(new BeanComponentDefinition(advisorDef, txAdvisorBeanName));
			parserContext.registerComponent(compositeDef);
		}
	}
}

public static void registerAutoProxyCreatorIfNecessary(
		ParserContext parserContext, Element sourceElement) {
	BeanDefinition beanDefinition = AopConfigUtils.registerAutoProxyCreatorIfNecessary(
			parserContext.getRegistry(), parserContext.extractSource(sourceElement));
	useClassProxyingIfNecessary(parserContext.getRegistry(), sourceElement);
	registerComponentIfNecessary(beanDefinition, parserContext);
}

private static void registerComponentIfNecessary(@Nullable BeanDefinition beanDefinition, ParserContext parserContext) {
	if (beanDefinition != null) {
	    // 注册的 beanName 是 org.springframework.aop.config.internalAutoProxyCreator
		parserContext.registerComponent(
				new BeanComponentDefinition(beanDefinition, AopConfigUtils.AUTO_PROXY_CREATOR_BEAN_NAME));
	}
}

public Object postProcessAfterInitialization(@Nullable Object bean, String beanName) {
	if (bean != null) {
		// 组装 key
		Object cacheKey = getCacheKey(bean.getClass(), beanName);
		if (this.earlyProxyReferences.remove(cacheKey) != bean) {
			// 如果适合被代理，则需要封装指定的 bean
			return wrapIfNecessary(bean, beanName, cacheKey);
		}
	}
	return bean;
}

public static boolean canApply(Advisor advisor, Class<?> targetClass, boolean hasIntroductions) {
	if (advisor instanceof IntroductionAdvisor) {
		return ((IntroductionAdvisor) advisor).getClassFilter().matches(targetClass);
	}
	else if (advisor instanceof PointcutAdvisor) {
		PointcutAdvisor pca = (PointcutAdvisor) advisor;
		return canApply(pca.getPointcut(), targetClass, hasIntroductions);
	}
	else {
		// It doesn't have a pointcut so we assume it applies.
		return true;
	}
}

private final TransactionAttributeSourcePointcut pointcut = new TransactionAttributeSourcePointcut() {
	@Override
	@Nullable
	protected TransactionAttributeSource getTransactionAttributeSource() {
		// 实现父类的方法，在子类中进行了扩展，返回之前在标签注册时的AnnotationTransactionAttributeSource
		return transactionAttributeSource;
	}
};

@Override
public boolean matches(Method method, Class<?> targetClass) {
	// 事务切点匹配的方法
	TransactionAttributeSource tas = getTransactionAttributeSource();
	return (tas == null || tas.getTransactionAttribute(method, targetClass) != null);
}

protected TransactionAttribute determineTransactionAttribute(AnnotatedElement element) {
	for (TransactionAnnotationParser parser : this.annotationParsers) {
		TransactionAttribute attr = parser.parseTransactionAnnotation(element);
		if (attr != null) {
			return attr;
		}
	}
	return null;
}

public TransactionAttribute parseTransactionAnnotation(AnnotatedElement element) {
	// 解析事务注解的属性
	AnnotationAttributes attributes = AnnotatedElementUtils.findMergedAnnotationAttributes(
			element, Transactional.class, false, false);
	if (attributes != null) {
		return parseTransactionAnnotation(attributes);
	}
	else {
		return null;
	}
}

protected TransactionAttribute parseTransactionAnnotation(AnnotationAttributes attributes) {
	RuleBasedTransactionAttribute rbta = new RuleBasedTransactionAttribute();
	// 注释 9.4 解析事务注解的每一个属性
	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;
}

public Object invoke(MethodInvocation invocation) throws Throwable {
	// 注释 9.5 执行事务拦截器，完成整个事务的逻辑
	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 {
	// 如果transaction属性为null，则该方法是非事务性的
	TransactionAttributeSource tas = getTransactionAttributeSource();
	// 获取对应事务属性
	final TransactionAttribute txAttr = (tas != null ? tas.getTransactionAttribute(method, targetClass) : null);
	// 获取事务管理器
	final PlatformTransactionManager tm = determineTransactionManager(txAttr);
	// 构造方法唯一标识（类.方法）
	final String joinpointIdentification = methodIdentification(method, targetClass, txAttr);
	if (txAttr == null || !(tm instanceof CallbackPreferringPlatformTransactionManager)) {
		// 声明式事务处理
		// 标准事务划分 : 使用 getTransaction 和 commit / rollback 调用
		TransactionInfo txInfo = createTransactionIfNecessary(tm, txAttr, joinpointIdentification);
		Object retVal;
		try {
			//传入的是回调函数对象： invocation.proceed。 执行被增强的方法
			retVal = invocation.proceedWithInvocation();
		}
		catch (Throwable ex) {
			// 异常回滚
			completeTransactionAfterThrowing(txInfo, ex);
			throw ex;
		}
		finally {
			// 清除信息
			cleanupTransactionInfo(txInfo);
		}
		// 提交事务
		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);
			...
			return result;
		}
	}
}

protected PlatformTransactionManager determineTransactionManager(@Nullable TransactionAttribute txAttr) {
	// Do not attempt to lookup tx manager if no tx attributes are set
	// 注释 9.6 寻找事务管理器
	if (txAttr == null || this.beanFactory == null) {
		// 如果没有事务属性或者 BeanFactory 为空时，从缓存里面寻找
		return asPlatformTransactionManager(getTransactionManager());
	}

	String qualifier = txAttr.getQualifier();
	// 如果注解配置中指定了事务管理器，直接取出使用
	if (StringUtils.hasText(qualifier)) {
		return determineQualifiedTransactionManager(this.beanFactory, qualifier);
	}
	else if (StringUtils.hasText(this.transactionManagerBeanName)) {
		return determineQualifiedTransactionManager(this.beanFactory, this.transactionManagerBeanName);
	}
	else {
		// 上面步骤都没找到，最后才去容器中，根据 className 来寻找
		PlatformTransactionManager defaultTransactionManager = asPlatformTransactionManager(getTransactionManager());
		...
		return defaultTransactionManager;
	}
}

protected TransactionInfo createTransactionIfNecessary(PlatformTransactionManager tm, TransactionAttribute txAttr, final String joinpointIdentification) {
	// 如果没有名称指定则使用方法唯一标识，并使用 DelegatingTransactionAttribute 包装 txAttr
	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) {
			// 获取 TransactionStatus
			status = tm.getTransaction(txAttr);
		}
	}
	// 根据指定的属性与 status 准备一个 TransactionInfo
	return prepareTransactionInfo(tm, txAttr, joinpointIdentification, status);
}

protected Object doGetTransaction() {
	DataSourceTransactionObject txObject = new DataSourceTransactionObject();
	txObject.setSavepointAllowed(isNestedTransactionAllowed());
	// 如果当前线程已经记录数据库链接则使用原有链接
	ConnectionHolder conHolder =
			(ConnectionHolder) TransactionSynchronizationManager.getResource(obtainDataSource());
	// false 表示非新创建连接
	txObject.setConnectionHolder(conHolder, false);
	return txObject;
}

private static final ThreadLocal<Map<Object, Object>> resources =
			new NamedThreadLocal<>("Transactional resources");
			
private static Object doGetResource(Object actualKey) {
	Map<Object, Object> map = resources.get();
	if (map == null) {
		return null;
	}
	Object value = map.get(actualKey);
	// Transparently remove ResourceHolder that was marked as void...
	if (value instanceof ResourceHolder && ((ResourceHolder) value).isVoid()) {
		map.remove(actualKey);
		// Remove entire ThreadLocal if empty...
		if (map.isEmpty()) {
			resources.remove();
		}
		value = null;
	}
	return value;
}

if (isExistingTransaction(transaction)) {
	// Existing transaction found -> check propagation behavior to find out how to behave.
	// 当前线程存在事务，分情况进行处理
	return handleExistingTransaction(def, transaction, debugEnabled);
}

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

if (definition.getPropagationBehavior() == TransactionDefinition.PROPAGATION_NOT_SUPPORTED) {
	Object suspendedResources = suspend(transaction);
	boolean newSynchronization = (getTransactionSynchronization() == SYNCHRONIZATION_ALWAYS);
	return prepareTransactionStatus(
			definition, null, false, newSynchronization, debugEnabled, suspendedResources);
}

protected Object doSuspend(Object transaction) {
	DataSourceTransactionObject txObject = (DataSourceTransactionObject) transaction;
	//  将数据库连接设置为 null
	txObject.setConnectionHolder(null);
	return TransactionSynchronizationManager.unbindResource(obtainDataSource());
}

private static Object doUnbindResource(Object actualKey) {
	Map<Object, Object> map = resources.get();
	if (map == null) {
		return null;
	}
	Object value = map.remove(actualKey);
	if (map.isEmpty()) {
		resources.remove();
	}
	if (value instanceof ResourceHolder && ((ResourceHolder) value).isVoid()) {
		value = null;
	}
	if (value != null && logger.isTraceEnabled()) {
        Thread.currentThread().getName() + "]");
	}
	return value;
}

SuspendedResourcesHolder suspendedResources = suspend(transaction);
try {
	boolean newSynchronization = (getTransactionSynchronization() != SYNCHRONIZATION_NEVER);
	DefaultTransactionStatus status = newTransactionStatus(
			definition, transaction, true, newSynchronization, debugEnabled, suspendedResources);
	// 新事务的建立
	doBegin(transaction, definition);
	prepareSynchronization(status, definition);
	return status;
}
catch (RuntimeException | Error beginEx) {
	resumeAfterBeginException(transaction, suspendedResources, beginEx);
	throw beginEx;
}

if (definition.getPropagationBehavior() == TransactionDefinition.PROPAGATION_NESTED) {
	// 嵌入式事务的处理
	if (useSavepointForNestedTransaction()) {
		DefaultTransactionStatus status =
				prepareTransactionStatus(definition, transaction, false, false, debugEnabled, null);
		// 创建 savepoint
		status.createAndHoldSavepoint();
		return status;
	}
}

protected void doBegin(Object transaction, TransactionDefinition definition) {
	DataSourceTransactionObject txObject = (DataSourceTransactionObject) transaction;
	Connection con = null;
	if (!txObject.hasConnectionHolder() ||
			txObject.getConnectionHolder().isSynchronizedWithTransaction()) {
		Connection newCon = obtainDataSource().getConnection();

		txObject.setConnectionHolder(new ConnectionHolder(newCon), true);
	}

	txObject.getConnectionHolder().setSynchronizedWithTransaction(true);
	con = txObject.getConnectionHolder().getConnection();
	// 设置隔离级别
	Integer previousIsolationLevel = DataSourceUtils.prepareConnectionForTransaction(con, definition);
	txObject.setPreviousIsolationLevel(previousIsolationLevel);

	// configured the connection pool to set it already).
	// 更改自动提交设置，由 spring 进行控制
	if (con.getAutoCommit()) {
		txObject.setMustRestoreAutoCommit(true);
		con.setAutoCommit(false);
	}
	// 准备事务连接
	prepareTransactionalConnection(con, definition);
	// 设置判断当前线程是否存在事务的依据
	txObject.getConnectionHolder().setTransactionActive(true);

	int timeout = determineTimeout(definition);
	if (timeout != TransactionDefinition.TIMEOUT_DEFAULT) {
		txObject.getConnectionHolder().setTimeoutInSeconds(timeout);
	}

	// Bind the connection holder to the thread.
	if (txObject.isNewConnectionHolder()) {
		// 将当前获取到的连接绑定到当前线程
		TransactionSynchronizationManager.bindResource(obtainDataSource(), txObject.getConnectionHolder());
		}
	}
}