Seata 分布式事务原理：AT 模式与全局锁

摘要

在微服务架构中，分布式事务是绕不过的难题。Seata 作为阿里巴巴开源的分布式事务解决方案，提供了 AT、TCC、SAGA 和 XA 四种模式。本文深入剖析 Seata 1.7.0 的 AT 模式（Auto Transaction）原理，通过源码分析和实战示例，揭示其如何通过"自动分支注册"和"全局锁机制"实现高性能的分布式事务一致性。

---

一、分布式事务的挑战与 Seata 架构

1.1 分布式事务的困境

在单体应用中，我们使用 ACID 事务保证数据一致性；但在微服务架构中，业务被拆分为多个独立服务，每个服务拥有独立的数据库，传统的本地事务（@Transactional）失效。

1.2 Seata 整体架构

Seata 采用 TAAC（Transaction As A Coordinator）架构，包含三个核心组件：

组件	角色	职责
TC	协调者	维护全局事务状态，驱动全局提交或回滚
TM	发起者	定义全局事务范围，发起提交或回滚
RM	参与者	管理分支事务，向 TC 注册分支、汇报状态

---

二、AT 模式核心原理

2.1 AT 模式 vs TCC vs SAGA vs XA

模式	一致性	性能	代码侵入	适用场景
AT	最终一致	⭐⭐⭐⭐⭐	无（@GlobalTransactional）	常见业务，基于关系型DB
TCC	强一致	⭐⭐⭐⭐	高（三阶段接口）	核心业务，性能要求高
SAGA	最终一致	⭐⭐⭐⭐	中（状态机定义）	长事务，业务流程复杂
XA	强一致	⭐⭐	无（JTA 标准）	对性能要求不高的场景

2.2 AT 模式两阶段提交

AT 模式的核心是"增强型两阶段提交"，通过自动生成 undo log 实现无侵入的分布式事务。

2.3 为什么 AT 模式性能高？

1. 第一阶段本地提交：不锁表，不阻塞读写
2. 第二阶段异步：提交时异步删除 undo log，回滚时才并发执行
3. 自动生成 undo log：无需手动编写补偿逻辑

---

三、源码分析：AT 模式实现

3.1 DataSource 代理机制

源码位置：seata-src/sqlparser/seata-sqlparser-core/src/main/java/io/seata/sqlparser/

// seata-jdbc/src/main/java/io/seata/rm/datasource/DataSourceProxy.java

public class DataSourceProxy extends AbstractDataSourceProxy {
    
    @Override
    public ConnectionProxy getConnection() throws SQLException {
        // 获取目标数据源连接
        Connection targetConnection = targetDataSource.getConnection();
        
        // 包装为代理连接
        return new ConnectionProxy(targetConnection, this);
    }
}

// seata-jdbc/src/main/java/io/seata/rm/datasource/ConnectionProxy.java

public class ConnectionProxy extends AbstractConnectionProxy {
    
    @Override
    public PreparedStatement prepareStatement(String sql) throws SQLException {
        // 判断是否为 SELECT 语句
        if (SQLRecognizerFactory.get(sql).getSQLType() == SQLType.SELECT) {
            // SELECT 语句不代理
            return targetConnection.prepareStatement(sql);
        }
        
        // INSERT/UPDATE/DELETE 创建 PreparedStatementProxy
        return new PreparedStatementProxy(targetConnection, sql);
    }
}

3.2 SQL 解析与镜像生成

源码位置：seata-src/core/src/main/java/io/seata/core/context/

// seata-rm-datasource/src/main/java/io/seata/rm/datasource/exec/AbstractDMLBaseExecutor.java

protected T executeAutoCommitTrue(Object[] args) throws Throwable {
    // 1. 解析 SQL 类型
    SQLRecognizer recognizer = SQLRecognizerFactory.get(statementSQL);
    
    // 2. 查询 before image
    TableRecords beforeImage = beforeImage();
    
    // 3. 执行业务 SQL
    T result = statementCallback.execute(statementProxy.getTargetStatement(), args);
    
    // 4. 查询 after image
    TableRecords afterImage = afterImage(beforeImage);
    
    // 5. 准备 undo log
    prepareUndoLog(beforeImage, afterImage);
    
    return result;
}

