多數(shù)據(jù)源@DS和@Transactional實戰(zhàn)
考慮到業(yè)務層面有多數(shù)據(jù)源切換的需求
同時又要考慮事務,我使用了Mybatis-Plus3中的@DS作為多數(shù)據(jù)源的切換,它的原理的就是一個攔截器
@Override
public Object invoke(MethodInvocation invocation) throws Throwable {
try {
DynamicDataSourceContextHolder.push(determineDatasource(invocation));
return invocation.proceed();
} finally {
DynamicDataSourceContextHolder.poll();
}
}
里面的pull和poll實際就是操作一個容器
在環(huán)繞里面進來做"壓棧",出去做"彈棧",數(shù)據(jù)結構是這樣的
public final class DynamicDataSourceContextHolder {
/**
* 為什么要用鏈表存儲(準確的是棧)
* <pre>
* 為了支持嵌套切換,如ABC三個service都是不同的數(shù)據(jù)源
* 其中A的某個業(yè)務要調B的方法,B的方法需要調用C的方法。一級一級調用切換,形成了鏈。
* 傳統(tǒng)的只設置當前線程的方式不能滿足此業(yè)務需求,必須模擬棧,后進先出。
* </pre>
*/
@SuppressWarnings("unchecked")
private static final ThreadLocal<Deque<String>> LOOKUP_KEY_HOLDER = new ThreadLocal() {
@Override
protected Object initialValue() {
return new ArrayDeque();
}
};
private DynamicDataSourceContextHolder() {
}
/**
* 獲得當前線程數(shù)據(jù)源
*
* @return 數(shù)據(jù)源名稱
*/
public static String peek() {
return LOOKUP_KEY_HOLDER.get().peek();
}
/**
* 設置當前線程數(shù)據(jù)源
* <p>
* 如非必要不要手動調用,調用后確保最終清除
* </p>
*
* @param ds 數(shù)據(jù)源名稱
*/
public static void push(String ds) {
LOOKUP_KEY_HOLDER.get().push(StringUtils.isEmpty(ds) ? "" : ds);
}
/**
* 清空當前線程數(shù)據(jù)源
* <p>
* 如果當前線程是連續(xù)切換數(shù)據(jù)源 只會移除掉當前線程的數(shù)據(jù)源名稱
* </p>
*/
public static void poll() {
Deque<String> deque = LOOKUP_KEY_HOLDER.get();
deque.poll();
if (deque.isEmpty()) {
LOOKUP_KEY_HOLDER.remove();
}
}
/**
* 強制清空本地線程
* <p>
* 防止內存泄漏,如手動調用了push可調用此方法確保清除
* </p>
*/
public static void clear() {
LOOKUP_KEY_HOLDER.remove();
}
上面就是@DS大概實現(xiàn),然后我就碰到坑了,外層service加了@Transactional,通過service調用另一個數(shù)據(jù)源做insert,在切面里看數(shù)據(jù)源切換了,但是還是顯示事務內的數(shù)據(jù)源還是舊的,代碼結構簡單羅列下:
數(shù)據(jù)源
dynamic:
primary: master
strict: false
datasource:
master:
driver-class-name: com.mysql.cj.jdbc.Driver
url: jdbc:mysql://***/phorcys-centre?useSSL=false
username: root
password: *****
interface:
url: jdbc:mysql://***/phorcys-interface?useSSL=false
username: root
password: *****
driver-class-name: com.mysql.cj.jdbc.Driver
外層controller調用的service
@Autowired
UserService userService;
@Autowired
RedisClient redisClient;
@GetMapping("/demo")
@Transactional
public GeneralResponse demo(@RequestBody(required = false) GeneralRequest request){
SysUser sysUser = new SysUser();
sysUser.setCode("wonder");
sysUser.setName("王吉坤");
sysUser.insert();
redisClient.set("token",sysUser);
List<SysUser> sysUsers = new SysUser().selectAll();
String item01 = userService.getUserInfo("ITEM01");
return GeneralResponse.success();
}
內層service
@Service
public class UserServiceImpl implements UserService {
@Override
@DS("interface")
@Transactional
// @Transactional(propagation = Propagation.REQUIRES_NEW)
public String getUserInfo(String name) {
SapItemRecord sr = new SapItemRecord();
sr.setBatchId(1L);
sr.setItemCode("ITEM01");
sr.setDescription("物料1號");
if(sr.insert()){
LambdaQueryWrapper<SapItemRecord> item01 = new QueryWrapper<SapItemRecord>().lambda().eq(SapItemRecord::getItemCode, name);
SapItemRecord sapItemRecord = new SapItemRecord().selectOne(item01);
ExceptionUtils.seed("內層事務異常");
// return sapItemRecord.getDescription();
}
return "response : wonder";
}
}
- 1.最開始內層不加事務,全局只有一個事務,無效;
- 2.內層加事務@Transactional,無效;
- 3.改變事務的傳播方式@Transactional(propagation = Propagation.REQUIRES_NEW),事務生效
看了java方法棧和源碼,springframework5 里面spring-tx,知道問題出在什么地方,貼一個調用棧截圖

spring的事務是基于aop的,這個不解釋了,直接進入事務攔截器TransactionInterceptor,找到它調用的invokeWithinTransaction方法,只看本文章關注部分
根據(jù)method的注解判斷是否開啟事務
處理異常,在finally里處理cleanupTransactionInfo
if (txAttr == null || !(ptm instanceof CallbackPreferringPlatformTransactionManager)) {
// Standard transaction demarcation with getTransaction and commit/rollback calls.
TransactionInfo txInfo = createTransactionIfNecessary(ptm, txAttr, joinpointIdentification);
Object retVal;
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);
}
....
}
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);
}
這里就是按照不同的事務傳播機制
去做不同的處理,判斷是否存在事務,存在事務就執(zhí)行handleExistingTransaction,不存在的話滿足創(chuàng)建的條件就startTransaction,這里我的情形就是第一次直接創(chuàng)建,第二次執(zhí)行exist邏輯
public final TransactionStatus getTransaction(@Nullable TransactionDefinition definition)
throws TransactionException {
// Use defaults if no transaction definition given.
TransactionDefinition def = (definition != null ? definition : TransactionDefinition.withDefaults());
Object transaction = doGetTransaction();
boolean debugEnabled = logger.isDebugEnabled();
if (isExistingTransaction(transaction)) {
// Existing transaction found -> check propagation behavior to find out how to behave.
return handleExistingTransaction(def, transaction, debugEnabled);
}
// Check definition settings for new transaction.
if (def.getTimeout() < TransactionDefinition.TIMEOUT_DEFAULT) {
throw new InvalidTimeoutException("Invalid transaction timeout", def.getTimeout());
}
// No existing transaction found -> check propagation behavior to find out how to proceed.
if (def.getPropagationBehavior() == TransactionDefinition.PROPAGATION_MANDATORY) {
throw new IllegalTransactionStateException(
"No existing transaction found for transaction marked with propagation 'mandatory'");
}
else if (def.getPropagationBehavior() == TransactionDefinition.PROPAGATION_REQUIRED ||
def.getPropagationBehavior() == TransactionDefinition.PROPAGATION_REQUIRES_NEW ||
def.getPropagationBehavior() == TransactionDefinition.PROPAGATION_NESTED) {
SuspendedResourcesHolder suspendedResources = suspend(null);
if (debugEnabled) {
logger.debug("Creating new transaction with name [" + def.getName() + "]: " + def);
}
try {
return startTransaction(def, transaction, debugEnabled, suspendedResources);
}
catch (RuntimeException | Error ex) {
resume(null, suspendedResources);
throw ex;
}
}
else {
// Create "empty" transaction: no actual transaction, but potentially synchronization.
if (def.getIsolationLevel() != TransactionDefinition.ISOLATION_DEFAULT && logger.isWarnEnabled()) {
logger.warn("Custom isolation level specified but no actual transaction initiated; " +
"isolation level will effectively be ignored: " + def);
}
boolean newSynchronization = (getTransactionSynchronization() == SYNCHRONIZATION_ALWAYS);
return prepareTransactionStatus(def, null, true, newSynchronization, debugEnabled, null);
}
}
這里是創(chuàng)建新事務
private TransactionStatus startTransaction(TransactionDefinition definition, Object transaction,
boolean debugEnabled, @Nullable SuspendedResourcesHolder suspendedResources) {
boolean newSynchronization = (getTransactionSynchronization() != SYNCHRONIZATION_NEVER);
DefaultTransactionStatus status = newTransactionStatus(
definition, transaction, true, newSynchronization, debugEnabled, suspendedResources);
doBegin(transaction, definition); //dobegin里面關乎數(shù)據(jù)源和數(shù)據(jù)庫連接
prepareSynchronization(status, definition);
return status;
}
doBegin 里我最關心兩點,一個是數(shù)據(jù)庫連接的選擇和初始化,一個是把事務的自動提交關掉

