逸仙電商Seata企業(yè)級落地實(shí)踐

1. 問題背景

2. 業(yè)務(wù)介紹

3. 原理分析

4. Demo演示

數(shù)據(jù)不一致的原因

• 人工補(bǔ)償數(shù)據(jù)

• 定時任務(wù)檢查和補(bǔ)償數(shù)據(jù)

原理

• 黃色，Transaction Manager（TM），client 端

• 藍(lán)色，Resource Manager（RM），client 端

• 綠色，Transaction Coordinator（TC），server 端

• 前置鏡像（Before Image）：保存數(shù)據(jù)變更前的樣子

• 后置鏡像（After Image）：保存數(shù)據(jù)變更后的樣子

• Undo Log：保存鏡像

有時候新項目接入的時候，有同事會問，為什么事務(wù)不生效，如果你也遇到過同樣的問題，那首先要檢查一下自己的數(shù)據(jù)源是否已經(jīng)代理成功。

• file

• db

• redis

以上所有操作都會保證在同一個本地事務(wù)中，保證業(yè)務(wù)操作和 Undo Log 操作的原子性。

一階段

另外一個需要注意的問題是，如果發(fā)現(xiàn)事務(wù)不生效，需要檢查XID是否成功往下傳遞。

二階段提交

• TC 清理全局事務(wù)對應(yīng)的信息

• RM 清理對應(yīng) Undo Log 信息

二階段回滾

反向回滾表示，如果調(diào)用鏈路順序為 A -> B -> C，那么回滾順序為 C -> B -> A。

例：A=Insert，B=Update，如果回滾時不按照反向的順序進(jìn)行回滾，則有可能出現(xiàn)回滾時先把 A 刪除了，再更新 A，引發(fā)錯誤。

• 分支事務(wù)注冊成功，但是由于網(wǎng)絡(luò)原因收不到成功的響應(yīng)，Undo Log 未被持久化；
• 同時全局事務(wù)超時(超時時間可自由配置)觸發(fā)回滾。

讀已提交

@GlobalLock@Transactionalpublic PayMoneyDto detail(ProcessOnEventRequestDto processOnEventRequestDto) {    return baseMapper.detail(processOnEventRequestDto.getProcessInfoDto().getBusinessKey())}@Mapperpublic interface PayMoneyMapper extends BaseMapper<PayMoney> {        @Select("select id, name, amount, account, has_repayment, pay_amount from pay_money m where m.business_key = #{businessKey} for update")    PayMoneyDto detail(@Param("businessKey") String businessKey);}

問題

@Overridepublic TableMeta getTableMeta(final Connection connection, final String tableName, String resourceId) {    if (StringUtils.isNullOrEmpty(tableName)) {        throw new IllegalArgumentException("TableMeta cannot be fetched without tableName");    }    TableMeta tmeta;    final String key = getCacheKey(connection, tableName, resourceId);    //錯誤關(guān)鍵處，嘗試從緩存獲取表結(jié)構(gòu)    tmeta = TABLE_META_CACHE.get(key, mappingFunction -> {        try {            return fetchSchema(connection, tableName);        } catch (SQLException e) {            LOGGER.error("get table meta of the table `{}` error: {}", tableName, e.getMessage(), e);            return null;        }    });    if (tmeta == null) {        throw new ShouldNeverHappenException(String.format("[xid:%s]get table meta failed," +                                                           " please check whether the table `%s` exists.", RootContext.getXID(), tableName));    }    return tmeta;}

修改表結(jié)構(gòu)，需要對應(yīng)用進(jìn)行重啟，即可解決此問題，非常簡單。

@Overridepublic GlobalStatus commit(String xid) throws TransactionException {    //根據(jù)xid查詢信息，如果開啟主從，會有可能導(dǎo)致查詢信息不完整    GlobalSession globalSession = SessionHolder.findGlobalSession(xid);    if (globalSession == null) {        return GlobalStatus.Finished;    }    globalSession.addSessionLifecycleListener(SessionHolder.getRootSessionManager());    // just lock changeStatus    boolean shouldCommit = SessionHolder.lockAndExecute(globalSession, () -> {        // Highlight: Firstly, close the session, then no more branch can be registered.        globalSession.closeAndClean();        if (globalSession.getStatus() == GlobalStatus.Begin) {            if (globalSession.canBeCommittedAsync()) {                globalSession.asyncCommit();                return false;            } else {                globalSession.changeStatus(GlobalStatus.Committing);                return true;            }        }        return false;    });    if (shouldCommit) {        boolean success = doGlobalCommit(globalSession, false);        //If successful and all remaining branches can be committed asynchronously, do async commit.        if (success && globalSession.hasBranch() && globalSession.canBeCommittedAsync()) {            globalSession.asyncCommit();            return GlobalStatus.Committed;        } else {            return globalSession.getStatus();        }    } else {        return globalSession.getStatus() == GlobalStatus.AsyncCommitting ? GlobalStatus.Committed : globalSession.getStatus();    }}

@Overridepublic GlobalStatus rollback(String xid) throws TransactionException {    //根據(jù)xid查詢信息，如果開啟主從，會有可能導(dǎo)致查詢信息不完整    GlobalSession globalSession = SessionHolder.findGlobalSession(xid);    if (globalSession == null) {        return GlobalStatus.Finished;    }    globalSession.addSessionLifecycleListener(SessionHolder.getRootSessionManager());    // just lock changeStatus    boolean shouldRollBack = SessionHolder.lockAndExecute(globalSession, () -> {        globalSession.close(); // Highlight: Firstly, close the session, then no more branch can be registered.        if (globalSession.getStatus() == GlobalStatus.Begin) {            globalSession.changeStatus(GlobalStatus.Rollbacking);            return true;        }        return false;    });    if (!shouldRollBack) {        return globalSession.getStatus();    }    doGlobalRollback(globalSession, false);    return globalSession.getStatus();}

相信此問題會在支持 Raft 之后得到完美的解決。

pr: https://github.com/seata/seata/pull/3086

有興趣的朋友也可以嘗試去 review 一下代碼。

部署-高可用

• nacos

• consul

• etcd3

• eureka

• redis

• sofa

• zookeeper

• nacos

• etcd3

• consul

• apollo

• zk

部署-單節(jié)點(diǎn)多應(yīng)用

部署-異地容災(zāi)

# 廣州機(jī)房registry {  # file 、nacos 、eureka、redis、zk、consul、etcd3、sofa  type = "nacos"  loadBalance = "RandomLoadBalance"  loadBalanceVirtualNodes = 10  nacos {    application = "seata-server"    serverAddr = "127.0.0.1:8848"    group = "SEATA_GROUP"    namespace = ""    cluster = "Guangzhou"    username = ""    password = ""  }}

# 上海機(jī)房registry {  # file 、nacos 、eureka、redis、zk、consul、etcd3、sofa  type = "nacos"  loadBalance = "RandomLoadBalance"  loadBalanceVirtualNodes = 10  nacos {    application = "seata-server"    serverAddr = "127.0.0.1:8848"    group = "SEATA_GROUP"    namespace = ""    cluster = "Shanghai"    username = ""    password = ""  }}

Demo

https://start.aliyun.com

https://start.aliyun.com/handson/isnEO76f/distributedtransaction

推薦閱讀：

歡迎關(guān)注微信公眾號：互聯(lián)網(wǎng)全棧架構(gòu)，收取更多有價值的信息。