// before image 生成逻辑
protected TableRecords beforeImage() throws SQLException {
    // 构建查询 SQL（主键条件）
    SQLSelectRecognizer visitor = SQLVisitorFactory.get(sql, dbType);
    StringBuilder beforeImageSql = new StringBuilder("SELECT * FROM ");
    beforeImageSql.append(tableName).append(" WHERE ");
    beforeImageSql.append(visitor.getWhereCondition());
    
    // 执行查询
    PreparedStatement pst = targetConnection.prepareStatement(beforeImageSql.toString());
    ResultSet rs = pst.executeQuery();
    
    // 构建 TableRecords
    return TableRecords.buildRecords(rs);
}

3.3 undo log 结构

源码位置：seata-rm-datasource/src/main/java/io/seata/rm/datasource/undo/

// seata-rm-datasource/src/main/java/io/seata/rm/datasource/undo/UndoLog.java

public class UndoLog {
    private Long id;                      // undo log 主键
    private Long branchId;                // 分支事务 ID
    private String xid;                   // 全局事务 ID
    private String context;               // 序列化的 undo 内容
    private Integer rollbackInfo;          // 回滚信息
    private LogStatus status;             // 状态（0:正常，1:已完成）
    private Date createTime;              // 创建时间
}

// seata-rm-datasource/src/main/java/io/seata/rm/datasource/undo/UndoLogManager.java

public class UndoLogManager {
    
    // 插入 undo log
    public static void insertUndoLog(String xid, long branchId, SQLUndoLog sqlUndoLog) {
        // 1. 序列化 undo log
        byte[] undoLogContent = serializer.serialize(sqlUndoLog);
        
        // 2. 构建 INSERT 语句
        String sql = "INSERT INTO undo_log " +
                     "(branch_id, xid, context, rollback_info, log_status, created) " +
                     "VALUES (?, ?, ?, ?, ?, now())";
        
        // 3. 执行插入
        PreparedStatement pst = connection.prepareStatement(sql);
        pst.setLong(1, branchId);
        pst.setString(2, xid);
        pst.setBytes(3, undoLogContent);
        pst.setInt(4, 0); // LogStatus.Normal
        pst.executeUpdate();
    }
}

---

四、全局锁机制

4.1 为什么需要全局锁？

问题场景：脏写问题

解决方案：全局锁

4.2 全局锁获取流程

4.3 全局锁源码

源码位置：seata-rm-datasource/src/main/java/io/seata/rm/datasource/

// seata-rm-datasource/src/main/java/io/seata/rm/datasource/ConnectionProxy.java

private boolean acquireLock(String lockKey) {
    // 1. 构建锁请求
    LockQueryRequest request = new LockQueryRequest();
    request.setXid(xid);
    request.setLockKey(lockKey);
    
    // 2. 向 TC 申请全局锁
    LockQueryResponse response = lockManager.acquireLock(request);
    
    // 3. 判断是否获取成功
    if (response.isLockable()) {
        return true;
    }
    
    // 4. 获取失败，回滚本地事务
    throw new LockConflictException("全局锁获取失败");
}

// seata-server/src/main/java/io/seata/server/lock/LockManager.java

public boolean acquireLock(LockQueryRequest request) {
    String xid = request.getXid();
    String lockKey = request.getLockKey();
    
    // 1. 检查是否已被其他全局事务锁定
    if (lockStore.isLocked(lockKey) && !lockStore.getOwner(lockKey).equals(xid)) {
        return false; // 锁已被占用
    }
    
    // 2. 加锁
    lockStore.lock(lockKey, xid);
    
    return true;
}

4.4 全局锁表结构

