多數(shù)據(jù)源@DS和@Transactional實(shí)戰(zhàn)
考慮到業(yè)務(wù)層面有多數(shù)據(jù)源切換的需求
同時(shí)又要考慮事務(wù),我使用了Mybatis-Plus3中的@DS作為多數(shù)據(jù)源的切換,它的原理的就是一個(gè)攔截器
@Override
public Object invoke(MethodInvocation invocation) throws Throwable {
try {
DynamicDataSourceContextHolder.push(determineDatasource(invocation));
return invocation.proceed();
} finally {
DynamicDataSourceContextHolder.poll();
}
}
里面的pull和poll實(shí)際就是操作一個(gè)容器
在環(huán)繞里面進(jìn)來(lái)做"壓棧",出去做"彈棧",數(shù)據(jù)結(jié)構(gòu)是這樣的
public final class DynamicDataSourceContextHolder {
/**
* 為什么要用鏈表存儲(chǔ)(準(zhǔn)確的是棧)
* <pre>
* 為了支持嵌套切換,如ABC三個(gè)service都是不同的數(shù)據(jù)源
* 其中A的某個(gè)業(yè)務(wù)要調(diào)B的方法,B的方法需要調(diào)用C的方法。一級(jí)一級(jí)調(diào)用切換,形成了鏈。
* 傳統(tǒng)的只設(shè)置當(dāng)前線程的方式不能滿足此業(yè)務(wù)需求,必須模擬棧,后進(jìn)先出。
* </pre>
*/
@SuppressWarnings("unchecked")
private static final ThreadLocal<Deque<String>> LOOKUP_KEY_HOLDER = new ThreadLocal() {
@Override
protected Object initialValue() {
return new ArrayDeque();
}
};
private DynamicDataSourceContextHolder() {
}
/**
* 獲得當(dāng)前線程數(shù)據(jù)源
*
* @return 數(shù)據(jù)源名稱
*/
public static String peek() {
return LOOKUP_KEY_HOLDER.get().peek();
}
/**
* 設(shè)置當(dāng)前線程數(shù)據(jù)源
* <p>
* 如非必要不要手動(dòng)調(diào)用,調(diào)用后確保最終清除
* </p>
*
* @param ds 數(shù)據(jù)源名稱
*/
public static void push(String ds) {
LOOKUP_KEY_HOLDER.get().push(StringUtils.isEmpty(ds) ? "" : ds);
}
/**
* 清空當(dāng)前線程數(shù)據(jù)源
* <p>
* 如果當(dāng)前線程是連續(xù)切換數(shù)據(jù)源 只會(huì)移除掉當(dāng)前線程的數(shù)據(jù)源名稱
* </p>
*/
public static void poll() {
Deque<String> deque = LOOKUP_KEY_HOLDER.get();
deque.poll();
if (deque.isEmpty()) {
LOOKUP_KEY_HOLDER.remove();
}
}
/**
* 強(qiáng)制清空本地線程
* <p>
* 防止內(nèi)存泄漏,如手動(dòng)調(diào)用了push可調(diào)用此方法確保清除
* </p>
*/
public static void clear() {
LOOKUP_KEY_HOLDER.remove();
}
上面就是@DS大概實(shí)現(xiàn),然后我就碰到坑了,外層service加了@Transactional,通過(guò)service調(diào)用另一個(gè)數(shù)據(jù)源做insert,在切面里看數(shù)據(jù)源切換了,但是還是顯示事務(wù)內(nèi)的數(shù)據(jù)源還是舊的,代碼結(jié)構(gòu)簡(jiǎn)單羅列下:
數(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調(diào)用的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();
}
內(nèi)層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號(hào)");
if(sr.insert()){
LambdaQueryWrapper<SapItemRecord> item01 = new QueryWrapper<SapItemRecord>().lambda().eq(SapItemRecord::getItemCode, name);
SapItemRecord sapItemRecord = new SapItemRecord().selectOne(item01);
ExceptionUtils.seed("內(nèi)層事務(wù)異常");
// return sapItemRecord.getDescription();
}
return "response : wonder";
}
}
- 1.最開始內(nèi)層不加事務(wù),全局只有一個(gè)事務(wù),無(wú)效;
- 2.內(nèi)層加事務(wù)@Transactional,無(wú)效;
- 3.改變事務(wù)的傳播方式@Transactional(propagation = Propagation.REQUIRES_NEW),事務(wù)生效
看了java方法棧和源碼,springframework5 里面spring-tx,知道問(wèn)題出在什么地方,貼一個(gè)調(diào)用棧截圖

spring的事務(wù)是基于aop的,這個(gè)不解釋了,直接進(jìn)入事務(wù)攔截器TransactionInterceptor,找到它調(diào)用的invokeWithinTransaction方法,只看本文章關(guān)注部分
根據(jù)method的注解判斷是否開啟事務(wù)
處理異常,在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) {
// 重點(diǎn)是這里,獲取事務(wù)
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);
}
這里就是按照不同的事務(wù)傳播機(jī)制
去做不同的處理,判斷是否存在事務(wù),存在事務(wù)就執(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)建新事務(wù)
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里面關(guān)乎數(shù)據(jù)源和數(shù)據(jù)庫(kù)連接
prepareSynchronization(status, definition);
return status;
}
doBegin 里我最關(guān)心兩點(diǎn),一個(gè)是數(shù)據(jù)庫(kù)連接的選擇和初始化,一個(gè)是把事務(wù)的自動(dòng)提交關(guān)掉

