java 事务解释。
面试的时候，面试人员总喜欢问在spring中，
1. 如果一个主线程上有一个事务，在事务中开启了一个线程。子线程跑出异常，对主线程有没有影响，或者主线程产生异常对子线程有没有影响。
这个时候，你只要记住主线程和子线程是不同的事务即可回答上面的问题，主线程跑出异常肯定对子线程没有影响，但是子线程抛出异常对主线程有没有影响，那就要看，抛出的异常是否为事务回滚有需要的异常类型，
如果是肯定会回滚，如果不是就不会回滚。
2.主线程为啥和子线程是不同的事务（在所有的事务传播机制中）
因为 spring 中使用的ThreadLocal变量来保存 事务的执行者，entityManager。而ThreadLocal只能保证在当前线程中获取entityManager。所以主子线程肯定是不同的事务。
JpaTransactionObject txObject = new JpaTransactionObject(); txObject.setSavepointAllowed(isNestedTransactionAllowed()); //事务中的EntityManager 是从当前线程中获取 即ThreadLocal EntityManagerHolder emHolder = (EntityManagerHolder) TransactionSynchronizationManager.getResource(obtainEntityManagerFactory()); if (emHolder != null) { if (logger.isDebugEnabled()) { logger.debug("Found thread-bound EntityManager [" + emHolder.getEntityManager() + "] for JPA transaction"); } txObject.setEntityManagerHolder(emHolder, false); } if (getDataSource() != null) {
//事务中connectionHolder也是从threadLocal中获取。 ConnectionHolder conHolder = (ConnectionHolder) TransactionSynchronizationManager.getResource(getDataSource()); txObject.setConnectionHolder(conHolder); }
3.事务的同步
提到同步，首先想到的肯定是多线程。多线程肯定属于不同的事务，事务的同步就是解决多线程之间事务的同步问题。
@Override
@Transactional(propagation = Propagation.REQUIRED)
public void saveRequire(Customer customer) throws Exception {
//1.保存客户信息
repository.save(customer);
exectorService.execute(()->{
//2.发送短信或邮件
});
这个例子是 1.保存客户信息，2.发送短信或邮件--由于短信和邮件比较耗时，所以用异步进行操作。 要求：必须在保存客户信息成功后，发送短信和邮件。
上面例子的问题是，发送短信和邮件，执行时，可能在事务完成后执行，也可能在事务完成之前执行。
事务的同步：就是事务完成后，同时执行发送短信和邮件。
因此上面的例子可以更改为
@Transactional(propagation = Propagation.REQUIRED)
public void saveRequire(Customer customer) throws Exception {
//1.保存客户信息
repository.save(customer);
//判断当前事务是否**
if (TransactionSynchronizationManager.isActualTransactionActive()) {
TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronization() {
@Override
public void afterCommit() {
exectorService.execute(()->{
//2.发送短信或邮件
});
}
});
}else {
exectorService.execute(()->{
//2.发送短信或邮件
});
}
4.事务中的主要对象介绍。
TransactionInfo 事务的对象。
protected final class TransactionInfo { //事务管理器（获取事务） @Nullable private final PlatformTransactionManager transactionManager; //事务属性 （@Transactional(propagation = Propagation.REQUIRED)对应的这个注解的属性） @Nullable private final TransactionAttribute transactionAttribute; //标记@Transactional的类和方法组成的字符串（com.zhou.test.transaction_test.service.impl.User1ServiceImpl.saveRequire） private final String joinpointIdentification; //当前事务的状态（是否是新的事务，是否完成，是否有保存点，当前事务对象，当前被暂停的资源） @Nullable private TransactionStatus transactionStatus; @Nullable
//被暂停的事务（例如require_new 需要一个新的事务，老的事务会被挂起） private TransactionInfo oldTransactionInfo;
JpaTransactionObject 事务对象 （保存在 transactionStatus里面）
private class JpaTransactionObject extends JdbcTransactionObjectSupport { //jpa 操作数据库对象 @Nullable private EntityManagerHolder entityManagerHolder; private boolean newEntityManagerHolder; //可能是jpaTransactionObject（事务暂停） 可能是SavepointManager（netesd，同一个事务回滚到保存点） @Nullable private Object transactionData;
JpaTransactionManager.SuspendedResourcesHolder 暂停的对象
private static class SuspendedResourcesHolder { //数据源操作对象 private final EntityManagerHolder entityManagerHolder; //数据源操作对象 @Nullable private final ConnectionHolder connectionHolder;
TransactionSynchronization 事务的同步类 实现下面的方法，可以在事务执行对应操作时，增加对应的处理。
default void suspend() { } default void resume() { } @Override default void flush() { } default void beforeCommit(boolean readOnly) { } default void beforeCompletion() { } default void afterCommit() { } default void afterCompletion(int status) { }
AbstractPlatformTransactionManager.SuspendedResourcesHolder 暂停的资源
protected static class SuspendedResourcesHolder {
//被暂停的对象，（存放 connectionHolder，entityManagerHoler）
@Nullable
private final Object suspendedResources;
//存放当前事务同步的事件
@Nullable
private List<TransactionSynchronization> suspendedSynchronizations;
DefaultTransactionStatus 事务状态类。
public class DefaultTransactionStatus extends AbstractTransactionStatus { //当前事务对象（JPATransactionObject）。 @Nullable private final Object transaction; private final boolean newTransaction; private final boolean newSynchronization; private final boolean readOnly; private final boolean debug; //暂停的资源（AbstractPlatformTransactionManager.SuspendedResourcesHolder） @Nullable private final Object suspendedResources;
5.nested 中的保存点
依赖的connection 中保存点实现。回滚时，回滚到保存点。
6.事务的回滚原理
protected void completeTransactionAfterThrowing(@Nullable TransactionInfo txInfo, Throwable ex) { if (txInfo != null && txInfo.getTransactionStatus() != null) { if (logger.isTraceEnabled()) { logger.trace("Completing transaction for [" + txInfo.getJoinpointIdentification() + "] after exception: " + ex); }
//根据注释transactionAttribute的rollbackFor属性判断此种异常是否回滚，如果找不到需要回滚的指定异常，就根据是否是父类中的RuntimeException和Error进行回滚 if (txInfo.transactionAttribute != null && txInfo.transactionAttribute.rollbackOn(ex)) { try { txInfo.getTransactionManager().rollback(txInfo.getTransactionStatus()); } catch (TransactionSystemException ex2) { logger.error("Application exception overridden by rollback exception", ex); ex2.initApplicationException(ex); throw ex2; } catch (RuntimeException | Error ex2) { logger.error("Application exception overridden by rollback exception", ex); throw ex2; } }
private void processRollback(DefaultTransactionStatus status, boolean unexpected) { try { boolean unexpectedRollback = unexpected; try { triggerBeforeCompletion(status); if (status.hasSavepoint()) { if (status.isDebug()) { logger.debug("Rolling back transaction to savepoint"); }
//1.回滚的事务的保存点 status.rollbackToHeldSavepoint(); } else if (status.isNewTransaction()) { if (status.isDebug()) { logger.debug("Initiating transaction rollback"); }
//2.如果是新建的事务，直接回滚 doRollback(status); } else { // Participating in larger transaction if (status.hasTransaction()) { if (status.isLocalRollbackOnly() || isGlobalRollbackOnParticipationFailure()) { if (status.isDebug()) { logger.debug("Participating transaction failed - marking existing transaction as rollback-only"); }
//3.如果用的已经存在的事务，则标记事务为回滚，等回到主事务中进行回滚 doSetRollbackOnly(status); } else { if (status.isDebug()) { logger.debug("Participating transaction failed - letting transaction originator decide on rollback"); } } } else { logger.debug("Should roll back transaction but cannot - no transaction available"); } // Unexpected rollback only matters here if we're asked to fail early if (!isFailEarlyOnGlobalRollbackOnly()) { unexpectedRollback = false; } } }
事务的处理过程 ：TransactionInterceptor（只需关注事务拦截器即可。）
最后领着大家看下spring 的源码，事务的处理原理：
1.先熟悉下 JPA事务 可以去hibernatenate官网下载实例：https://github.com/hibernate/hibernate-demos
EntityManager entityManager = openEntityManager(); Session session = entityManager.unwrap(Session.class); session.beginTransaction(); Tool tool = new Tool(); tool.setName("Hammer111"); session.save(tool); Future<?> future= exectorService.submit(()->{ Tool tool1 = new Tool(); tool1.setName("Hammer222"); session.save(tool1); throw new RuntimeException(); }); future.get(); session.getTransaction().commit();
2. spring 中处理过程。
主要查看 JpaTransactionManager.getTransaction(@Nullable TransactionDefinition definition)
@Override public final TransactionStatus getTransaction(@Nullable TransactionDefinition definition) throws TransactionException {
//1.先获取一个事务对象 Object transaction = doGetTransaction();
protected Object doGetTransaction() { JpaTransactionObject txObject = new JpaTransactionObject(); txObject.setSavepointAllowed(isNestedTransactionAllowed()); //2.获取当前线程对应的EntityManagerHolder EntityManagerHolder emHolder = (EntityManagerHolder) TransactionSynchronizationManager.getResource(obtainEntityManagerFactory()); if (emHolder != null) { if (logger.isDebugEnabled()) { logger.debug("Found thread-bound EntityManager [" + emHolder.getEntityManager() + "] for JPA transaction"); } txObject.setEntityManagerHolder(emHolder, false); } if (getDataSource() != null) { ConnectionHolder conHolder = (ConnectionHolder) TransactionSynchronizationManager.getResource(getDataSource()); txObject.setConnectionHolder(conHolder); } return txObject; }
public final TransactionStatus getTransaction(@Nullable TransactionDefinition definition) throws TransactionException { Object transaction = doGetTransaction(); // Cache debug flag to avoid repeated checks. boolean debugEnabled = logger.isDebugEnabled(); if (definition == null) { // Use defaults if no transaction definition given. definition = new DefaultTransactionDefinition(); } //3.判断当前是否有一个事务，并且事务是开启的状态 if (isExistingTransaction(transaction)) { // Existing transaction found -> check propagation behavior to find out how to behave. return handleExistingTransaction(definition, transaction, debugEnabled); } // Check definition settings for new transaction. if (definition.getTimeout() < TransactionDefinition.TIMEOUT_DEFAULT) { throw new InvalidTimeoutException("Invalid transaction timeout", definition.getTimeout()); } // No existing transaction found -> check propagation behavior to find out how to proceed. if (definition.getPropagationBehavior() == TransactionDefinition.PROPAGATION_MANDATORY) { throw new IllegalTransactionStateException( "No existing transaction found for transaction marked with propagation 'mandatory'"); } else if (definition.getPropagationBehavior() == TransactionDefinition.PROPAGATION_REQUIRED || definition.getPropagationBehavior() == TransactionDefinition.PROPAGATION_REQUIRES_NEW || definition.getPropagationBehavior() == TransactionDefinition.PROPAGATION_NESTED) { SuspendedResourcesHolder suspendedResources = suspend(null); if (debugEnabled) { logger.debug("Creating new transaction with name [" + definition.getName() + "]: " + definition); } try { boolean newSynchronization = (getTransactionSynchronization() != SYNCHRONIZATION_NEVER); DefaultTransactionStatus status = newTransactionStatus( definition, transaction, true, newSynchronization, debugEnabled, suspendedResources);
//4.如果当前没有事务，就开启一个事务，并且把这个事务绑定到当前线程 doBegin(transaction, definition); prepareSynchronization(status, definition); return status; } catch (RuntimeException | Error ex) { resume(null, suspendedResources); throw ex; } } else { // Create "empty" transaction: no actual transaction, but potentially synchronization. if (definition.getIsolationLevel() != TransactionDefinition.ISOLATION_DEFAULT && logger.isWarnEnabled()) { logger.warn("Custom isolation level specified but no actual transaction initiated; " + "isolation level will effectively be ignored: " + definition); } boolean newSynchronization = (getTransactionSynchronization() == SYNCHRONIZATION_ALWAYS); return prepareTransactionStatus(definition, null, true, newSynchronization, debugEnabled, null); } }
5.在一个代理类的内部方法互相调用是不会添加切面方法的。（这个时候，如果想在内部方法互相调用时加入代理该怎么办？）
1.启用暴露代理对象属性
@EnableAspectJAutoProxy(exposeProxy=true)
2.内部互相调用时，写法修改。
((Service) AopContext.currentProxy()).callMethodB();
6. JDBC事务