逸仙电商Seata企业级落地实践

1. 问题背景

2. 业务介绍

3. 原理分析

4. Demo演示

数据不一致的原因

• 人工补偿数据

• 定时任务检查和补偿数据

原理

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

• 蓝色，Resource Manager（RM），client 端

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

• 前置镜像（Before Image）：保存数据变更前的样子

• 后置镜像（After Image）：保存数据变更后的样子

• Undo Log：保存镜像

有时候新项目接入的时候，有同事会问，为什么事务不生效，如果你也遇到过同样的问题，那首先要检查一下自己的数据源是否已经代理成功。

• file

• db

• redis

以上所有操作都会保证在同一个本地事务中，保证业务操作和 Undo Log 操作的原子性。

一阶段

另外一个需要注意的问题是，如果发现事务不生效，需要检查XID是否成功往下传递。

二阶段提交

• TC 清理全局事务对应的信息

• RM 清理对应 Undo Log 信息

二阶段回滚

反向回滚表示，如果调用链路顺序为 A -> B -> C，那么回滚顺序为 C -> B -> A。

例：A=Insert，B=Update，如果回滚时不按照反向的顺序进行回滚，则有可能出现回滚时先把 A 删除了，再更新 A，引发错误。

• 分支事务注册成功，但是由于网络原因收不到成功的响应，Undo Log 未被持久化；
• 同时全局事务超时(超时时间可自由配置)触发回滚。

读已提交

@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);    //错误关键处，尝试从缓存获取表结构    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;}

修改表结构，需要对应用进行重启，即可解决此问题，非常简单。

@Overridepublic GlobalStatus commit(String xid) throws TransactionException {    //根据xid查询信息，如果开启主从，会有可能导致查询信息不完整    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 {    //根据xid查询信息，如果开启主从，会有可能导致查询信息不完整    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

部署-单节点多应用

部署-异地容灾

# 广州机房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 = ""  }}

# 上海机房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

推荐阅读：

欢迎关注微信公众号：互联网全栈架构，收取更多有价值的信息。