spring项目中多数据源无法切换的原因分析

布满荆棘的人生 2022-03-16 15:05 391阅读 0赞

*最近公司人员变动比较大，一个个同事们离职后剩下的老员工只剩下几个了。  
正好遇到一个新功能，需要进行来开发。此种场景，新人又需要一定的时间去带去了解下业务后才能将功能开发下去，由于这个功能的开发权重还算是比较重，在讨论这个新功能开发时，为了能让它快速稳定开发完毕，作为还算比较熟悉业务的我便自告奋勇来开发这个新的功能了。*

在开发的过程中，由于要用到多数据源，并且还能根据业务场景来切换数据源，然后看到之前项目里有一个现成的代码，我便直接用它来做多数据源的切换功能。  
但当代码开发完毕，进行测试时，确发现无法进行数据源的切换，每次都是用的默认的数据源来进行处理的，当时研究了下，知道了是由于spring的事务管理导致的，知道问题的所在后，为了项目需要，我便先将功能完成了，但一直觉得还是很有必要来进行总结记录一下的，今天正好不是太忙又赶上周六，择日不如撞日，就今天来再次记录下吧！

为此我特意写了一个demo来重新重现下问题.项目地址如下：  
[https://github.com/puhaiyang/JavaStu/tree/master/spring-transaction-manager][https_github.com_puhaiyang_JavaStu_tree_master_spring-transaction-manager]  
本文中实现多数据源动态切换的原理是采用继承AbstractRoutingDataSource类并重写

protected Object determineCurrentLookupKey()

方法来实现的，具体的实现步骤这里不再做记录了，百度上一搜有一大把的实现，搜出来的变来变去主要都是两种：

1.  写一个Holder通过改变ThreadLocal来直接修改数据源的名称
2.  写一个注解，再在aop里配置一下，需要切换时通过注解来完成

这两种实现其实都是大同小异的，知道其实现原理才是最重要的。  
以下为我探究在springBoot项目里多数据源无法切换的原因的主要分析过程记录，通过debug来跟踪源码一步步的记录：

其入口业务代码为：

@GetMapping(value = "student/{stuId}")
        public String hello(@PathVariable(value = "stuId") Integer stuId) {
    //        DatasourceChangeStrategy.changeDatabaseIfNecessary(stuId);
            List<Map<String, Object>> stuList = stuService.getStuInfo(stuId);
            logger.info("stuId:{} result:{}", stuId, stuList.toString());
            return stuList.toString();
        }

其demo的stuService.getStuInfo(stuId)代码为：  
//加入注解，开启事物

@Transactional
        @Override
        public List<Map<String, Object>> getStuInfo(Integer stuId) {
            DatasourceChangeStrategy.changeDatabaseIfNecessary(stuId);
            return stuDao.getStuInfoById(stuId);
        }
        
        public class DatasourceChangeStrategy {
    	    public static void changeDatabaseIfNecessary(Integer stuId) {
    	        if (stuId != null) {
    	            if (stuId.equals(1) || stuId.equals(2)) {
    	                DatabaseContextHolder.setDatabaseType(DatabaseType.mytestdb);
    	            } else {
    	                DatabaseContextHolder.setDatabaseType(DatabaseType.mytestdb2);
    	            }
    	        }
    	    }
    }

在controller里在调用service时会进入CglibAopProxy这个类的如下方法：

@Override
    		@Nullable
    		public Object intercept(Object proxy, Method method, Object[] args, MethodProxy methodProxy) throws Throwable {
    			Object oldProxy = null;
    			boolean setProxyContext = false;
    			Object target = null;
    			TargetSource targetSource = this.advised.getTargetSource();
    			try {
    				if (this.advised.exposeProxy) {
    					// Make invocation available if necessary.
    					oldProxy = AopContext.setCurrentProxy(proxy);
    					setProxyContext = true;
    				}
    				// Get as late as possible to minimize the time we "own" the target, in case it comes from a pool...
    				target = targetSource.getTarget();
    				Class<?> targetClass = (target != null ? target.getClass() : null);
    				List<Object> chain = this.advised.getInterceptorsAndDynamicInterceptionAdvice(method, targetClass);
    				Object retVal;
    				// Check whether we only have one InvokerInterceptor: that is,
    				// no real advice, but just reflective invocation of the target.
    				if (chain.isEmpty() && Modifier.isPublic(method.getModifiers())) {
    					// We can skip creating a MethodInvocation: just invoke the target directly.
    					// Note that the final invoker must be an InvokerInterceptor, so we know
    					// it does nothing but a reflective operation on the target, and no hot
    					// swapping or fancy proxying.
    					Object[] argsToUse = AopProxyUtils.adaptArgumentsIfNecessary(method, args);
    					retVal = methodProxy.invoke(target, argsToUse);
    				}
    				else {
    					// We need to create a method invocation...
    					retVal = new CglibMethodInvocation(proxy, target, method, args, targetClass, chain, methodProxy).proceed();
    				}
    				retVal = processReturnType(proxy, target, method, retVal);
    				return retVal;
    			}
    			finally {
    				if (target != null && !targetSource.isStatic()) {
    					targetSource.releaseTarget(target);
    				}
    				if (setProxyContext) {
    					// Restore old proxy.
    					AopContext.setCurrentProxy(oldProxy);
    				}
    			}
    		}

这个方法是spring的aop代码执行的方法，之所以会进入这个方法，是由于spring项目在初始化时指定的。  
这块的代码属于DynamicAdvisedInterceptor这个静态类的方法，而DynamicAdvisedInterceptor的初始化也是在spring容器加载过程中完成的。  
上面代码里与本文相关的代码先关注这一句：

List<Object> chain = this.advised.getInterceptorsAndDynamicInterceptionAdvice(method, targetClass);

它会先获取将要执行的service方法所拥有的方法拦截器，如果service方法上有@Transactional注解，则会返回一个TransactionInterceptor对象在chain中，这也就是责任链设计模式在此框架中的一个小应用  
如果将要执行的代码方法（也就是service）的chain不为空，则会执行

retVal = new CglibMethodInvocation(proxy, target, method, args, targetClass, chain, methodProxy).proceed();

方法，否则就会直接执行cgLib代理方法，然后再返回下对象就完成了。  
然后再跟踪下proceed()方法，它会拿着带有TransactionInterceptor的拦截器对service方法做什么呢？

@Override
    	@Nullable
    	public Object proceed() throws Throwable {
    		//	We start with an index of -1 and increment early.
    		if (this.currentInterceptorIndex == this.interceptorsAndDynamicMethodMatchers.size() - 1) {
    			return invokeJoinpoint();
    		}
    
    		Object interceptorOrInterceptionAdvice =
    				this.interceptorsAndDynamicMethodMatchers.get(++this.currentInterceptorIndex);
    		if (interceptorOrInterceptionAdvice instanceof InterceptorAndDynamicMethodMatcher) {
    			// Evaluate dynamic method matcher here: static part will already have
    			// been evaluated and found to match.
    			InterceptorAndDynamicMethodMatcher dm =
    					(InterceptorAndDynamicMethodMatcher) interceptorOrInterceptionAdvice;
    			Class<?> targetClass = (this.targetClass != null ? this.targetClass : this.method.getDeclaringClass());
    			if (dm.methodMatcher.matches(this.method, targetClass, this.arguments)) {
    				return dm.interceptor.invoke(this);
    			}
    			else {
    				// Dynamic matching failed.
    				// Skip this interceptor and invoke the next in the chain.
    				return proceed();
    			}
    		}
    		else {
    			// It's an interceptor, so we just invoke it: The pointcut will have
    			// been evaluated statically before this object was constructed.
    			return ((MethodInterceptor) interceptorOrInterceptionAdvice).invoke(this);
    		}
    	}

从代码里可以看出，它主要是对之前方法转入的chain进行遍历并进行处理  
这里主要跟踪下只有一个责任链的情况，也就是上面方法中的最后一行代码：

return ((MethodInterceptor) interceptorOrInterceptionAdvice).invoke(this);

从代码也可以看出，它是TransactionInterceptor的方法了  
它的具体实现代码为：

@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);
    	}