這里就能解釋得通,為什么@Transactional里的數(shù)據(jù)源還是舊的。因?yàn)殚_啟事務(wù)的同時(shí),會(huì)去數(shù)據(jù)庫(kù)連接池拿數(shù)據(jù)庫(kù)連接,如果只開啟一個(gè)事務(wù),在切面時(shí)候會(huì)獲取數(shù)據(jù)源,設(shè)置dataSource;如果在內(nèi)層的service使用@DS切換了數(shù)據(jù)源,實(shí)際上是又做了一層攔截,改變了DataSourceHolder的棧頂dataSource,對(duì)于整個(gè)事務(wù)的連接是沒有影響的,在這個(gè)事務(wù)切面內(nèi)的所有數(shù)據(jù)庫(kù)的操作都會(huì)使用代理之后的事務(wù)連接,所以會(huì)產(chǎn)生數(shù)據(jù)源沒有切換的問(wèn)題
對(duì)于數(shù)據(jù)源的切換,必然要更替數(shù)據(jù)庫(kù)連接
我的理解是必須改變事務(wù)的傳播機(jī)制,產(chǎn)生新的事務(wù),所以第一內(nèi)層service不僅要加@DS,還要加@Transactional注解,并且指定
Propagation.REQUIRES_NEW,因?yàn)檫@樣在處理handleExistingTransaction 時(shí),就會(huì)走這段邏輯
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)建新事務(wù),重新拿切換之后的dataSource作為新事務(wù)的conn,這樣內(nèi)層事務(wù)的數(shù)據(jù)源就是@DS注解內(nèi)的,從而完成了數(shù)據(jù)源切換并且事務(wù)生效,PROPAGATION_REQUIRES_NEW 方式下,事務(wù)的回滾都是生效的,親測(cè),所以使用MybatisPlus3.x的可以使用@DS了,當(dāng)然你也可以自己寫切面去切換DataSource,原理跟DS差不多,我用baomidou,因?yàn)樗惆?!但是我覺得baomidou在考慮切換數(shù)據(jù)源的時(shí)候,本身要考慮事務(wù)的,但是人家是這樣說(shuō)的

以上為個(gè)人經(jīng)驗(yàn),希望能給大家一個(gè)參考,也希望大家多多支持腳本之家。
相關(guān)文章
java使用websocket,并且獲取HttpSession 源碼分析(推薦)
這篇文章主要介紹了java使用websocket,并且獲取HttpSession,通過(guò)使用配置源碼分析了各方面知識(shí)點(diǎn),具體操作步驟大家可查看下文的詳細(xì)講解,感興趣的小伙伴們可以參考一下。2017-08-08
Java中構(gòu)造函數(shù),set/get方法和toString方法使用及注意說(shuō)明
這篇文章主要介紹了Java中構(gòu)造函數(shù),set/get方法和toString方法的使用及注意說(shuō)明,具有很好的參考價(jià)值,希望對(duì)大家有所幫助。如有錯(cuò)誤或未考慮完全的地方,望不吝賜教2023-01-01
java 利用反射獲取內(nèi)部類靜態(tài)成員變量的值操作
這篇文章主要介紹了java 利用反射獲取內(nèi)部類靜態(tài)成員變量的值操作,具有很好的參考價(jià)值,希望對(duì)大家有所幫助。一起跟隨小編過(guò)來(lái)看看吧2020-12-12
Spring Boot 實(shí)例代碼之通過(guò)接口安全退出
這篇文章主要介紹了Spring Boot 實(shí)例代碼之通過(guò)接口安全退出的相關(guān)資料,需要的朋友可以參考下2017-09-09
Spring很常用的@Conditional注解的使用場(chǎng)景和源碼解析
今天要分享的是Spring的注解@Conditional,@Conditional是一個(gè)條件注解,它的作用是判斷Bean是否滿足條件,本文詳細(xì)介紹了@Conditional注解的使用場(chǎng)景和源碼,需要的朋友可以參考一下2023-04-04
詳解Java線程池如何實(shí)現(xiàn)優(yōu)雅退出
這篇文章我們將從源碼角度深度解析線程池是如何優(yōu)雅的退出程序的,文中的示例代碼講解詳細(xì),對(duì)我們學(xué)習(xí)java線程池有一定幫助,需要的可以參考一下2022-07-07
SpringMVC表單標(biāo)簽知識(shí)點(diǎn)詳解
這篇文章主要為大家詳細(xì)介紹了SpringMVC表單標(biāo)簽知識(shí)點(diǎn),具有一定的參考價(jià)值,感興趣的小伙伴們可以參考一下2017-10-10
因不會(huì)遠(yuǎn)程debug調(diào)試我被項(xiàng)目經(jīng)理嘲笑了
這篇文章主要介紹了遠(yuǎn)程debug調(diào)試的相關(guān)內(nèi)容,本文給大家介紹的非常詳細(xì),對(duì)大家的學(xué)習(xí)或工作具有一定的參考借鑒價(jià)值,需要的朋友可以參考下2021-08-08

