多數(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)來做"壓棧",出去做"彈棧",數(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,通過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ù),無效;
- 2.內(nèi)層加事務(wù)@Transactional,無效;
- 3.改變事務(wù)的傳播方式@Transactional(propagation = Propagation.REQUIRES_NEW),事務(wù)生效
看了java方法棧和源碼,springframework5 里面spring-tx,知道問題出在什么地方,貼一個(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ù)庫連接 prepareSynchronization(status, definition); return status; }
doBegin 里我最關(guān)心兩點(diǎn),一個(gè)是數(shù)據(jù)庫連接的選擇和初始化,一個(gè)是把事務(wù)的自動(dòng)提交關(guān)掉
這里就能解釋得通,為什么@Transactional里的數(shù)據(jù)源還是舊的。因?yàn)殚_啟事務(wù)的同時(shí),會(huì)去數(shù)據(jù)庫連接池拿數(shù)據(jù)庫連接,如果只開啟一個(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ù)庫的操作都會(huì)使用代理之后的事務(wù)連接,所以會(huì)產(chǎn)生數(shù)據(jù)源沒有切換的問題
對(duì)于數(shù)據(jù)源的切換,必然要更替數(shù)據(jù)庫連接
我的理解是必須改變事務(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ù)的,但是人家是這樣說的
以上為個(gè)人經(jīng)驗(yàn),希望能給大家一個(gè)參考,也希望大家多多支持腳本之家。
相關(guān)文章
java使用websocket,并且獲取HttpSession 源碼分析(推薦)
這篇文章主要介紹了java使用websocket,并且獲取HttpSession,通過使用配置源碼分析了各方面知識(shí)點(diǎn),具體操作步驟大家可查看下文的詳細(xì)講解,感興趣的小伙伴們可以參考一下。2017-08-08Java中構(gòu)造函數(shù),set/get方法和toString方法使用及注意說明
這篇文章主要介紹了Java中構(gòu)造函數(shù),set/get方法和toString方法的使用及注意說明,具有很好的參考價(jià)值,希望對(duì)大家有所幫助。如有錯(cuò)誤或未考慮完全的地方,望不吝賜教2023-01-01java 利用反射獲取內(nèi)部類靜態(tài)成員變量的值操作
這篇文章主要介紹了java 利用反射獲取內(nèi)部類靜態(tài)成員變量的值操作,具有很好的參考價(jià)值,希望對(duì)大家有所幫助。一起跟隨小編過來看看吧2020-12-12Spring Boot 實(shí)例代碼之通過接口安全退出
這篇文章主要介紹了Spring Boot 實(shí)例代碼之通過接口安全退出的相關(guān)資料,需要的朋友可以參考下2017-09-09Spring很常用的@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-07SpringMVC表單標(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