事务生效，但数据源切换失败。
1. 前言
问题：使用mybatis-plus-dynamicdatasource实现数据源切换操作功能，在遇到同时借助spring-tx进行事务处理时，会发现数据源无法正确地切换到辅库。
能点开这篇博客的基本都是奔着解决方案来的，所以以下我们先介绍解决方案和相应的最佳实践；最后再分析下原因，过下相关源代码。
2. 解决方案
以下直接以代码形式，同时展示问题场景和解决方案。
@Service
public class TestService {
// 主库
@Autowired
private MasterxMapper masterMapper;
// 辅库
@Autowired
private slaveMapper slaveMapper;
//启用事务注解
@Transactional
public void test(){
Console.log("main logic");
User user = new User();
user.setUuid(IdUtil.simpleUUID());
user.setXxx("12332112311");
// 主库相关操作
masterMapper.insert(user);
// 辅库操作, 注意这里的写法; 这是为了让Spring AOP生效
getService().slaveLook();
}
// 声明本方法内，以非事务方式运行，以实现数据源切换
@Transactional(propagation = Propagation.NOT_SUPPORTED)
// 声明本方法内, 我们将要操作的是副库
@DS("slave")
// 必须为public ;;; 参见 AnnotationTransactionAttributeSource.allowPublicMethodsOnly()
public void slaveLook(){
// 操作辅库
Console.log(slaveMapper.selectById(123));
// 如果抛出异常，整个调用链路中的主库操作将回滚，对应本例中是 masterMapper.insert(user);
//throw new RuntimeException();
}
private TestService getService(){
// 从Spring容器中取出本实例
return SpringUtil.getBean(this.getClass());
}
3. 最佳实践
-
再次提醒，副库并不会获得事务的支持。如果你需要同时支持辅库上的修改操作，请使用Atomikos等分布式事务。
-
@DS建议打在Service上，这是源自mybatis-plus-dynamicdatasource官方的建议。 -
上面的
getService()方法可以考虑抽取为一个接口，在Java8的默认接口方法实现的支持下，这将是个非常简单的事情。public interface SelfAop<T> { default T getSelf() { return ((T)SpringUtil.getBean(this.getClass())); // ParameterizedType currentType = Stream.of(getClass().getGenericInterfaces()).map(s -> (ParameterizedType) s) // .filter(s -> s.getRawType() == SelfAop.class).findFirst().get(); // // @SuppressWarnings("unchecked") // return SpringUtil.getBean((Class<T>)currentType.getActualTypeArguments()[0]); } }
4. 原理分析
以上TestService.test() 执行时，将获得如下堆栈：
// ==================================== 1. DataSourceTransactionManager.java
@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);
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);
}
// 绑定到当前线程上下文,
// 在 DataSourceUtils.doGetConnection(dataSource) 中用到； 呼应上面的调用堆栈
// 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);
}
}
// ==================================== 2. TransactionSynchronizationManager.java
// 注意这是一个静态方法
public static void bindResource(Object key, Object value) throws IllegalStateException {
Object actualKey = TransactionSynchronizationUtils.unwrapResourceIfNecessary(key);
Assert.notNull(value, "Value must not be null");
Map<Object, Object> map = resources.get();
// set ThreadLocal Map if none found
if (map == null) {
map = new HashMap<>();
// resources类型为: ThreadLocal<Map<Object, Object>>
resources.set(map);
}
// 本例中:
// 1. actualKey实际类型为: com.baomidou.dynamic.datasource.DynamicRoutingDataSource
// 2. value实际类型为: org.springframework.jdbc.datasource.ConnectionHolder
Object oldValue = map.put(actualKey, value);
// Transparently suppress a ResourceHolder that was marked as void...
if (oldValue instanceof ResourceHolder && ((ResourceHolder) oldValue).isVoid()) {
oldValue = null;
}
if (oldValue != null) {
throw new IllegalStateException(
"Already value [" + oldValue + "] for key [" + actualKey + "] bound to thread");
}
}
// ==================================== 3. DataSourceUtils
public static Connection doGetConnection(DataSource dataSource) throws SQLException {
Assert.notNull(dataSource, "No DataSource specified");
ConnectionHolder conHolder = (ConnectionHolder) TransactionSynchronizationManager.getResource(dataSource);
if (conHolder != null && (conHolder.hasConnection() || conHolder.isSynchronizedWithTransaction())) {
//主库逻辑执行完毕，再次进行副库操作时，逻辑将走这条分支
conHolder.requested();
if (!conHolder.hasConnection()) {
logger.debug("Fetching resumed JDBC Connection from DataSource");
conHolder.setConnection(fetchConnection(dataSource));
}
return conHolder.getConnection();
}
// Else we either got no holder or an empty thread-bound holder here.
// ---- 从DataSource中获取Connection, 想要触发@DS, 必须走到这里
logger.debug("Fetching JDBC Connection from DataSource");
Connection con = fetchConnection(dataSource);
if (TransactionSynchronizationManager.isSynchronizationActive()) {
......
}
return con;
}
/ 针对以上 DataSourceUtils.doGetConnection(DataSource dataSource), 与Mybatis相关的堆栈如下：
SimpleExecutor.prepareStatement()
BaseExecutor.getConnection(Log statementLog)
transaction.getConnection(); // Connection是从事务中获取到的，实际类型为 SpringManagedTransaction
DataSourceUtils.doGetConnection(DataSource dataSource) // 如果ThreadLocal中发现存在Connect, 直接返回
4.1 相关类
SpringTransactionAnnotationParser。 启动阶段解析@TransactionalTransactionInterceptor// 执行阶段@TransactionalDynamicDataSourceAnnotationInterceptor// 针对注解@DS