這里就能解釋得通,為什么@Transactional里的數(shù)據(jù)源還是舊的。因為開啟事務的同時,會去數(shù)據(jù)庫連接池拿數(shù)據(jù)庫連接,如果只開啟一個事務,在切面時候會獲取數(shù)據(jù)源,設置dataSource;如果在內層的service使用@DS切換了數(shù)據(jù)源,實際上是又做了一層攔截,改變了DataSourceHolder的棧頂dataSource,對于整個事務的連接是沒有影響的,在這個事務切面內的所有數(shù)據(jù)庫的操作都會使用代理之后的事務連接,所以會產生數(shù)據(jù)源沒有切換的問題
對于數(shù)據(jù)源的切換,必然要更替數(shù)據(jù)庫連接
我的理解是必須改變事務的傳播機制,產生新的事務,所以第一內層service不僅要加@DS,還要加@Transactional注解,并且指定
Propagation.REQUIRES_NEW,因為這樣在處理handleExistingTransaction 時,就會走這段邏輯
if (definition.getPropagationBehavior() == TransactionDefinition.PROPAGATION_REQUIRES_NEW) {
if (debugEnabled) {
logger.debug("Suspending current transaction, creating new transaction with name [" +
definition.getName() + "]");
}
SuspendedResourcesHolder suspendedResources = suspend(transaction);
try {
return startTransaction(definition, transaction, debugEnabled, suspendedResources);
}
catch (RuntimeException | Error beginEx) {
resumeAfterBeginException(transaction, suspendedResources, beginEx);
throw beginEx;
}
}
走startTransaction,再doBegin,創(chuàng)建新事務,重新拿切換之后的dataSource作為新事務的conn,這樣內層事務的數(shù)據(jù)源就是@DS注解內的,從而完成了數(shù)據(jù)源切換并且事務生效,PROPAGATION_REQUIRES_NEW 方式下,事務的回滾都是生效的,親測,所以使用MybatisPlus3.x的可以使用@DS了,當然你也可以自己寫切面去切換DataSource,原理跟DS差不多,我用baomidou,因為它香啊!但是我覺得baomidou在考慮切換數(shù)據(jù)源的時候,本身要考慮事務的,但是人家是這樣說的

以上為個人經驗,希望能給大家一個參考,也希望大家多多支持腳本之家。
相關文章
詳解SpringBoot AOP 攔截器(Aspect注解方式)
這篇文章主要介紹了詳解SpringBoot AOP 攔截器 Aspect,小編覺得挺不錯的,現(xiàn)在分享給大家,也給大家做個參考。一起跟隨小編過來看看吧2017-05-05
java使用CountDownLatch實現(xiàn)多線程協(xié)作
在多線程編程中,經常需要實現(xiàn)一種機制來協(xié)調多個線程的執(zhí)行,以確保某些操作在所有線程完成后再進行,CountDownLatch?就是?Java?并發(fā)包中提供的一種同步工具,下面我們就來看看如何使用CountDownLatch實現(xiàn)多線程協(xié)作吧2023-11-11