然后再继续关注invokeWithinTransaction方法：

/**
    	 * General delegate for around-advice-based subclasses, delegating to several other template
    	 * methods on this class. Able to handle {@link CallbackPreferringPlatformTransactionManager}
    	 * as well as regular {@link PlatformTransactionManager} implementations.
    	 * @param method the Method being invoked
    	 * @param targetClass the target class that we're invoking the method on
    	 * @param invocation the callback to use for proceeding with the target invocation
    	 * @return the return value of the method, if any
    	 * @throws Throwable propagated from the target invocation
    	 */
    	@Nullable
    	protected Object invokeWithinTransaction(Method method, @Nullable Class<?> targetClass,
    			final InvocationCallback invocation) throws Throwable {
    
    		// If the transaction attribute is null, the method is non-transactional.
    		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)) {
    			// Standard transaction demarcation with getTransaction and commit/rollback calls.
    			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.
    				retVal = invocation.proceedWithInvocation();
    			}
    			catch (Throwable ex) {
    				// target invocation exception
    				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);
    					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;
    			}
    		}
    	}

关注其中的这一行：

TransactionInfo txInfo = createTransactionIfNecessary(tm, txAttr, joinpointIdentification);

其源码为：

/**
    	 * Create a transaction if necessary based on the given TransactionAttribute.
    	 * <p>Allows callers to perform custom TransactionAttribute lookups through
    	 * the TransactionAttributeSource.
    	 * @param txAttr the TransactionAttribute (may be {@code null})
    	 * @param joinpointIdentification the fully qualified method name
    	 * (used for monitoring and logging purposes)
    	 * @return a TransactionInfo object, whether or not a transaction was created.
    	 * The {@code hasTransaction()} method on TransactionInfo can be used to
    	 * tell if there was a transaction created.
    	 * @see #getTransactionAttributeSource()
    	 */
    	@SuppressWarnings("serial")
    	protected TransactionInfo createTransactionIfNecessary(@Nullable PlatformTransactionManager tm,
    			@Nullable TransactionAttribute txAttr, final String joinpointIdentification) {
    
    		// If no name specified, apply method identification as transaction name.
    		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) {
    				status = tm.getTransaction(txAttr);
    			}
    			else {
    				if (logger.isDebugEnabled()) {
    					logger.debug("Skipping transactional joinpoint [" + joinpointIdentification +
    							"] because no transaction manager has been configured");
    				}
    			}
    		}
    		return prepareTransactionInfo(tm, txAttr, joinpointIdentification, status);
    	}