CREATE TABLE `distributed_lock` (
  `lock_key` varchar(128) NOT NULL COMMENT '锁键',
  `xid` varchar(128) NOT NULL COMMENT '全局事务ID',
  `transaction_id` bigint(20) DEFAULT NULL COMMENT '事务ID',
  `branch_id` bigint(20) NOT NULL COMMENT '分支ID',
  `resource_id` varchar(256) DEFAULT NULL COMMENT '资源ID',
  `table_name` varchar(64) DEFAULT NULL COMMENT '表名',
  `pk` varchar(128) DEFAULT NULL COMMENT '主键',
  `gmt_create` datetime DEFAULT CURRENT_TIMESTAMP,
  `gmt_modified` datetime DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
  PRIMARY KEY (`lock_key`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

---

五、实战：Spring Boot + Seata 实现 AT 模式

5.1 业务场景

电商下单：扣减库存 → 创建订单 → 扣减账户余额

// 业务接口
@GlobalTransactional // Seata 全局事务注解
public void purchase(String userId, String commodityCode, int orderCount) {
    // 1. 扣减库存
    storageFeign.deduct(commodityCode, orderCount);
    
    // 2. 创建订单
    orderFeign.create(userId, commodityCode, orderCount);
    
    // 3. 扣减余额
    accountFeign.deduct(userId, orderCount * 100);
}

5.2 依赖配置

<!-- pom.xml -->
<dependency>
    <groupId>com.alibaba.cloud</groupId>
    <artifactId>spring-cloud-starter-alibaba-seata</artifactId>
    <version>2022.0.0.0</version>
</dependency>

<dependency>
    <groupId>io.seata</groupId>
    <artifactId>seata-spring-boot-starter</artifactId>
    <version>1.7.0</version>
</dependency>

5.3 数据源代理配置

# application.yml
seata:
  enabled: true
  application-id: order-service
  tx-service-group: my_test_tx_group
  service:
    vgroup-mapping:
      my_test_tx_group: default
  registry:
    type: nacos
    nacos:
      server-addr: 127.0.0.1:8848
      namespace: seata
      group: SEATA_GROUP
      application: seata-server
  config:
    type: nacos
    nacos:
      server-addr: 127.0.0.1:8848
      namespace: seata
      group: SEATA_GROUP

// DataSourceProxyConfig.java
@Configuration
public class DataSourceProxyConfig {
    
    @Bean
    @ConfigurationProperties(prefix = "spring.datasource.druid.master")
    public DataSource dataSource() {
        DruidDataSource dataSource = new DruidDataSource();
        return dataSource;
    }
    
    @Bean
    public DataSourceProxy dataSourceProxy(DataSource dataSource) {
        return new DataSourceProxy(dataSource); // Seata 数据源代理
    }
    
    @Bean
    public SqlSessionFactory sqlSessionFactory(DataSourceProxy dataSourceProxy) throws Exception {
        SqlSessionFactoryBean factoryBean = new SqlSessionFactoryBean();
        factoryBean.setDataSource(dataSourceProxy);
        return factoryBean.getObject();
    }
}

5.4 undo log 表

-- 每个业务数据库都需要创建 undo_log 表
CREATE TABLE `undo_log` (
  `id` bigint(20) NOT NULL AUTO_INCREMENT,
  `branch_id` bigint(20) NOT NULL,
  `xid` varchar(100) NOT NULL,
  `context` varchar(128) NOT NULL,
  `rollback_info` longblob NOT NULL,
  `log_status` int(11) NOT NULL,
  `log_created` datetime NOT NULL,
  `log_modified` datetime NOT NULL,
  `ext` varchar(100) DEFAULT NULL,
  PRIMARY KEY (`id`),
  UNIQUE KEY `ux_undo_log` (`xid`,`branch_id`)
) ENGINE=InnoDB AUTO_INCREMENT=1 DEFAULT CHARSET=utf8mb4;

---

六、回滚流程与日志清理

6.1 全局回滚流程

回滚 SQL 生成：

// seata-rm-datasource/src/main/java/io/seata/rm/datasource/undo/UndoExecutor.java

public String buildUndoSQL() {
    // 根据 SQL 类型构建回滚语句
    switch (sqlType) {
        case INSERT:
            // INSERT → DELETE
            return "DELETE FROM " + tableName + " WHERE " + buildWhereCondition(afterImage);
            
        case UPDATE:
            // UPDATE → UPDATE (set before image)
            return "UPDATE " + tableName + " SET " + buildSetClause(beforeImage) + 
                   " WHERE " + buildWhereCondition(afterImage);
            
        case DELETE:
            // DELETE → INSERT
            return "INSERT INTO " + tableName + " (" + buildColumns(beforeImage) + 
                   ") VALUES (" + buildValues(beforeImage) + ")";
            
        default:
            throw new UnsupportedOperationException("不支持的 SQL 类型");
    }
}

6.2 日志清理策略

策略	触发条件	实现方式
自动清理	第二阶段提交成功	异步删除 undo log
定时清理	cron 任务	log_status=1 且超过 7 天
手动清理	运维操作	DELETE FROM undo_log WHERE log_status=1

---

七、性能优化与最佳实践

7.1 性能调优参数

参数	默认值	推荐值	说明
client.rm.lock.retryInterval	10	30	全局锁重试间隔
client.rm.lock.retryTimes	30	100	全局锁重试次数
client.rm.reportSuccessCount	5	1	成功上报次数
client.rm.asyncCommitBufferLimit	10000	5000	异步提交缓冲区大小

7.2 避免长事务

// ❌ 错误：长事务（持有全局锁时间过长）
@GlobalTransactional
public void badLongTransaction() {
    // 业务逻辑 1
    repository.update(data);
    
    // 调用外部接口（耗时 10 秒）
    externalApi.call();
    
    // 业务逻辑 2
    repository.update(data2);
}

// ✅ 正确：拆分短事务
public void goodShortTransaction() {
    // 业务逻辑 1
    transaction1();
    
    // 外部接口在事务外
    externalApi.call();
    
    // 业务逻辑 2
    transaction2();
}

7.3 全局锁超时配置

# seata.conf
client {
    rm {
        lock {
            retryInterval = 30  # 重试间隔
            retryTimes = 100    # 重试次数
            retryPolicyBranchRollbackOnConflict = true # 冲突时回滚分支
        }
    }
}

---

八、版本演进：Seata 1.7.0 新特性

8.1 优化 undo log 序列化

// Seata 1.7.0 引入 Kryo 序列化器（替代 JDK 序列化）
public class KryoSerializer implements Serializer {
    
    @Override
    public <T> byte[] serialize(T obj) {
        // Kryo 性能是 JDK 序列化的 10 倍
        try (Output output = new Output(new ByteArrayOutputStream())) {
            kryo.writeClassAndObject(output, obj);
            return output.toBytes();
        }
    }
}

8.2 支持 PostgreSQL 数据库

# seata 1.7.0 支持 PostgreSQL AT 模式
datasource:
  url: jdbc:postgresql://localhost:5432/seata
  driverClassName: org.postgresql.Driver
  dbType: postgresql

---

九、总结

Seata AT 模式通过以下机制实现高性能分布式事务：

1. 自动生成 undo log：无需手动编写补偿逻辑
2. 增强型两阶段提交：第一阶段本地提交，第二阶段异步
3. 全局锁机制：防止脏写，保证数据一致性
4. 轻量级框架：无业务代码侵入，易于集成

适用场景：基于关系型数据库的常见业务场景，对性能有较高要求。

不适用场景：非关系型数据库（Redis、MongoDB）、跨数据库类型事务。

---

参考资料

• Seata 官方文档：https://seata.io/zh-cn/
• Seata GitHub 源码（1.7.0）：https://github.com/seata/seata/tree/v1.7.0
• 《深入理解 Seata》第3章：AT 模式原理

---

标签：Seata,分布式事务,AT模式,全局锁,微服务