|
@Transactional源码分析
事务不生效的几种常见情况
-
方法为非public方法。 -
同一个类中的方法直接内部调用，会导致事务失效。 System.out.println("方法内部调用");
this.internMethod();
}
@Transactional(rollbackFor = Exception.class)
public void internMethod(){
User user = new User();
user.setId("1");
userMapper.insert(user);
int 1 = 1/0;
}
解决方法：
在方法内部获得当前类代理对象的方式,通过代理对象调用内部方法
AopContext.currentProxy()方式获取（注 ：需要Springboot启动类上增加@EnableAspectJAutoProxy(exposeProxy = true)的注解 ）
public void testTransactional() {
System.out.println("方法内部调用");
UserServiceImpl service =(UserServiceImpl) AopContext.currentProxy();
service.internMethod();
}
-
调用方法为异步方法，导致事务失效。 @Transactional(rollbackFor = Exception.class)
public void testTransactional() {
User user = new User();
user.setId("1");
userMapper.insert(user);
System.out.println("异步调用");
UserServiceImpl service =(UserServiceImpl) AopContext.currentProxy();
service.internMethod();
}
@Async
@Transactional(rollbackFor = Exception.class)
public void internMethod(){
User user = new User();
user.setId("2");
userMapper.insert(user);
int i = 1/0;
}
可以思考一下这种情况下，数据库的数据会是什么？
答：数据库中只会有id为1的一条数据。 -
类未被spring管理 -
Mysql存储引擎并发InnoDB
Transactional注解源码解析
Spring设计的几个接口定义
-
PlatformTransactionManager 事务管理器
public interface PlatformTransactionManager extends TransactionManager { // 从当前线程中获取绑定的ConnectionHolder，可能为null,如果为null，则会在下一个开启事务的过程中，
// 从dataSource中获取一个Connection，封装成ConnectionHolder，然后再绑定到当前线程
// 从当前线程中获取绑定的ConnectionHolder，可能为null,如果为null，则会在下一个开启事务的过程中，从dataSource中获取一个Connection，封装成ConnectionHolder，然后再绑定到当前线程
TransactionStatus getTransaction(@Nullable TransactionDefinition var1) throws TransactionException; // DataSourceTransactionManager就是使用DataSourceTransactionObject中的Connection来进行回滚操作
void commit(TransactionStatus var1) throws TransactionException;
//DataSourceTransactionManager依托内部的Connection来完成提交操作
void rollback(TransactionStatus var1) throws TransactionException;
-
AbstractPlatformTransactionManager
- DataSourceTransactionManager（重点分析）
- HibernateTransactionManager
- JpaTransactionManager
-
TransactionDefinition 事务定义 public interface TransactionDefinition {
//事务传播行为类型：如果当前没有事务，就新建一个事务，如果已经存在一个事务中，加入到这个事务中。
int PROPAGATION_REQUIRED = 0;
//事务传播行为类型：支持当前事务，如果当前没有事务，就以非事务方式执行。
int PROPAGATION_SUPPORTS = 1;
//事务传播行为类型：当前如果有事务，Spring就会使用该事务；否则会抛出异常
int PROPAGATION_MANDATORY = 2;
//事务传播行为类型：新建事务，如果当前存在事务，把当前事务挂起。
int PROPAGATION_REQUIRES_NEW = 3;
//事务传播行为类型：以非事务方式执行操作，如果当前存在事务，就把当前事务挂起。
int PROPAGATION_NOT_SUPPORTED = 4;
//事务传播行为类型：即使当前有事务，Spring也会在非事务环境下执行。如果当前有事务，则抛出异常
int PROPAGATION_NEVER = 5;
//事务传播行为类型：如果当前存在事务，则在嵌套事务内执行。
int PROPAGATION_NESTED = 6;
//隔离级别：默认的隔离级别（对mysql数据库来说就是ISOLATION_ READ_COMMITTED，可以重复读）
int ISOLATION_DEFAULT = -1;
//隔离级别：读未提交(最低)
int ISOLATION_READ_UNCOMMITTED = Connection.TRANSACTION_READ_UNCOMMITTED;
//隔离级别：读提交
int ISOLATION_READ_COMMITTED = Connection.TRANSACTION_READ_COMMITTED;
//隔离级别：可重复度
int ISOLATION_REPEATABLE_READ = Connection.TRANSACTION_REPEATABLE_READ;
//隔离级别：序列化操作（最高）
int ISOLATION_SERIALIZABLE = Connection.TRANSACTION_SERIALIZABLE;
//默认事务的超时时间
int TIMEOUT_DEFAULT = -1;
default int getPropagationBehavior() {
return 0;
}
default int getIsolationLevel() {
return -1;
}
default int getTimeout() {
return -1;
}
default boolean isReadOnly() {
return false;
}
@Nullable
default String getName() {
return null;
}
static TransactionDefinition withDefaults() {
return StaticTransactionDefinition.INSTANCE;
}
TransactionAttribute是TransactionDefinition的实现接口 -
TransactionStatus 事务状态 public class DefaultTransactionStatus extends AbstractTransactionStatus {
@Nullable
private final Object transaction;
private final boolean newTransaction;
private final boolean newSynchronization;
private final boolean readOnly;
private final boolean debug;
@Nullable
private final Object suspendedResources;
}
DefaultTransactionStatus的获取
DefaultTransactionStatus的获取是事务管理器的方法，这里的实现是AbstractPlatformTransactionManager提供的模板方法
在获取Object transaction之后，先进行判断，是否是已存在的事务。因为这个Object transaction的获取过程就是直接从线程绑定的获取的,可能当前线程已经存在事务 -
TransactionInfo 事务信息 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 TransactionAspectSupport.TransactionInfo oldTransactionInfo;
源码分析
-
TransactionInterceptor.invoke() @Nullable
public Object invoke(MethodInvocation invocation) throws Throwable {
Class<?> targetClass = invocation.getThis() != null ? AopUtils.getTargetClass(invocation.getThis()) : null;
Method var10001 = invocation.getMethod();
invocation.getClass();
return this.invokeWithinTransaction(var10001, targetClass, invocation::proceed);
}
-
TransactionAspectSupport.invokeWithinTransaction()
重点在 this.createTransactionIfNecessary(ptm, txAttr, joinpointIdentification)*,它会判断是否存在事务，根据事务的传播属性。做出不同的处理，也是做了一层包装，核心是通过TransactionStatus来判断事务的属性。
通过持有的PlatformTransactionManager来获取TransactionStatus @Nullable
protected Object invokeWithinTransaction(Method method, @Nullable Class<?> targetClass, TransactionAspectSupport.InvocationCallback invocation) throws Throwable {
TransactionAttributeSource tas = this.getTransactionAttributeSource();
// 获取的事务相关属性
TransactionAttribute txAttr = tas != null ? tas.getTransactionAttribute(method, targetClass) : null;
// 获取 beanFactory 中的 transactionManage，确定事务管理器
TransactionManager tm = this.determineTransactionManager(txAttr);
// 响应式编程
if (this.reactiveAdapterRegistry != null && tm instanceof ReactiveTransactionManager) {
TransactionAspectSupport.ReactiveTransactionSupport txSupport = (TransactionAspectSupport.ReactiveTransactionSupport)this.transactionSupportCache.computeIfAbsent(method, (key) -> {
if (KotlinDetector.isKotlinType(method.getDeclaringClass()) && TransactionAspectSupport.KotlinDelegate.isSuspend(method)) {
throw new TransactionUsageException("Unsupported annotated transaction on suspending function detected: " + method + ". Use TransactionalOperator.transactional extensions instead.");
} else {
ReactiveAdapter adapter = this.reactiveAdapterRegistry.getAdapter(method.getReturnType());
if (adapter == null) {
throw new IllegalStateException("Cannot apply reactive transaction to non-reactive return type: " + method.getReturnType());
} else {
return new TransactionAspectSupport.ReactiveTransactionSupport(adapter);
}
}
});
return txSupport.invokeWithinTransaction(method, targetClass, invocation, txAttr, (ReactiveTransactionManager)tm);
} else {
PlatformTransactionManager ptm = this.asPlatformTransactionManager(tm);
// 获取需要被代理的方法
String joinpointIdentification = this.methodIdentification(method, targetClass, txAttr);
// 回调型事物管理对象: 事物管理对象可以是非数据库事物管理，如kafka事物. 在执行事物的过程中跟数据库很不一样，所有需要基础该类自己实现.
if (txAttr != null && ptm instanceof CallbackPreferringPlatformTransactionManager) {
TransactionAspectSupport.ThrowableHolder throwableHolder = new TransactionAspectSupport.ThrowableHolder();
Object result;
try {
result = ((CallbackPreferringPlatformTransactionManager)ptm).execute(txAttr, (statusx) -> {
TransactionAspectSupport.TransactionInfo txInfo = this.prepareTransactionInfo(ptm, txAttr, joinpointIdentification, statusx);
Object var9;
try {
Object retVal = invocation.proceedWithInvocation();
if (retVal != null && vavrPresent && TransactionAspectSupport.VavrDelegate.isVavrTry(retVal)) {
retVal = TransactionAspectSupport.VavrDelegate.evaluateTryFailure(retVal, txAttr, statusx);
}
var9 = retVal;
return var9;
} catch (Throwable var13) {
if (txAttr.rollbackOn(var13)) {
if (var13 instanceof RuntimeException) {
throw (RuntimeException)var13;
}
throw new TransactionAspectSupport.ThrowableHolderException(var13);
}
throwableHolder.throwable = var13;
var9 = null;
} finally {
this.cleanupTransactionInfo(txInfo);
}
return var9;
});
} catch (TransactionAspectSupport.ThrowableHolderException var20) {
throw var20.getCause();
} catch (TransactionSystemException var21) {
if (throwableHolder.throwable != null) {
this.logger.error("Application exception overridden by commit exception", throwableHolder.throwable);
var21.initApplicationException(throwableHolder.throwable);
}
throw var21;
} catch (Throwable var22) {
if (throwableHolder.throwable != null) {
this.logger.error("Application exception overridden by commit exception", throwableHolder.throwable);
}
throw var22;
}
if (throwableHolder.throwable != null) {
throw throwableHolder.throwable;
} else {
return result;
}
} else {
// 非回调型事物管理对象,如Mysql，我们重点关注的方法也是这个createTransactionIfNecessary()
TransactionAspectSupport.TransactionInfo txInfo = this.createTransactionIfNecessary(ptm, txAttr, joinpointIdentification);
Object retVal;
try {
// 目标方法执行
retVal = invocation.proceedWithInvocation();
} catch (Throwable var18) {
// 异常处理
this.completeTransactionAfterThrowing(txInfo, var18);
throw var18;
} finally {
// 清除事务信息
this.cleanupTransactionInfo(txInfo);
}
if (retVal != null && vavrPresent && TransactionAspectSupport.VavrDelegate.isVavrTry(retVal)) {
TransactionStatus status = txInfo.getTransactionStatus();
if (status != null && txAttr != null) {
retVal = TransactionAspectSupport.VavrDelegate.evaluateTryFailure(retVal, txAttr, status);
}
}
this.commitTransactionAfterReturning(txInfo);
return retVal;
}
}
-
TransactionAspectSupport.createTransactionIfNecessary() protected TransactionAspectSupport.TransactionInfo createTransactionIfNecessary(@Nullable PlatformTransactionManager tm, @Nullable TransactionAttribute txAttr, final String joinpointIdentification) {
if (txAttr != null && ((TransactionAttribute)txAttr).getName() == null) {
txAttr = new DelegatingTransactionAttribute((TransactionAttribute)txAttr) {
public String getName() {
return joinpointIdentification;
}
};
}
TransactionStatus status = null;
if (txAttr != null) {
if (tm != null) {
// 获取事务信息
status = tm.getTransaction((TransactionDefinition)txAttr);
} else if (this.logger.isDebugEnabled()) {
this.logger.debug("Skipping transactional joinpoint [" + joinpointIdentification + "] because no transaction manager has been configured");
}
}
return this.prepareTransactionInfo(tm, (TransactionAttribute)txAttr, joinpointIdentification, status);
}
-
AbstractPlatformTransactionManager.getTransaction() public final TransactionStatus getTransaction(@Nullable TransactionDefinition definition) throws TransactionException {
TransactionDefinition def = definition != null ? definition : TransactionDefinition.withDefaults();
Object transaction = this.doGetTransaction();
boolean debugEnabled = this.logger.isDebugEnabled();
// 如果当前线程存在事务
if (this.isExistingTransaction(transaction)) {
return this.handleExistingTransaction(def, transaction, debugEnabled);
// 验证事务超时的设置
} else if (def.getTimeout() < -1) {
throw new InvalidTimeoutException("Invalid transaction timeout", def.getTimeout());
// 判断事务传播机制
} else if (def.getPropagationBehavior() == 2) {
throw new IllegalTransactionStateException("No existing transaction found for transaction marked with propagation 'mandatory'");
} else if (def.getPropagationBehavior() != 0 && def.getPropagationBehavior() != 3 && def.getPropagationBehavior() != 6) {
if (def.getIsolationLevel() != -1 && this.logger.isWarnEnabled()) {
this.logger.warn("Custom isolation level specified but no actual transaction initiated; isolation level will effectively be ignored: " + def);
}
boolean newSynchronization = this.getTransactionSynchronization() == 0;
return this.prepareTransactionStatus(def, (Object)null, true, newSynchronization, debugEnabled, (Object)null);
} else {
// 对于DataSourceTransactionManager来说，事务的挂起，就是把当前线程关联的ConnectionHolder解除绑定、同理事务的恢复就是把上述ConnectionHolder再重新绑定到当前线程，继续执行该事务
AbstractPlatformTransactionManager.SuspendedResourcesHolder suspendedResources = this.suspend((Object)null);
if (debugEnabled) {
this.logger.debug("Creating new transaction with name [" + def.getName() + "]: " + def);
}
try {
// 如果Propagation为：ROPAGATION_REQUIRED,PROPAGATION_REQUIRES_NEW,PROPAGATION_NESTED中的一个将开启一个新事物，new一个新的DefaultTransactionStatus
// 创建事务
return this.startTransaction(def, transaction, debugEnabled, suspendedResources);
} catch (Error | RuntimeException var7) {
this.resume((Object)null, suspendedResources);
throw var7;
}
}
}
private TransactionStatus startTransaction(TransactionDefinition definition, Object transaction, boolean debugEnabled, @Nullable AbstractPlatformTransactionManager.SuspendedResourcesHolder suspendedResources) {
boolean newSynchronization = this.getTransactionSynchronization() != 2;
// newTransaction设置为true，这个状态很重要，因为后面的不论回滚、提交都是根据这个属性来判断是否在这个TransactionStatus上来进行。
DefaultTransactionStatus status = this.newTransactionStatus(definition, transaction, true, newSynchronization, debugEnabled, suspendedResources);
// 开启新事物
this.doBegin(transaction, definition);
this.prepareSynchronization(status, definition);
return status;
}
-
DataSourceTransactionManager.doBegin() protected void doBegin(Object transaction, TransactionDefinition definition) {
DataSourceTransactionObject txObject = (DataSourceTransactionObject) transaction;
Connection con = null;
try {
// 1、判断是否复用当前线程连接
if (!txObject.hasConnectionHolder() ||
txObject.getConnectionHolder().isSynchronizedWithTransaction()) {
Connection newCon = obtainDataSource().getConnection();
if (logger.isDebugEnabled()) {
logger.debug("Acquired Connection [" + newCon + "] for JDBC transaction");
}
txObject.setConnectionHolder(new ConnectionHolder(newCon), true);
}
// 2、设置隔离级别
txObject.getConnectionHolder().setSynchronizedWithTransaction(true);
con = txObject.getConnectionHolder().getConnection();
Integer previousIsolationLevel = DataSourceUtils.prepareConnectionForTransaction(con, definition);
txObject.setPreviousIsolationLevel(previousIsolationLevel);
txObject.setReadOnly(definition.isReadOnly());
// Switch to manual commit if necessary. This is very expensive in some JDBC drivers,
// so we don't want to do it unnecessarily (for example if we've explicitly
// configured the connection pool to set it already).
if (con.getAutoCommit()) {
txObject.setMustRestoreAutoCommit(true);
if (logger.isDebugEnabled()) {
logger.debug("Switching JDBC Connection [" + con + "] to manual commit");
}
// 关闭自动提交
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()) {
// 绑定到当前线程中，这也是为什么异步调用会导致当前事务失效的原因,与数据库的连接保存在线程中
// 这里将当前的connection放入TransactionSynchronizationManager中持有，如果下次调用可以判断为已有的事务
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);
}
}
-
ConnectionImpl.setAutoCommit
此时可以看出transaction底层是使用了mysql中SET autocommit=1的命令！！！ this.session.getServerSession().setAutoCommit(autoCommitFlag);
if (needsSetOnServer) {
this.session.execSQL((Query)null, autoCommitFlag ? "SET autocommit=1" : "SET autocommit=0", -1, (NativePacketPayload)null, false, this.nullStatementResultSetFactory, (ColumnDefinition)null, false);
}
-
回到TransactionAspectSupport.createTransactionIfNecessary() protected TransactionAspectSupport.TransactionInfo prepareTransactionInfo(@Nullable PlatformTransactionManager tm, @Nullable TransactionAttribute txAttr, String joinpointIdentification, @Nullable TransactionStatus status) {
TransactionAspectSupport.TransactionInfo txInfo = new TransactionAspectSupport.TransactionInfo(tm, txAttr, joinpointIdentification);
if (txAttr != null) {
if (this.logger.isTraceEnabled()) {
this.logger.trace("Getting transaction for [" + txInfo.getJoinpointIdentification() + "]");
}
txInfo.newTransactionStatus(status);
} else if (this.logger.isTraceEnabled()) {
this.logger.trace("No need to create transaction for [" + joinpointIdentification + "]: This method is not transactional.");
}
// 绑定到当前线程，生成一个TransactionInfo并绑定到当前线程的ThreadLocal
txInfo.bindToThread();
return txInfo;
}
-
回到最初的invokeWithinTransaction()方法中 completeTransactionAfterThrowing() // 回滚
commitTransactionAfterReturning(txInfo); // commit
Spring手动开启事务
编程式事务原理与声明式事务原理相同
@Autowired
private DataSourceTransactionManager transactionManager;
public void testHandleTransactional() {
DefaultTransactionDefinition def = new DefaultTransactionDefinition();
def.setPropagationBehavior(TransactionDefinition.PROPAGATION_REQUIRES_NEW);
TransactionStatus status = transactionManager.getTransaction(def);
try {
User user = new User();
user.setId("1");
userMapper.insert(user);
}catch (Exception exception){
transactionManager.rollback(status);
}
transactionManager.commit(status);
}
Mysql中的事务
-
Mysql开启binlog没有未提交的事务的相关日志。 -
Mysql中开启事务的三种方法
- START TRANSACTION
- BEGIN
- SET AUTOCOMMIT = 1
但是在binlog日志中保存的命令语句均为 BEGIN
参考文章
- https://www.cnblogs.com/chihirotan/category/988426.html
- https://mp.weixin.qq.com/s/_9SVoba91OZmr-mJj8PnkA
|