分散式事務(Seata)原理 詳解篇，建議收藏
2022-07-10 18:02:33
前言
在之前的系列中，我們講解了關於Seata基本介紹和實際應用，今天帶來的這篇，就給大家分析一下Seata的原始碼是如何一步一步實現的。讀原始碼的時候我們需要俯瞰起全貌，不要去扣一個一個的細節，這樣我們學習起來會快捷而且有效率，我們學習原始碼需要掌握的是整體思路和核心點。
首先 Seata 使用者端啟動一般分為以下幾個流程：
- 自動載入Bean屬性和設定資訊
- 初始化TM
- 初始化RM
- 初始化分散式事務使用者端完成，完成代理資料庫設定
- 連線TC(Seata伺服器端)，註冊RM和TM
- 開啟全域性事務
在這篇原始碼的講解中，我們主要以AT模式為主導，官網也是主推AT模式，我們在上篇的文章中也講解過，感興趣的小夥伴可以去看一看分散式事務(Seata) 四大模式詳解，在官網中也提供了對應的流程地址：https://seata.io/zh-cn/docs/dev/mode/at-mode.html ，在這裡我們只是做一些簡單的介紹，AT模式主要分為兩個階段：
一階段：
- 解析SQL，獲取SQL型別（CRUD）、表資訊、條件(where) 等相關資訊
- 查詢前映象(改變之前的資料)，根據解析得到的條件資訊，生成查詢語句，定位資料
- 執行業務SQL，更新資料
- 查詢後映象（改變後的資料），根據前映象的結果，通過主鍵都給你為資料
- 插入回滾紀錄檔，將前後映象資料以及業務SQL等資訊，組織成一條回滾紀錄檔記錄，插入到undo Log表中
- 提交前，向TC註冊分支，申請全域性鎖
- 本地事務提交，業務資料的更細膩和生成的undoLog一起提交
- 將本地事務提交的結果通知給TC
二階段：
如果TC收到的是回滾請求
- 開啟本地事務，通過XID和BranchID查詢到對應的undo Log記錄
- 根據undoLog中的前映象和業務SQL的相關資訊生成並執行回滾語句
- 提交本地事務，將本地事務的執行結果（分支事務回滾的資訊）通知給TC
如果沒問題，執行提交操作
- 收到TC分支提交請求，將請求放入到一個非同步任務的佇列中，馬上返回提交成功的結果給TC
- 非同步任務階段的分支提交請求刪除undoLog中記錄
原始碼入口
接下來，我們就需要從官網中去下載原始碼，下載地址：https://seata.io/zh-cn/blog/download.html，選擇 source 即可，下載完成之後，通過IDEA開啟專案。
原始碼下載下來之後，我們應該如何去找入口呢？首先我們需要找到對應引入的 Seata 包 spring-alibaba-seata，我們在回想一下，我們開啟事務的時候，是不是新增過一個@GlobalTransactional的註解，這個註解就是我們入手的一個點，我們在 spring.factories 中看到有一個 GlobalTransactionAutoConfiguration，這個就是我們需要關注的點，也就是我們原始碼的入口
在 GlobalTransactionAutoConfiguration 中我們找到一個用Bean注入的方法 globalTransactionScanner ，這個就是全域性事務掃描器，這個型別主要負責載入設定，注入相關的Bean
這裡給大家展示了當前GlobalTransactionScanner的類關係圖，其中我們現在繼承了Aop的AbstractAutoProxyCreator型別，在這其中有一個重點方法，這個方法就是判斷Bean物件是否需要代理，是否需要增強。
@Configuration
@EnableConfigurationProperties(SeataProperties.class)
public class GlobalTransactionAutoConfiguration {
//全域性事務掃描器
@Bean
public GlobalTransactionScanner globalTransactionScanner() {
String applicationName = applicationContext.getEnvironment()
.getProperty("spring.application.name");
String txServiceGroup = seataProperties.getTxServiceGroup();
if (StringUtils.isEmpty(txServiceGroup)) {
txServiceGroup = applicationName + "-fescar-service-group";
seataProperties.setTxServiceGroup(txServiceGroup);
}
// 構建全域性掃描器，傳入引數：應用名、事務分組名，失敗處理器
return new GlobalTransactionScanner(applicationName, txServiceGroup);
}
}
在這其中我們要關心的是 GlobalTransactionScanner 這個型別，這個型別掃描 @GlobalTransactional 註解，並對代理方法進行攔截增強事務的功能。我們就從原始碼中搜尋這個GlobalTransactionScanner類，看看裡面具體是做了什麼
/**
* The type Global transaction scanner.
* 全域性事務掃描器
* @author slievrly
*/
public class GlobalTransactionScanner
//AbstractAutoProxyCreator AOP動態代理 增強Bean
extends AbstractAutoProxyCreator
/**
* ConfigurationChangeListener: 監聽器基準介面
* InitializingBean： Bean初始化
* ApplicationContextAware： Spring容器
* DisposableBean： Spring 容器銷燬
*/
implements ConfigurationChangeListener, InitializingBean, ApplicationContextAware, DisposableBean {
private final String applicationId;//服務名
private final String txServiceGroup;//事務分組
private void initClient() {
//啟動紀錄檔
if (LOGGER.isInfoEnabled()) {
LOGGER.info("Initializing Global Transaction Clients ... ");
}
//檢查應用名以及事務分組名，為空丟擲異常IllegalArgumentException
if (DEFAULT_TX_GROUP_OLD.equals(txServiceGroup)) {
LOGGER.warn("the default value of seata.tx-service-group: {} has already changed to {} since Seata 1.5, " +
"please change your default configuration as soon as possible " +
"and we don't recommend you to use default tx-service-group's value provided by seata",
DEFAULT_TX_GROUP_OLD, DEFAULT_TX_GROUP);
}
if (StringUtils.isNullOrEmpty(applicationId) || StringUtils.isNullOrEmpty(txServiceGroup)) {
throw new IllegalArgumentException(String.format("applicationId: %s, txServiceGroup: %s", applicationId, txServiceGroup));
}
//init TM
//初始化TM
TMClient.init(applicationId, txServiceGroup, accessKey, secretKey);
if (LOGGER.isInfoEnabled()) {
LOGGER.info("Transaction Manager Client is initialized. applicationId[{}] txServiceGroup[{}]", applicationId, txServiceGroup);
}
//init RM
//初始化RM
RMClient.init(applicationId, txServiceGroup);
if (LOGGER.isInfoEnabled()) {
LOGGER.info("Resource Manager is initialized. applicationId[{}] txServiceGroup[{}]", applicationId, txServiceGroup);
}
if (LOGGER.isInfoEnabled()) {
LOGGER.info("Global Transaction Clients are initialized. ");
}
registerSpringShutdownHook();
}
@Override
public void afterPropertiesSet() {
if (disableGlobalTransaction) {
if (LOGGER.isInfoEnabled()) {
LOGGER.info("Global transaction is disabled.");
}
ConfigurationCache.addConfigListener(ConfigurationKeys.DISABLE_GLOBAL_TRANSACTION,
(ConfigurationChangeListener)this);
return;
}
if (initialized.compareAndSet(false, true)) {
initClient();
}
}
private void initClient() {
//啟動紀錄檔
if (LOGGER.isInfoEnabled()) {
LOGGER.info("Initializing Global Transaction Clients ... ");
}
//檢查應用名以及事務分組名，為空丟擲異常IllegalArgumentException
if (DEFAULT_TX_GROUP_OLD.equals(txServiceGroup)) {
LOGGER.warn("the default value of seata.tx-service-group: {} has already changed to {} since Seata 1.5, " +
"please change your default configuration as soon as possible " +
"and we don't recommend you to use default tx-service-group's value provided by seata",
DEFAULT_TX_GROUP_OLD, DEFAULT_TX_GROUP);
}
//檢查應用名以及事務分組名，為空丟擲異常IllegalArgumentException
if (StringUtils.isNullOrEmpty(applicationId) || StringUtils.isNullOrEmpty(txServiceGroup)) {
throw new IllegalArgumentException(String.format("applicationId: %s, txServiceGroup: %s", applicationId, txServiceGroup));
}
//init TM
//初始化TM
TMClient.init(applicationId, txServiceGroup, accessKey, secretKey);
if (LOGGER.isInfoEnabled()) {
LOGGER.info("Transaction Manager Client is initialized. applicationId[{}] txServiceGroup[{}]", applicationId, txServiceGroup);
}
//init RM
//初始化RM
RMClient.init(applicationId, txServiceGroup);
if (LOGGER.isInfoEnabled()) {
LOGGER.info("Resource Manager is initialized. applicationId[{}] txServiceGroup[{}]", applicationId, txServiceGroup);
}
if (LOGGER.isInfoEnabled()) {
LOGGER.info("Global Transaction Clients are initialized. ");
}
registerSpringShutdownHook();
}
//代理增強，Spring 所有的Bean都會經過這個方法
@Override
protected Object wrapIfNecessary(Object bean, String beanName, Object cacheKey) {
// do checkers
//檢查bean和beanName
if (!doCheckers(bean, beanName)) {
return bean;
}
try {
//加鎖防止並行
synchronized (PROXYED_SET) {
if (PROXYED_SET.contains(beanName)) {
return bean;
}
interceptor = null;
//check TCC proxy
//檢查是否為TCC模式
if (TCCBeanParserUtils.isTccAutoProxy(bean, beanName, applicationContext)) {
// init tcc fence clean task if enable useTccFence
//如果啟用useTccFence 失敗 ，則初始化TCC清理任務
TCCBeanParserUtils.initTccFenceCleanTask(TCCBeanParserUtils.getRemotingDesc(beanName), applicationContext);
//TCC interceptor, proxy bean of sofa:reference/dubbo:reference, and LocalTCC
//如果是，新增TCC攔截器
interceptor = new TccActionInterceptor(TCCBeanParserUtils.getRemotingDesc(beanName));
ConfigurationCache.addConfigListener(ConfigurationKeys.DISABLE_GLOBAL_TRANSACTION,
(ConfigurationChangeListener)interceptor);
} else {
//不是TCC
Class<?> serviceInterface = SpringProxyUtils.findTargetClass(bean);
Class<?>[] interfacesIfJdk = SpringProxyUtils.findInterfaces(bean);
//判斷是否有相關事務註解，如果沒有不進行代理
if (!existsAnnotation(new Class[]{serviceInterface})
&& !existsAnnotation(interfacesIfJdk)) {
return bean;
}
//發現存在全域性事務註解標註的Bean物件，新增攔截器
if (globalTransactionalInterceptor == null) {
//新增攔截器
globalTransactionalInterceptor = new GlobalTransactionalInterceptor(failureHandlerHook);
ConfigurationCache.addConfigListener(
ConfigurationKeys.DISABLE_GLOBAL_TRANSACTION,
(ConfigurationChangeListener)globalTransactionalInterceptor);
}
interceptor = globalTransactionalInterceptor;
}
LOGGER.info("Bean[{}] with name [{}] would use interceptor [{}]", bean.getClass().getName(), beanName, interceptor.getClass().getName());
//檢查是否為代理物件
if (!AopUtils.isAopProxy(bean)) {
//不是代理物件，呼叫父級
bean = super.wrapIfNecessary(bean, beanName, cacheKey);
} else {
//是代理物件，反射獲取代理類中已經存在的攔截器組合，然後新增到這個集合中
AdvisedSupport advised = SpringProxyUtils.getAdvisedSupport(bean);
Advisor[] advisor = buildAdvisors(beanName, getAdvicesAndAdvisorsForBean(null, null, null));
int pos;
for (Advisor avr : advisor) {
// Find the position based on the advisor's order, and add to advisors by pos
pos = findAddSeataAdvisorPosition(advised, avr);
advised.addAdvisor(pos, avr);
}
}
PROXYED_SET.add(beanName);
return bean;
}
} catch (Exception exx) {
throw new RuntimeException(exx);
}
}
}
InitializingBean：中實現了一個 afterPropertiesSet()方法，在這個方法中，呼叫了initClient()
AbstractAutoProxyCreator：APO動態代理，在之前的的Nacos和Sentiel中都有這個代理類，AOP在我們越往深入學習，在學習原始碼的會見到的越來越多，越來越重要，很多相關代理，都是通過AOP進行增強,在這個類中，我們需要關注有一個wrapIfNecessary()方法， 這個方法主要是判斷被代理的bean或者類是否需要代理增強，在這個方法中會呼叫GlobalTransactionalInterceptor.invoke()進行帶來增強。
具體程式碼如下：
public class GlobalTransactionalInterceptor implements ConfigurationChangeListener, MethodInterceptor, SeataInterceptor {
public GlobalTransactionalInterceptor(FailureHandler failureHandler) {
this.failureHandler = failureHandler == null ? DEFAULT_FAIL_HANDLER : failureHandler;
this.disable = ConfigurationFactory.getInstance().getBoolean(ConfigurationKeys.DISABLE_GLOBAL_TRANSACTION,
DEFAULT_DISABLE_GLOBAL_TRANSACTION);
this.order =
ConfigurationFactory.getInstance().getInt(ConfigurationKeys.TM_INTERCEPTOR_ORDER, TM_INTERCEPTOR_ORDER);
degradeCheck = ConfigurationFactory.getInstance().getBoolean(ConfigurationKeys.CLIENT_DEGRADE_CHECK,
DEFAULT_TM_DEGRADE_CHECK);
if (degradeCheck) {
ConfigurationCache.addConfigListener(ConfigurationKeys.CLIENT_DEGRADE_CHECK, this);
degradeCheckPeriod = ConfigurationFactory.getInstance()
.getInt(ConfigurationKeys.CLIENT_DEGRADE_CHECK_PERIOD, DEFAULT_TM_DEGRADE_CHECK_PERIOD);
degradeCheckAllowTimes = ConfigurationFactory.getInstance()
.getInt(ConfigurationKeys.CLIENT_DEGRADE_CHECK_ALLOW_TIMES, DEFAULT_TM_DEGRADE_CHECK_ALLOW_TIMES);
EVENT_BUS.register(this);
if (degradeCheckPeriod > 0 && degradeCheckAllowTimes > 0) {
startDegradeCheck();
}
}
this.initDefaultGlobalTransactionTimeout();
}
@Override
public Object invoke(final MethodInvocation methodInvocation) throws Throwable {
//獲取執行的方法
Class<?> targetClass =
methodInvocation.getThis() != null ? AopUtils.getTargetClass(methodInvocation.getThis()) : null;
Method specificMethod = ClassUtils.getMostSpecificMethod(methodInvocation.getMethod(), targetClass);
if (specificMethod != null && !specificMethod.getDeclaringClass().equals(Object.class)) {
final Method method = BridgeMethodResolver.findBridgedMethod(specificMethod);
//獲取GlobalTransactional（全域性事務）、GlobalLock(全域性鎖)後設資料
final GlobalTransactional globalTransactionalAnnotation =
getAnnotation(method, targetClass, GlobalTransactional.class);
//GlobalLock會將本地事務的執行納入Seata分散式事務的管理，共同競爭全域性鎖
//保證全域性事務在執行的時候，本地事務不可以操作全域性事務的記錄
final GlobalLock globalLockAnnotation = getAnnotation(method, targetClass, GlobalLock.class);//獲取全域性鎖
boolean localDisable = disable || (degradeCheck && degradeNum >= degradeCheckAllowTimes);
if (!localDisable) {
if (globalTransactionalAnnotation != null || this.aspectTransactional != null) {
AspectTransactional transactional;
if (globalTransactionalAnnotation != null) {
transactional = new AspectTransactional(globalTransactionalAnnotation.timeoutMills(),
globalTransactionalAnnotation.name(), globalTransactionalAnnotation.rollbackFor(),
globalTransactionalAnnotation.noRollbackForClassName(),
globalTransactionalAnnotation.noRollbackFor(),
globalTransactionalAnnotation.noRollbackForClassName(),
globalTransactionalAnnotation.propagation(),
globalTransactionalAnnotation.lockRetryInterval(),
globalTransactionalAnnotation.lockRetryTimes());
} else {
transactional = this.aspectTransactional;
}
//執行全域性事務
return handleGlobalTransaction(methodInvocation, transactional);
} else if (globalLockAnnotation != null) {
//執行全域性鎖
return handleGlobalLock(methodInvocation, globalLockAnnotation);
}
}
}
return methodInvocation.proceed();
}
}
具體流程圖如下所示：
核心原始碼
在上面我們講解到 GlobalTransactionalInterceptor 作為全域性事務攔截器，一旦執行攔截，就會進入invoke方法，其中，我們會做 @GlobalTransactional 註解的判斷，如果有這個註解的存在，會執行全域性事務和全域性鎖，再執行全域性事務的時候會呼叫 handleGlobalTransaction 全域性事務處理器，獲取事務資訊，那我們接下來就來看一下 GlobalTransactionalInterceptor.handleGlobalTransaction 到底是如何執行全域性事務的
Object handleGlobalTransaction(final MethodInvocation methodInvocation,
final AspectTransactional aspectTransactional) throws Throwable {
boolean succeed = true;
try {
return transactionalTemplate.execute(new TransactionalExecutor() {
@Override
public Object execute() throws Throwable {
return methodInvocation.proceed();
}
//獲取事務名稱，預設獲取方法名
public String name() {
String name = aspectTransactional.getName();
if (!StringUtils.isNullOrEmpty(name)) {
return name;
}
return formatMethod(methodInvocation.getMethod());
}
/**
* 解析GlobalTransation註解屬性，封裝對物件
* @return
*/
@Override
public TransactionInfo getTransactionInfo() {
// reset the value of timeout
//獲取超時時間，預設60秒
int timeout = aspectTransactional.getTimeoutMills();
if (timeout <= 0 || timeout == DEFAULT_GLOBAL_TRANSACTION_TIMEOUT) {
timeout = defaultGlobalTransactionTimeout;
}
//構建事務資訊物件
TransactionInfo transactionInfo = new TransactionInfo();
transactionInfo.setTimeOut(timeout);//超時時間
transactionInfo.setName(name());//事務名稱
transactionInfo.setPropagation(aspectTransactional.getPropagation());//事務傳播
transactionInfo.setLockRetryInterval(aspectTransactional.getLockRetryInterval());//校驗或佔用全域性鎖重試間隔
transactionInfo.setLockRetryTimes(aspectTransactional.getLockRetryTimes());//校驗或佔用全域性鎖重試次數
Set<RollbackRule> rollbackRules = new LinkedHashSet<>();
//其他構建資訊
for (Class<?> rbRule : aspectTransactional.getRollbackFor()) {
rollbackRules.add(new RollbackRule(rbRule));
}
for (String rbRule : aspectTransactional.getRollbackForClassName()) {
rollbackRules.add(new RollbackRule(rbRule));
}
for (Class<?> rbRule : aspectTransactional.getNoRollbackFor()) {
rollbackRules.add(new NoRollbackRule(rbRule));
}
for (String rbRule : aspectTransactional.getNoRollbackForClassName()) {
rollbackRules.add(new NoRollbackRule(rbRule));
}
transactionInfo.setRollbackRules(rollbackRules);
return transactionInfo;
}
});
} catch (TransactionalExecutor.ExecutionException e) {
//執行異常
TransactionalExecutor.Code code = e.getCode();
switch (code) {
case RollbackDone:
throw e.getOriginalException();
case BeginFailure:
succeed = false;
failureHandler.onBeginFailure(e.getTransaction(), e.getCause());
throw e.getCause();
case CommitFailure:
succeed = false;
failureHandler.onCommitFailure(e.getTransaction(), e.getCause());
throw e.getCause();
case RollbackFailure:
failureHandler.onRollbackFailure(e.getTransaction(), e.getOriginalException());
throw e.getOriginalException();
case RollbackRetrying:
failureHandler.onRollbackRetrying(e.getTransaction(), e.getOriginalException());
throw e.getOriginalException();
default:
throw new ShouldNeverHappenException(String.format("Unknown TransactionalExecutor.Code: %s", code));
}
} finally {
if (degradeCheck) {
EVENT_BUS.post(new DegradeCheckEvent(succeed));
}
}
}
在這裡我們，主要關注一個重點方法 execute() ，這個方法主要用來執行事務的具體流程：
- 獲取事務資訊
- 執行全域性事務
- 發生異常全域性回滾，各個資料通過UndoLog進行事務補償
- 全域性事務提交
- 清除所有資源
這個位置也是一個非常核心的一個位置，因為我們所有的業務進來以後都會去走這個位置，具體原始碼如下所示：
public Object execute(TransactionalExecutor business) throws Throwable {
// 1. Get transactionInfo
//獲取事務資訊
TransactionInfo txInfo = business.getTransactionInfo();
if (txInfo == null) {
throw new ShouldNeverHappenException("transactionInfo does not exist");
}
// 1.1 Get current transaction, if not null, the tx role is 'GlobalTransactionRole.Participant'.
//獲取當前事務，主要獲取XID
GlobalTransaction tx = GlobalTransactionContext.getCurrent();
// 1.2 Handle the transaction propagation.
//根據設定的不同事務傳播行為，執行不同的邏輯
Propagation propagation = txInfo.getPropagation();
SuspendedResourcesHolder suspendedResourcesHolder = null;
try {
switch (propagation) {
case NOT_SUPPORTED:
// If transaction is existing, suspend it.
if (existingTransaction(tx)) {
suspendedResourcesHolder = tx.suspend();
}
// Execute without transaction and return.
return business.execute();
case REQUIRES_NEW:
// If transaction is existing, suspend it, and then begin new transaction.
if (existingTransaction(tx)) {
suspendedResourcesHolder = tx.suspend();
tx = GlobalTransactionContext.createNew();
}
// Continue and execute with new transaction
break;
case SUPPORTS:
// If transaction is not existing, execute without transaction.
if (notExistingTransaction(tx)) {
return business.execute();
}
// Continue and execute with new transaction
break;
case REQUIRED:
// If current transaction is existing, execute with current transaction,
// else continue and execute with new transaction.
break;
case NEVER:
// If transaction is existing, throw exception.
if (existingTransaction(tx)) {
throw new TransactionException(
String.format("Existing transaction found for transaction marked with propagation 'never', xid = %s"
, tx.getXid()));
} else {
// Execute without transaction and return.
return business.execute();
}
case MANDATORY:
// If transaction is not existing, throw exception.
if (notExistingTransaction(tx)) {
throw new TransactionException("No existing transaction found for transaction marked with propagation 'mandatory'");
}
// Continue and execute with current transaction.
break;
default:
throw new TransactionException("Not Supported Propagation:" + propagation);
}
// 1.3 If null, create new transaction with role 'GlobalTransactionRole.Launcher'.
//如果當前事務為空，建立一個新的事務
if (tx == null) {
tx = GlobalTransactionContext.createNew();
}
// set current tx config to holder
GlobalLockConfig previousConfig = replaceGlobalLockConfig(txInfo);
try {
// 2. If the tx role is 'GlobalTransactionRole.Launcher', send the request of beginTransaction to TC,
// else do nothing. Of course, the hooks will still be triggered.
//開始執行全域性事務
beginTransaction(txInfo, tx);
Object rs;
try {
// Do Your Business
// 執行當前業務邏輯
//1、在TC註冊當前分支事務，TC會在branch_table中插入一條分支事務資料
//2、執行本地update語句，並在執行前後查詢資料狀態，並把資料前後映象存入到undo_log中
//3、遠端呼叫其他應用，遠端應用接收到XID，也會註冊分支事務，寫入branch_table以及本地undo_log表
//4、會在lock_table表中插入全域性鎖資料（一個分支一條）
rs = business.execute();
} catch (Throwable ex) {
// 3. The needed business exception to rollback.
//發生異常全域性回滾，每個事務通過undo_log表進行事務補償
completeTransactionAfterThrowing(txInfo, tx, ex);
throw ex;
}
// 4. everything is fine, commit.
//全域性提交
commitTransaction(tx);
return rs;
} finally {
//5. clear
//清理所有資源
resumeGlobalLockConfig(previousConfig);
triggerAfterCompletion();
cleanUp();
}
} finally {
// If the transaction is suspended, resume it.
if (suspendedResourcesHolder != null) {
tx.resume(suspendedResourcesHolder);
}
}
}
這其中的第三步和第四步其實在向 TC（Seata-Server）發起全域性事務的提交或者回滾，在這裡我們首先關注執行全域性事務的 beginTransaction() 方法
// 向TC發起請求，採用模板模式
private void beginTransaction(TransactionInfo txInfo, GlobalTransaction tx) throws TransactionalExecutor.ExecutionException {
try {
triggerBeforeBegin();
//對TC發起請求
tx.begin(txInfo.getTimeOut(), txInfo.getName());
triggerAfterBegin();
} catch (TransactionException txe) {
throw new TransactionalExecutor.ExecutionException(tx, txe,
TransactionalExecutor.Code.BeginFailure);
}
}
在來關注其中，向TC發起請求的 tx.begin() 方法，而呼叫begin()方法的類為：DefaultGlobalTransaction
@Override
public void begin(int timeout, String name) throws TransactionException {
//判斷呼叫者是否為TM
if (role != GlobalTransactionRole.Launcher) {
assertXIDNotNull();
if (LOGGER.isDebugEnabled()) {
LOGGER.debug("Ignore Begin(): just involved in global transaction [{}]", xid);
}
return;
}
assertXIDNull();
String currentXid = RootContext.getXID();
if (currentXid != null) {
throw new IllegalStateException("Global transaction already exists," +
" can't begin a new global transaction, currentXid = " + currentXid);
}
//獲取XID
xid = transactionManager.begin(null, null, name, timeout);
status = GlobalStatus.Begin;
//繫結XID
RootContext.bind(xid);
if (LOGGER.isInfoEnabled()) {
LOGGER.info("Begin new global transaction [{}]", xid);
}
}
再來看一下 transactionManager.begin() 方法，這個時候使用的是 DefaultTransactionManager.begin 預設的事務管理者，來獲取XID，傳入事務相關的資訊 ，最好TC返回對應的全域性事務XID，它呼叫的是DefaultTransactionManager.begin()方法
public String begin(String applicationId, String transactionServiceGroup, String name, int timeout)
throws TransactionException {
GlobalBeginRequest request = new GlobalBeginRequest();
request.setTransactionName(name);
request.setTimeout(timeout);
//傳送請求得到響應
GlobalBeginResponse response = (GlobalBeginResponse) syncCall(request);
if (response.getResultCode() == ResultCode.Failed) {
throw new TmTransactionException(TransactionExceptionCode.BeginFailed, response.getMsg());
}
//返回XID
return response.getXid();
}
在這裡我們需要關注一個syncCall，在這裡採用的是Netty通訊方式
private AbstractTransactionResponse syncCall(AbstractTransactionRequest request) throws TransactionException {
try {
// 通過Netty傳送請求
return (AbstractTransactionResponse) TmNettyRemotingClient.getInstance().sendSyncRequest(request);
} catch (TimeoutException toe) {
throw new TmTransactionException(TransactionExceptionCode.IO, "RPC timeout", toe);
}
}
具體圖解如下：
在這裡我們需要重點了解 GlobalTransactionScanner 這個型別，在這個型別中繼承了一些介面和抽象類，這個類主要作用就是掃描有註解的Bean，並做AOP增強。
-
ApplicationContextAware：繼承這個型別以後，需要實現其方法setApplicationContext()，當Spring啟動完成以後，會自動呼叫這個型別，將ApplicationContext給bean，也就是說，GlobalTransactionScanner能夠很自然的使用Spring環境 -
InitializingBean： 繼承這個介面，需要實現afterPropertiesSet()，但凡是繼承這個介面的類，在初始化的時候，當所有的properties設定完成以後，會執行這個方法 -
DisposableBean： 這個類，實現了一個destroy()這個方法是在銷燬的時候去呼叫 -
AbstractAutoProxyCreator： 這個類是Spring實現AOP的一種方式，本質上是一個BeanPostProcessor，在Bean初始化至去年，呼叫內部createProxy()，建立一個Bean的AOP代理Bean並返回，對Bean進行增強。
Seata資料來源代理
在上面的環節中，我們講解了Seata AT模式2PC的執行流程，那麼現在我們就來帶大家瞭解一下關於AT資料來源代理的資訊，這也是AT模式中非常關鍵的一個重要知識點，大家可以拿起小本子，記下來。
首先AT模式的核心主要分為一下兩個
- 開啟全域性事務，獲取全域性鎖。
- 解析SQL並寫入undoLog中。
關於第一點我們已經分析清楚了，第二點就是關於AT模式如何解析SQL並寫入undoLog中，但是在這之前，我們需要知道Seata是如何選擇資料來源，並進行資料來源代理的。雖然全域性事務攔截成功後最終還是執行了業務方法進行SQL提交和操作，但是由於Seata對資料來源進行了代理，所以SQL的解析和undoLog的操作，是在資料來源代理中進行完成的。
資料來源代理是Seata中一個非常重要的知識點，在分散式事務執行過程中，undoLog的記錄、資源的鎖定，使用者都是無感知的，因為這些操作都是資料來源的代理中完成了，恰恰是這樣，我們才要去了解，這樣不僅有利於我們瞭解Seata的核心操作，還能對以後原始碼閱讀有所幫助，因為其實很多底層程式碼都會去使用這樣使用者無感知的方式(代理)去實現。
同樣，我們在之前的尋找原始碼入口的時候，通過我們專案中引入的jar找到一個 SeataAutoConfiguration 類，我們在裡面找到一個SeataDataSourceBeanPostProcessor()，這個就是我們資料來源代理的入口方法
我們進入SeataDataSourceBeanPostProcessor類裡面，發現繼承了一個 BeanPostProcessor ,這個介面我們應該很熟悉，這個是Sprng的拓展介面，所有的Bean物件，都有進入兩個方法 postProcessAfterInitialization() 和 postProcessBeforeInitialization() 這兩個方法都是由 BeanPostProcessor提供的，這兩個方法，一個是初始化之前執行Before。一個是在初始化之後執行After，主要用來對比我們的的Bean是否為資料來源代理物件。
在這裡我們需要關注到一個postProcessAfterInitialization.proxyDataSource() 方法，這個裡面
private Object proxyDataSource(Object originBean) {
DataSourceProxy dataSourceProxy = DataSourceProxyHolder.get().putDataSource((DataSource) originBean);
if (this.useJdkProxy) {
return Proxy.newProxyInstance(Thread.currentThread().getContextClassLoader(), SpringProxyUtils.getAllInterfaces(originBean), (proxy, method, args) -> handleMethodProxy(dataSourceProxy, method, args, originBean));
} else {
return Enhancer.create(originBean.getClass(), (MethodInterceptor) (proxy, method, args, methodProxy) -> handleMethodProxy(dataSourceProxy, method, args, originBean));
}
}
這裡有一個DataSourceProxy代理物件，我們需要看的就是這個類，這個就是我們資料庫代理的物件，我們從我們下載的原始碼專案中，搜尋這個代理物件，當我們開啟這個類的目錄時發現，除了這個，還有ConnectionProxy 連線物件、StatementProxy、PreparedStatementProxy SQL執行物件，這些都被Seata進行了代理，為什麼要對這些都進行代理，代理的目的其實為了執行Seata的業務邏輯，生成undoLog，全域性事務的開啟，事務的提交回滾等操作
DataSourceProxy 具體做了什麼，主要功能有哪些，我們來看一下。他在原始碼中是如何體現的，我們需要關注的是init()
public class DataSourceProxy extends AbstractDataSourceProxy implements Resource {
private String resourceGroupId;
private void init(DataSource dataSource, String resourceGroupId) {
//資源組ID，預設是「default」這個預設值
this.resourceGroupId = resourceGroupId;
try (Connection connection = dataSource.getConnection()) {
//根據原始資料來源得到JDBC連線和資料庫型別
jdbcUrl = connection.getMetaData().getURL();
dbType = JdbcUtils.getDbType(jdbcUrl);
if (JdbcConstants.ORACLE.equals(dbType)) {
userName = connection.getMetaData().getUserName();
} else if (JdbcConstants.MARIADB.equals(dbType)) {
dbType = JdbcConstants.MYSQL;
}
} catch (SQLException e) {
throw new IllegalStateException("can not init dataSource", e);
}
initResourceId();
DefaultResourceManager.get().registerResource(this);
if (ENABLE_TABLE_META_CHECKER_ENABLE) {
//如果設定開關開啟，會定時線上程池不斷更新表的後設資料快取資訊
tableMetaExecutor.scheduleAtFixedRate(() -> {
try (Connection connection = dataSource.getConnection()) {
TableMetaCacheFactory.getTableMetaCache(DataSourceProxy.this.getDbType())
.refresh(connection, DataSourceProxy.this.getResourceId());
} catch (Exception ignore) {
}
}, 0, TABLE_META_CHECKER_INTERVAL, TimeUnit.MILLISECONDS);
}
//Set the default branch type to 'AT' in the RootContext.
RootContext.setDefaultBranchType(this.getBranchType());
}
}
從上面我們可以看出，他主要做了以下幾點的增強：
- 給每個資料來源標識資源組ID
- 如果開啟設定，會有一個定時執行緒池定時更新表的後設資料資訊並快取到本地
- 生成代理連線
ConnectionProxy物件
在這三個增強功能裡面，第三個是最重要的，在AT模式裡面，會自動記錄undoLog，資源鎖定，都是通過ConnectionProxy完成的，除此之外 DataSrouceProxy重寫了一個方法 getConnection，因為這裡返回的是一個 ConnectionProxy，而不是原生的Connection
@Override
public ConnectionProxy getConnection() throws SQLException {
Connection targetConnection = targetDataSource.getConnection();
return new ConnectionProxy(this, targetConnection);
}
@Override
public ConnectionProxy getConnection(String username, String password) throws SQLException {
Connection targetConnection = targetDataSource.getConnection(username, password);
return new ConnectionProxy(this, targetConnection);
}
ConnectionProxy
ConnectionProxy 繼承 AbstractConnectionProxy ，在這個父類別中有很多公用的方法，在這個父類別有 PreparedStatementProxy 、StatementProxy 、DataSourceProxy
所以我們需要先來看一下AbstractConnectionProxy，因為這裡封裝了需要我們用到的通用方法和邏輯，在其中我們需要關注的主要在於 PreparedStatementProxy 和 StatementProxy ，在這裡的邏輯主要是資料來源連線的步驟，連線獲取，建立執行物件等等
@Override
public Statement createStatement() throws SQLException {
//呼叫真實連線物件獲取Statement物件
Statement targetStatement = getTargetConnection().createStatement();
//建立Statement的代理
return new StatementProxy(this, targetStatement);
}
@Override
public PreparedStatement prepareStatement(String sql) throws SQLException {
//獲取資料庫型別 mysql/oracle
String dbType = getDbType();
// support oracle 10.2+
PreparedStatement targetPreparedStatement = null;
//如果是AT模式且開啟全域性事務
if (BranchType.AT == RootContext.getBranchType()) {
List<SQLRecognizer> sqlRecognizers = SQLVisitorFactory.get(sql, dbType);
if (sqlRecognizers != null && sqlRecognizers.size() == 1) {
SQLRecognizer sqlRecognizer = sqlRecognizers.get(0);
if (sqlRecognizer != null && sqlRecognizer.getSQLType() == SQLType.INSERT) {
//獲取表的後設資料
TableMeta tableMeta = TableMetaCacheFactory.getTableMetaCache(dbType).getTableMeta(getTargetConnection(),
sqlRecognizer.getTableName(), getDataSourceProxy().getResourceId());
//得到表的主鍵列名
String[] pkNameArray = new String[tableMeta.getPrimaryKeyOnlyName().size()];
tableMeta.getPrimaryKeyOnlyName().toArray(pkNameArray);
targetPreparedStatement = getTargetConnection().prepareStatement(sql,pkNameArray);
}
}
}
if (targetPreparedStatement == null) {
targetPreparedStatement = getTargetConnection().prepareStatement(sql);
}
//建立PreparedStatementProxy代理
return new PreparedStatementProxy(this, targetPreparedStatement, sql);
}
在這兩個代理物件中，都用到了以下幾個方法：
@Override
public ResultSet executeQuery(String sql) throws SQLException {
this.targetSQL = sql;
return ExecuteTemplate.execute(this, (statement, args) -> statement.executeQuery((String) args[0]), sql);
}
@Override
public int executeUpdate(String sql) throws SQLException {
this.targetSQL = sql;
return ExecuteTemplate.execute(this, (statement, args) -> statement.executeUpdate((String) args[0]), sql);
}
@Override
public boolean execute(String sql) throws SQLException {
this.targetSQL = sql;
return ExecuteTemplate.execute(this, (statement, args) -> statement.execute((String) args[0]), sql);
}
在這些方法中都呼叫了 ExecuteTemplate.execute()，所以我們就看一下在 ExecuteTemplate類中具體是做了什麼操作：
public class ExecuteTemplate {
public static <T, S extends Statement> T execute(List<SQLRecognizer> sqlRecognizers,
StatementProxy<S> statementProxy,
StatementCallback<T, S> statementCallback,
Object... args) throws SQLException {
//如果沒有全域性鎖，並且不是AT模式，直接執行SQL
if (!RootContext.requireGlobalLock() && BranchType.AT != RootContext.getBranchType()) {
// Just work as original statement
return statementCallback.execute(statementProxy.getTargetStatement(), args);
}
//得到資料庫型別- mysql/oracle
String dbType = statementProxy.getConnectionProxy().getDbType();
if (CollectionUtils.isEmpty(sqlRecognizers)) {
//sqlRecognizers 為SQL語句的解析器，獲取執行的SQL，通過它可以獲得SQL語句表名、相關的列名、型別等資訊，最後解析出對應的SQL表示式
sqlRecognizers = SQLVisitorFactory.get(
statementProxy.getTargetSQL(),
dbType);
}
Executor<T> executor;
if (CollectionUtils.isEmpty(sqlRecognizers)) {
//如果seata沒有找到合適的SQL語句解析器，那麼便建立簡單執行器PlainExecutor
//PlainExecutor直接使用原生的Statment物件執行SQL
executor = new PlainExecutor<>(statementProxy, statementCallback);
} else {
if (sqlRecognizers.size() == 1) {
SQLRecognizer sqlRecognizer = sqlRecognizers.get(0);
switch (sqlRecognizer.getSQLType()) {
//新增
case INSERT:
executor = EnhancedServiceLoader.load(InsertExecutor.class, dbType,
new Class[]{StatementProxy.class, StatementCallback.class, SQLRecognizer.class},
new Object[]{statementProxy, statementCallback, sqlRecognizer});
break;
//修改
case UPDATE:
executor = new UpdateExecutor<>(statementProxy, statementCallback, sqlRecognizer);
break;
//刪除
case DELETE:
executor = new DeleteExecutor<>(statementProxy, statementCallback, sqlRecognizer);
break;
//加鎖
case SELECT_FOR_UPDATE:
executor = new SelectForUpdateExecutor<>(statementProxy, statementCallback, sqlRecognizer);
break;
//插入加鎖
case INSERT_ON_DUPLICATE_UPDATE:
switch (dbType) {
case JdbcConstants.MYSQL:
case JdbcConstants.MARIADB:
executor =
new MySQLInsertOrUpdateExecutor(statementProxy, statementCallback, sqlRecognizer);
break;
default:
throw new NotSupportYetException(dbType + " not support to INSERT_ON_DUPLICATE_UPDATE");
}
break;
//原生
default:
executor = new PlainExecutor<>(statementProxy, statementCallback);
break;
}
} else {
//批次處理SQL語句
executor = new MultiExecutor<>(statementProxy, statementCallback, sqlRecognizers);
}
}
T rs;
try {
//執行
rs = executor.execute(args);
} catch (Throwable ex) {
if (!(ex instanceof SQLException)) {
// Turn other exception into SQLException
ex = new SQLException(ex);
}
throw (SQLException) ex;
}
return rs;
}
}
在 ExecuteTemplate就一個 execute()，Seata將SQL執行委託給不同的執行器(模板)，Seata提供了6種執行器也就是我們程式碼 case 中（INSERT，UPDATE，DELETE，SELECT_FOR_UPDATE,INSERT_ON_DUPLICATE_UPDATE），這些執行器的父類別都是AbstractDMLBaseExecutor
UpdateExecutor: 執行update語句InsertExecutor: 執行insert語句DeleteExecutor: 執行delete語句SelectForUpdateExecutor: 執行select for update語句PlainExecutor: 執行普通查詢語句MultiExecutor: 複合執行器，在一條SQL語句中執行多條語句
關係圖如下：
然後我們找到 rs = executor.execute(args); 最終執行的方法，找到最頂級的父類別BaseTransactionalExecutor.execute()
@Override
public T execute(Object... args) throws Throwable {
String xid = RootContext.getXID();
if (xid != null) {
//獲取XID
statementProxy.getConnectionProxy().bind(xid);
}
//設定全域性鎖
statementProxy.getConnectionProxy().setGlobalLockRequire(RootContext.requireGlobalLock());
return doExecute(args);
}
在根據doExecute(args);找到其中的重寫方法 AbstractDMLBaseExecutor.doExecute()
@Override
public T doExecute(Object... args) throws Throwable {
AbstractConnectionProxy connectionProxy = statementProxy.getConnectionProxy();
//是否自動提交
if (connectionProxy.getAutoCommit()) {
return executeAutoCommitTrue(args);
} else {
return executeAutoCommitFalse(args);
}
}
對於資料庫而言，本身都是自動提交的，所以我們進入executeAutoCommitTrue()
protected T executeAutoCommitTrue(Object[] args) throws Throwable {
ConnectionProxy connectionProxy = statementProxy.getConnectionProxy();
try {
//設定為手動提交
connectionProxy.changeAutoCommit();
return new LockRetryPolicy(connectionProxy).execute(() -> {
//呼叫手動提交方法，得到分支執行的最終結果
T result = executeAutoCommitFalse(args);
//執行提交
connectionProxy.commit();
return result;
});
} catch (Exception e) {
// when exception occur in finally,this exception will lost, so just print it here
LOGGER.error("execute executeAutoCommitTrue error:{}", e.getMessage(), e);
if (!LockRetryPolicy.isLockRetryPolicyBranchRollbackOnConflict()) {
connectionProxy.getTargetConnection().rollback();
}
throw e;
} finally {
connectionProxy.getContext().reset();
connectionProxy.setAutoCommit(true);
}
}
connectionProxy.changeAutoCommit()方法，修改為手動提交後，我們看來最關鍵的程式碼executeAutoCommitFalse()
protected T executeAutoCommitFalse(Object[] args) throws Exception {
if (!JdbcConstants.MYSQL.equalsIgnoreCase(getDbType()) && isMultiPk()) {
throw new NotSupportYetException("multi pk only support mysql!");
}
//獲取前映象
TableRecords beforeImage = beforeImage();
//執行具體業務
T result = statementCallback.execute(statementProxy.getTargetStatement(), args);
//獲取執行數量
int updateCount = statementProxy.getUpdateCount();
//判斷如果執行數量大於0
if (updateCount > 0) {
//獲取後映象
TableRecords afterImage = afterImage(beforeImage);
//暫存到undolog中，在Commit的時候儲存到資料庫
prepareUndoLog(beforeImage, afterImage);
}
return result;
}
我們再回到executeAutoCommitTrue中，去看看提交做了哪些操作connectionProxy.commit();
@Override
public void commit() throws SQLException {
try {
lockRetryPolicy.execute(() -> {
//具體執行
doCommit();
return null;
});
} catch (SQLException e) {
if (targetConnection != null && !getAutoCommit() && !getContext().isAutoCommitChanged()) {
rollback();
}
throw e;
} catch (Exception e) {
throw new SQLException(e);
}
}
進入到doCommit()中
private void doCommit() throws SQLException {
//判斷是否存在全域性事務
if (context.inGlobalTransaction()) {
processGlobalTransactionCommit();
} else if (context.isGlobalLockRequire()) {
processLocalCommitWithGlobalLocks();
} else {
targetConnection.commit();
}
}
作為分散式事務，一定是存在全域性事務的，所以我們進入 processGlobalTransactionCommit()
private void processGlobalTransactionCommit() throws SQLException {
try {
//註冊分支事務
register();
} catch (TransactionException e) {
recognizeLockKeyConflictException(e, context.buildLockKeys());
}
try {
//寫入資料庫undolog
UndoLogManagerFactory.getUndoLogManager(this.getDbType()).flushUndoLogs(this);
//執行原生提交 一階段提交
targetConnection.commit();
} catch (Throwable ex) {
LOGGER.error("process connectionProxy commit error: {}", ex.getMessage(), ex);
report(false);
throw new SQLException(ex);
}
if (IS_REPORT_SUCCESS_ENABLE) {
report(true);
}
context.reset();
}
其中register()方法就是註冊分支事務的方法，同時還會將undoLog寫入資料庫和執行提交等操作
//註冊分支事務，生成分支事務ID
private void register() throws TransactionException {
if (!context.hasUndoLog() || !context.hasLockKey()) {
return;
}
//註冊分支事務
Long branchId = DefaultResourceManager.get().branchRegister(BranchType.AT, getDataSourceProxy().getResourceId(),
null, context.getXid(), context.getApplicationData(), context.buildLockKeys());
context.setBranchId(branchId);
}
然後我們在回到processGlobalTransactionCommit中，看看寫入資料庫中的flushUndoLogs()
@Override
public void flushUndoLogs(ConnectionProxy cp) throws SQLException {
ConnectionContext connectionContext = cp.getContext();
if (!connectionContext.hasUndoLog()) {
return;
}
//獲取XID
String xid = connectionContext.getXid();
//獲取分支ID
long branchId = connectionContext.getBranchId();
BranchUndoLog branchUndoLog = new BranchUndoLog();
branchUndoLog.setXid(xid);
branchUndoLog.setBranchId(branchId);
branchUndoLog.setSqlUndoLogs(connectionContext.getUndoItems());
UndoLogParser parser = UndoLogParserFactory.getInstance();
byte[] undoLogContent = parser.encode(branchUndoLog);
if (LOGGER.isDebugEnabled()) {
LOGGER.debug("Flushing UNDO LOG: {}", new String(undoLogContent, Constants.DEFAULT_CHARSET));
}
CompressorType compressorType = CompressorType.NONE;
if (needCompress(undoLogContent)) {
compressorType = ROLLBACK_INFO_COMPRESS_TYPE;
undoLogContent = CompressorFactory.getCompressor(compressorType.getCode()).compress(undoLogContent);
}
//寫入資料庫具體位置
insertUndoLogWithNormal(xid, branchId, buildContext(parser.getName(), compressorType), undoLogContent, cp.getTargetConnection());
}
具體寫入方法，此時我們使用的是MySql，所以執行的是MySql實現類MySQLUndoLogManager.insertUndoLogWithNormal()
@Override
protected void insertUndoLogWithNormal(String xid, long branchId, String rollbackCtx, byte[] undoLogContent,
Connection conn) throws SQLException {
insertUndoLog(xid, branchId, rollbackCtx, undoLogContent, State.Normal, conn);
}
//具體寫入操作
private void insertUndoLog(String xid, long branchId, String rollbackCtx, byte[] undoLogContent,
State state, Connection conn) throws SQLException {
try (PreparedStatement pst = conn.prepareStatement(INSERT_UNDO_LOG_SQL)) {
pst.setLong(1, branchId);
pst.setString(2, xid);
pst.setString(3, rollbackCtx);
pst.setBytes(4, undoLogContent);
pst.setInt(5, state.getValue());
pst.executeUpdate();
} catch (Exception e) {
if (!(e instanceof SQLException)) {
e = new SQLException(e);
}
throw (SQLException) e;
}
}
具體流程如下所示：
Seata 伺服器端
我們找到Server.java 這裡就是啟動入口，在這個入口中找到協調者，因為TC整體的操作就是協調整體的全域性事務
//預設協調者
DefaultCoordinator coordinator = DefaultCoordinator.getInstance(nettyRemotingServer);
在DefaultCoordinator類中我們找到 一個doGlobalBegin 這個就是處理全域性事務開始的方法，以及全域性提交 doGlobalCommit 和全域性回滾 doGlobalRollback
//處理全域性事務
@Override
protected void doGlobalBegin(GlobalBeginRequest request, GlobalBeginResponse response, RpcContext rpcContext)
throws TransactionException {
//響應使用者端xid
response.setXid(core.begin(rpcContext.getApplicationId(), rpcContext.getTransactionServiceGroup(),
request.getTransactionName(), request.getTimeout()));
if (LOGGER.isInfoEnabled()) {
LOGGER.info("Begin new global transaction applicationId: {},transactionServiceGroup: {}, transactionName: {},timeout:{},xid:{}",
rpcContext.getApplicationId(), rpcContext.getTransactionServiceGroup(), request.getTransactionName(), request.getTimeout(), response.getXid());
}
}
//處理全域性提交
@Override
protected void doGlobalCommit(GlobalCommitRequest request, GlobalCommitResponse response, RpcContext rpcContext)
throws TransactionException {
MDC.put(RootContext.MDC_KEY_XID, request.getXid());
response.setGlobalStatus(core.commit(request.getXid()));
}
//處理全域性回滾
@Override
protected void doGlobalRollback(GlobalRollbackRequest request, GlobalRollbackResponse response,
RpcContext rpcContext) throws TransactionException {
MDC.put(RootContext.MDC_KEY_XID, request.getXid());
response.setGlobalStatus(core.rollback(request.getXid()));
}
在這裡我們首先關注 doGlobalBegin 中 core.begin()
@Override
public String begin(String applicationId, String transactionServiceGroup, String name, int timeout)
throws TransactionException {
//建立全域性事務Session
GlobalSession session = GlobalSession.createGlobalSession(applicationId, transactionServiceGroup, name,
timeout);
MDC.put(RootContext.MDC_KEY_XID, session.getXid());
//為Session重新增回撥監聽，SessionHolder.getRootSessionManager() 獲取一個全域性Session管理器DataBaseSessionManager
//觀察者設計模式，建立DataBaseSessionManager
session.addSessionLifecycleListener(SessionHolder.getRootSessionManager());
//全域性事務開始
session.begin();
// transaction start event
MetricsPublisher.postSessionDoingEvent(session, false);
return session.getXid();
}
然後我們在來看一下SessionHolder.getRootSessionManager()
/**
* Gets root session manager.
* 獲取一個全域性Session管理器
* @return the root session manager
*/
public static SessionManager getRootSessionManager() {
if (ROOT_SESSION_MANAGER == null) {
throw new ShouldNeverHappenException("SessionManager is NOT init!");
}
return ROOT_SESSION_MANAGER;
}
public static void init(String mode) {
if (StringUtils.isBlank(mode)) {
mode = CONFIG.getConfig(ConfigurationKeys.STORE_SESSION_MODE,
CONFIG.getConfig(ConfigurationKeys.STORE_MODE, SERVER_DEFAULT_STORE_MODE));
}
StoreMode storeMode = StoreMode.get(mode);
//判斷Seata模式，當前為DB
if (StoreMode.DB.equals(storeMode)) {
//通過SPI機制讀取SessionManager介面實現類，讀取的META-INF.services目錄，在通過反射機制建立物件DataBaseSessionManager
ROOT_SESSION_MANAGER = EnhancedServiceLoader.load(SessionManager.class, StoreMode.DB.getName());
........
}
}
在這裡他其實讀取的是DB模式下 io.seata.server.session.SessionManager檔案的內容
我們在回到begin方法中，去檢視session.begin()
@Override
public void begin() throws TransactionException {
//宣告全域性事務開始
this.status = GlobalStatus.Begin;
//開始時間
this.beginTime = System.currentTimeMillis();
//啟用全域性事務
this.active = true;
//將SessionManager放入到集合中，呼叫onBegin方法
for (SessionLifecycleListener lifecycleListener : lifecycleListeners) {
//呼叫父級抽象類的方法
lifecycleListener.onBegin(this);
}
}
這裡我們來看一下 onBegin() 方法，呼叫的是父級的方法，在這其中我們要關注 addGlobalSession() 方法，但是要注意，這裡我們用的是db模式所以呼叫的是db模式的 DateBaseSessionManager
@Override
public void onBegin(GlobalSession globalSession) throws TransactionException {
//這裡呼叫的是DateBaseSessionManager
addGlobalSession(globalSession);
}
@Override
public void addGlobalSession(GlobalSession session) throws TransactionException {
if (StringUtils.isBlank(taskName)) {
//寫入session
boolean ret = transactionStoreManager.writeSession(LogOperation.GLOBAL_ADD, session);
if (!ret) {
throw new StoreException("addGlobalSession failed.");
}
} else {
boolean ret = transactionStoreManager.writeSession(LogOperation.GLOBAL_UPDATE, session);
if (!ret) {
throw new StoreException("addGlobalSession failed.");
}
}
}
然後在看查詢其中關鍵的方法DataBaseTransactionStoreManager.writeSession()
@Override
public boolean writeSession(LogOperation logOperation, SessionStorable session) {
//第一次進入是寫入 會進入當前方法
//全域性新增
if (LogOperation.GLOBAL_ADD.equals(logOperation)) {
return logStore.insertGlobalTransactionDO(SessionConverter.convertGlobalTransactionDO(session));
//全域性修改
} else if (LogOperation.GLOBAL_UPDATE.equals(logOperation)) {
return logStore.updateGlobalTransactionDO(SessionConverter.convertGlobalTransactionDO(session));
//全域性刪除
} else if (LogOperation.GLOBAL_REMOVE.equals(logOperation)) {
return logStore.deleteGlobalTransactionDO(SessionConverter.convertGlobalTransactionDO(session));
//分支新增
} else if (LogOperation.BRANCH_ADD.equals(logOperation)) {
return logStore.insertBranchTransactionDO(SessionConverter.convertBranchTransactionDO(session));
//分支更新
} else if (LogOperation.BRANCH_UPDATE.equals(logOperation)) {
return logStore.updateBranchTransactionDO(SessionConverter.convertBranchTransactionDO(session));
//分支移除
} else if (LogOperation.BRANCH_REMOVE.equals(logOperation)) {
return logStore.deleteBranchTransactionDO(SessionConverter.convertBranchTransactionDO(session));
} else {
throw new StoreException("Unknown LogOperation:" + logOperation.name());
}
}
我們就看第一次進去的方法logStore.insertGlobalTransactionDO(SessionConverter.convertGlobalTransactionDO(session));
@Override
public boolean insertGlobalTransactionDO(GlobalTransactionDO globalTransactionDO) {
String sql = LogStoreSqlsFactory.getLogStoreSqls(dbType).getInsertGlobalTransactionSQL(globalTable);
Connection conn = null;
PreparedStatement ps = null;
try {
int index = 1;
conn = logStoreDataSource.getConnection();
conn.setAutoCommit(true);
ps = conn.prepareStatement(sql);
ps.setString(index++, globalTransactionDO.getXid());
ps.setLong(index++, globalTransactionDO.getTransactionId());
ps.setInt(index++, globalTransactionDO.getStatus());
ps.setString(index++, globalTransactionDO.getApplicationId());
ps.setString(index++, globalTransactionDO.getTransactionServiceGroup());
String transactionName = globalTransactionDO.getTransactionName();
transactionName = transactionName.length() > transactionNameColumnSize ?
transactionName.substring(0, transactionNameColumnSize) :
transactionName;
ps.setString(index++, transactionName);
ps.setInt(index++, globalTransactionDO.getTimeout());
ps.setLong(index++, globalTransactionDO.getBeginTime());
ps.setString(index++, globalTransactionDO.getApplicationData());
return ps.executeUpdate() > 0;
} catch (SQLException e) {
throw new StoreException(e);
} finally {
IOUtil.close(ps, conn);
}
}
在這裡有一個 GlobalTransactionDO 物件，裡面有xid、transactionId 等等，到這裡是不是就很熟悉了、
還記得我們第一次使用Seata的時候會建立三張表
- branch_table 分支事務表
- global_table 全域性事務表
- lock_table 全域性鎖表
而這裡就是對應我們的global_table表，其他兩個也是差不多，都是一樣的操作
流程圖如下：
總結
完整流程圖：
對於Seata原始碼來說主要是瞭解從哪裡入口以及核心點在哪裡，遇到有疑問的，可以Debug，對於Seata AT模式，我們主要掌握的核心點是
- 如何獲取全域性鎖、開啟全域性事務
- 解析SQL並寫入undolog
圍繞這兩點去看的話，會有針對性一點，到這裡我們的Seata原始碼就講解完了，有疑問的小夥伴記得在下方留言。
我是牧小農，怕什麼真理無窮，進一步有進一步的歡喜，大家加油！