查看代码可知道，它将会根据方法的事务属性来判断是否要需开启一个事务，这就是所谓的spring的事务传播行为在springTransaction中的具体实现了  
再进入getTransaction方法：

//---------------------------------------------------------------------
    	// Implementation of PlatformTransactionManager
    	//---------------------------------------------------------------------
    
    	/**
    	 * This implementation handles propagation behavior. Delegates to
    	 * {@code doGetTransaction}, {@code isExistingTransaction}
    	 * and {@code doBegin}.
    	 * @see #doGetTransaction
    	 * @see #isExistingTransaction
    	 * @see #doBegin
    	 */
    	@Override
    	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();
    		}
    
    		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);
    				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);
    		}
    	}

上面的代码是开启spring事务的代码，然后继续跟踪上面代码里的doBegin(transaction, definition)方法

/**
    	 * This implementation sets the isolation level but ignores the timeout.
    	 */
    	@Override
    	protected void doBegin(Object transaction, TransactionDefinition definition) {
    		DataSourceTransactionObject txObject = (DataSourceTransactionObject) transaction;
    		Connection con = null;
    
    		try {
    			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);
    			}
    
    			txObject.getConnectionHolder().setSynchronizedWithTransaction(true);
    			con = txObject.getConnectionHolder().getConnection();
    
    			Integer previousIsolationLevel = DataSourceUtils.prepareConnectionForTransaction(con, definition);
    			txObject.setPreviousIsolationLevel(previousIsolationLevel);
    
    			// 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()) {
    				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);
    		}
    	}

在上面的代码中就可以看到本文的关键代码了：

Connection newCon = obtainDataSource().getConnection();

通过这个方法的名称便知道它从确定的数据源中获取到了一个Connection.  
跟踪到这里，就能完全知道本文的标题的答案了：

@Override
    	public Connection getConnection() throws SQLException {
    		return determineTargetDataSource().getConnection();
    	}

这个方法的确定目标数据源方法（determineTargetDataSource）代码为：

/**
    	 * Retrieve the current target DataSource. Determines the
    	 * {@link #determineCurrentLookupKey() current lookup key}, performs
    	 * a lookup in the {@link #setTargetDataSources targetDataSources} map,
    	 * falls back to the specified
    	 * {@link #setDefaultTargetDataSource default target DataSource} if necessary.
    	 * @see #determineCurrentLookupKey()
    	 */
    	protected DataSource determineTargetDataSource() {
    		Assert.notNull(this.resolvedDataSources, "DataSource router not initialized");
    		Object lookupKey = determineCurrentLookupKey();
    		DataSource dataSource = this.resolvedDataSources.get(lookupKey);
    		if (dataSource == null && (this.lenientFallback || lookupKey == null)) {
    			dataSource = this.resolvedDefaultDataSource;
    		}
    		if (dataSource == null) {
    			throw new IllegalStateException("Cannot determine target DataSource for lookup key [" + lookupKey + "]");
    		}
    		return dataSource;
    	}

看到这里的**determineCurrentLookupKey**方法的时候一切都能明了了。

那么再回到本文的题目，spring项目中多数据源无法切换，为什么呢？  
因为我在代码中对于动态数据源的切换代码写在了service方法内了，而那个service的方法头上又加上了@Transactional注解，那么在spring框架执行的过程中，它先执行TransactionInterceptor的方法，而后再去执行我所写的service的具体代码，而在TransactionInterceptor执行的过程中它会先获取到数据源的连接，正因为如此，那么切换无论我在service里怎么写切换的数据源的代码它都不会再去执行了。  
So？那么这个要怎么解决呢？我暂时也没有想到好的方法，当时我便用一个比较笨的办法来解决的，在dao里写了一个策略模式，然后在策略模式里通过不同的策略上下文注入不同的dataSource来进行查询的。  
如果业务代码能进行改变的话，当然也可以考虑如下的两种方式处理：

1.  需要切换数据源的service代码不要开启事务
2.  将切换数据源的代码写在service层之前，即写在controller层里

[https_github.com_puhaiyang_JavaStu_tree_master_spring-transaction-manager]: https://github.com/puhaiyang/JavaStu/tree/master/spring-transaction-manager