2024-08-07

分布式 / 分布式事务

36 分钟读完 (大约 5456 个字) 0次访问

240807-分布式事务解决方案Seata之AT事务

Seata AT事务模式详解

Seata AT事务模式详解

AT（Automatic Transaction）模式是 Seata 提供的零侵入分布式事务解决方案，无需修改业务代码，仅需添加注解即可实现分布式事务。

一、AT 模式核心概念

1.1 全局事务与分支事务

在 Seata 的 AT 模式中，理解分布式事务生命周期需要掌握两个核心概念：

全局事务 (Global Transaction):

是整个业务链路（跨多个微服务）
由 TM（事务管理器）发起和管理
拥有全局唯一的 XID（事务 ID）

分支事务 (Branch Transaction):

是单个微服务内的本地数据库操作
自动注册到全局事务
拥有 Branch ID 标识

全局事务 (XID: xxx)
    ↓
├─ 分支事务 1 (订单服务 → 订单 DB)
├─ 分支事务 2 (库存服务 → 库存 DB)
├─ 分支事务 3 (账户服务 → 账户 DB)
└─ 分支事务 4 (积分服务 → 积分 DB)

1.2 两阶段提交模型

AT 模式采用改进的两阶段提交（2PC）模型：

阶段一（Phase 1）:

拦截业务 SQL，解析语义
查询数据前镜像（Before Image）
执行业务 SQL
查询数据后镜像（After Image）
保存 undo_log 回滚日志
提交本地事务（立即释放锁）
向 TC 汇报执行结果

阶段二（Phase 2）:

成功提交: 异步批量删除 undo_log（快速返回）
失败回滚: 根据 undo_log 前镜像恢复数据

┌─────────────────────────────────────────────┐
│          Phase 1: 执行并记录                 │
│  查询前镜像 → 执行 SQL → 查询后镜像          │
│  → 保存 undo_log → 提交本地事务              │
└─────────────────────────────────────────────┘
                    ↓
        ┌───────┴───────┐
        ↓               ↓
   成功则提交       失败则回滚
   (删除 undo_log)  (根据 undo_log 恢复)

二、自动注册原理

2.1 数据源代理技术

Seata 通过 DataSourceProxy（数据源代理）和 SQL 拦截技术实现分支事务的自动注册。

核心组件:

DataSourceProxy: 代理真实数据源
ConnectionProxy: 代理数据库连接
StatementProxy: 代理 SQL 执行
ExecuteTemplate: SQL 执行模板

2.2 自动注册流程

1. 业务代码执行 SQL
   ↓
   调用 mapper.insert() 或 jdbcTemplate.update()
   
2. Seata 拦截 SQL 执行
   ↓
   DataSourceProxy 拦截数据库连接
   检查 RootContext 中是否有 XID
   
3. 自动注册分支事务
   ↓
   有 XID → 向 TC 注册分支事务，获取 Branch ID
   无 XID → 普通本地事务，直接执行
   
4. 执行 SQL 并记录日志
   ↓
   查询前镜像 → 执行业务 SQL → 查询后镜像
   保存 undo_log → 提交本地事务
   
5. 汇报执行结果
   ↓
   向 TC 汇报 Phase 1 执行成功

2.3 XID 传递机制

XID 是什么:

全局事务的唯一标识符
格式：{ip}:{port}:{transactionId}
例如：192.168.1.100:8091:1234567890

XID 传递路径:

入口服务 (@GlobalTransactional)
   ↓ RootContext.bind(xid)
Spring Cloud OpenFeign / Dubbo
   ↓ HTTP Header / RPC Context
下游服务 1
   ↓ RootContext.bind(xid)
下游服务 2

注意事项:

✅ Spring Cloud/Dubbo 默认支持 XID 透传
⚠️ 自定义线程需手动传递 XID
⚠️ 消息队列需手动绑定 XID

三、快速开始

3.1 数据库初始化

在各个微服务的数据库中创建 undo_log 表：

-- AT 模式必须创建 undo_log 表
CREATE TABLE IF NOT EXISTS `undo_log` (
  `id` BIGINT(20) NOT NULL AUTO_INCREMENT COMMENT '主键',
  `branch_id` BIGINT(20) NOT NULL COMMENT '分支事务 ID',
  `xid` VARCHAR(100) NOT NULL COMMENT '全局事务 ID',
  `context` VARCHAR(128) NOT NULL COMMENT '上下文信息',
  `rollback_info` LONGBLOB NOT NULL COMMENT '回滚日志',
  `log_status` INT(11) NOT NULL COMMENT '日志状态',
  `log_created` DATETIME NOT NULL COMMENT '创建时间',
  `log_modified` DATETIME NOT NULL COMMENT '修改时间',
  `ext` VARCHAR(100) DEFAULT NULL COMMENT '扩展字段',
  PRIMARY KEY (`id`),
  UNIQUE KEY `ux_undo_log` (`xid`, `branch_id`) USING BTREE
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='AT事务回滚日志表';

表结构说明:

字段	类型	说明
`branch_id`	BIGINT	分支事务 ID
`xid`	VARCHAR	全局事务 ID
`context`	VARCHAR	序列化方式等上下文
`rollback_info`	LONGBLOB	前后镜像数据
`log_status`	INT	0-正常，1-防御性回滚
`log_created`	DATETIME	创建时间
`log_modified`	DATETIME	修改时间

3.2 添加依赖

各微服务项目引入 Seata 依赖：

Maven:

<dependencies>
    <!-- Spring Cloud Alibaba Seata -->
    <dependency>
        <groupId>com.alibaba.cloud</groupId>
        <artifactId>spring-cloud-starter-alibaba-seata</artifactId>
        <version>2021.0.1.0</version>
    </dependency>
    
    <!-- MyBatis Plus (可选，如果使用的话) -->
    <dependency>
        <groupId>com.baomidou</groupId>
        <artifactId>mybatis-plus-boot-starter</artifactId>
        <version>3.5.2</version>
    </dependency>
</dependencies>

Gradle:

1
2
3

dependencies {
    implementation 'com.alibaba.cloud:spring-cloud-starter-alibaba-seata:2021.0.1.0'
}

3.3 配置文件

application.yml:

seata:
  enabled: true
  tx-service-group: default_tx_group  # 事务组名称
  service:
    vgroup-mapping:
      default_tx_group: DEFAULT  # 映射到 Seata 集群
    grouplist:
      DEFAULT: 192.168.1.100:8091  # Seata Server 地址
  registry:
    type: nacos
    nacos:
      server-addr: laosan.xin:8848
      namespace: dev-wdg
      group: DEFAULT_GROUP
  config:
    type: nacos
    nacos:
      server-addr: laosan.xin:8848
      namespace: dev-wdg
      group: DEFAULT_GROUP
      data-id: seataService.properties

mybatis-plus:
  configuration:
    map-underscore-to-camel-case: true
    # 关闭 MyBatis 自身的事务管理器，使用 Seata
    default-executor-type: simple

3.4 开启全局事务

在业务入口类添加 @GlobalTransactional 注解：

@Service
public class OrderServiceImpl implements OrderService {
    
    @Autowired
    private OrderMapper orderMapper;
    
    @Autowired
    private InventoryService inventoryService;
    
    @Autowired
    private AccountService accountService;
    
    /**
     * 创建订单（分布式事务入口）
     */
    @Override
    @GlobalTransactional(timeoutMills = 300000, name = "create-order-tx")
    public void createOrder(OrderDTO orderDTO) {
        // 1. 创建订单
        OrderPO order = convertToPO(orderDTO);
        orderMapper.insert(order);
        
        // 2. 扣减库存（自动注册分支事务）
        inventoryService.decreaseStock(
            orderDTO.getProductId(), 
            orderDTO.getCount()
        );
        
        // 3. 扣减余额（自动注册分支事务）
        accountService.decreaseBalance(
            orderDTO.getUserId(), 
            orderDTO.getAmount()
        );
        
        // 如果任何一步抛出异常，所有操作都会回滚
    }
    
    private OrderPO convertToPO(OrderDTO dto) {
        OrderPO po = new OrderPO();
        BeanUtils.copyProperties(dto, po);
        return po;
    }
}

注解参数说明:

参数	类型	默认值	说明
`timeoutMills`	long	60000	超时时间（毫秒）
`name`	String	“”	事务名称（便于监控）
`rollbackFor`	Class[]	Exception.class	回滚异常类型
`noRollbackFor`	Class[]	-	不回滚异常类型

3.5 下游服务实现

下游服务无需添加注解，自动参与分布式事务：

@Service
public class InventoryServiceImpl implements InventoryService {
    
    @Autowired
    private InventoryMapper inventoryMapper;
    
    /**
     * 扣减库存
     * 无需 @GlobalTransactional，自动注册为分支事务
     */
    @Override
    public void decreaseStock(Long productId, Integer count) {
        int updated = inventoryMapper.decreaseStock(productId, count);
        if (updated == 0) {
            throw new RuntimeException("库存不足");
        }
    }
}

@Service
public class AccountServiceImpl implements AccountService {
    
    @Autowired
    private AccountMapper accountMapper;
    
    /**
     * 扣减余额
     * 无需 @GlobalTransactional，自动注册为分支事务
     */
    @Override
    public void decreaseBalance(Long userId, BigDecimal amount) {
        int updated = accountMapper.decreaseBalance(userId, amount);
        if (updated == 0) {
            throw new RuntimeException("余额不足");
        }
    }
}

3.6 测试验证

单元测试

@SpringBootTest
public class OrderServiceTest {
    
    @Autowired
    private OrderService orderService;
    
    @Autowired
    private OrderMapper orderMapper;
    
    @Autowired
    private InventoryMapper inventoryMapper;
    
    @Test
    public void testCreateOrderSuccess() {
        // 准备数据
        OrderDTO orderDTO = new OrderDTO();
        orderDTO.setOrderId(10001L);
        orderDTO.setUserId(888L);
        orderDTO.setProductId(2001L);
        orderDTO.setCount(2);
        orderDTO.setAmount(new BigDecimal("199.00"));
        
        // 执行（应该成功）
        assertDoesNotThrow(() -> {
            orderService.createOrder(orderDTO);
        });
        
        // 验证订单已创建
        assertNotNull(orderMapper.selectById(10001L));
        
        // 验证库存已扣减
        Integer stock = inventoryMapper.selectById(2001L).getStock();
        assertEquals(98, stock); // 假设原库存 100
    }
    
    @Test
    public void testCreateOrderRollback() {
        // 准备数据
        OrderDTO orderDTO = new OrderDTO();
        orderDTO.setOrderId(10002L);
        orderDTO.setUserId(888L);
        orderDTO.setProductId(2001L);
        orderDTO.setCount(1000); // 超出库存
        
        // 执行（应该失败并回滚）
        assertThrows(RuntimeException.class, () -> {
            orderService.createOrder(orderDTO);
        });
        
        // 验证订单未创建
        assertNull(orderMapper.selectById(10002L));
        
        // 验证库存未扣减（回滚成功）
        Integer stock = inventoryMapper.selectById(2001L).getStock();
        assertEquals(100, stock); // 库存保持不变
    }
}

查看 Seata Server 日志

启动测试后，查看 Seata Server 日志：

1 2	# 查看事务日志 tail -f /opt/seata/logs/seata/seata-server.log

关键日志:

INFO  - Global transaction begin, xid: 192.168.1.100:8091:1234567890
INFO  - Branch register to tc, xid: 192.168.1.100:8091:1234567890, branchId: 1
INFO  - Branch register to tc, xid: 192.168.1.100:8091:1234567890, branchId: 2
INFO  - Branch register to tc, xid: 192.168.1.100:8091:1234567890, branchId: 3
INFO  - Global transaction commit, xid: 192.168.1.100:8091:1234567890

Seata Server 日志

四、常见问题排查

4.1 分支事务未自动注册

现象: TC 上没有分支事务记录

可能原因:

1. 手动创建了 DataSource

问题代码:

@Configuration
public class DataSourceConfig {
    
    @Bean
    public DataSource dataSource() {
        // ❌ 错误：直接返回 HikariDataSource
        return new HikariDataSource();
    }
}

解决方案:

@Configuration
public class DataSourceConfig {
    
    @Bean
    public DataSource dataSource() {
        // ✅ 正确：使用 DataSourceProxy 包装
        DataSource dataSource = new HikariDataSource();
        return new DataSourceProxy(dataSource);
    }
}

最佳实践:

使用 Spring Boot 自动配置，无需手动创建 DataSource
如必须手动创建，确保包装为 DataSourceProxy

2. 多数据源场景

问题描述: 使用 DynamicDataSource 切换数据源时，部分数据源未注册

解决方案:

@Configuration
public class MultiDataSourceConfig {
    
    @Bean
    public DataSource dynamicDataSource() {
        // 创建多个数据源
        DataSource ds1 = createDataSource("ds1");
        DataSource ds2 = createDataSource("ds2");
        
        // 包装为 DataSourceProxy
        DataSourceProxy proxy1 = new DataSourceProxy(ds1);
        DataSourceProxy proxy2 = new DataSourceProxy(ds2);
        
        // 配置动态数据源
        DynamicDataSource dynamicDataSource = new DynamicDataSource();
        Map<Object, Object> targetDataSources = new HashMap<>();
        targetDataSources.put("ds1", proxy1);
        targetDataSources.put("ds2", proxy2);
        dynamicDataSource.setTargetDataSources(targetDataSources);
        
        return dynamicDataSource;
    }
    
    private DataSource createDataSource(String name) {
        HikariDataSource ds = new HikariDataSource();
        ds.setJdbcUrl("jdbc:mysql://localhost:3306/" + name);
        ds.setUsername("root");
        ds.setPassword("password");
        return ds;
    }
}

3. XID 丢失（最常见）

场景 1: 子线程中执行

❌ 错误示例:

@GlobalTransactional
public void createOrder(OrderDTO orderDTO) {
    // 主线程有 XID
    
    // ❌ 新线程没有 XID 上下文
    new Thread(() -> {
        inventoryService.decreaseStock(); // 不会注册分支事务
    }).start();
}

✅ 正确示例:

@GlobalTransactional
public void createOrder(OrderDTO orderDTO) {
    // 手动传递 XID
    String xid = RootContext.getXID();
    
    new Thread(() -> {
        try {
            // 绑定 XID 到新线程
            RootContext.bind(xid);
            inventoryService.decreaseStock(); // 会注册分支事务
        } finally {
            // 清理 XID
            RootContext.unbind();
        }
    }).start();
}

场景 2: Feign/Dubbo 调用未透传 XID

检查点:

✅ Spring Cloud OpenFeign 默认支持 XID 透传
✅ Dubbo 默认支持 XID 透传
⚠️ 自定义 HTTP 客户端需手动传递

自定义 Feign 配置:

@Configuration
public class FeignConfig {
    
    @Bean
    public RequestInterceptor requestInterceptor() {
        return template -> {
            String xid = RootContext.getXID();
            if (xid != null) {
                // 将 XID 放入 HTTP Header
                template.header(RootContext.KEY_XID, xid);
            }
        };
    }
}

场景 3: 消息队列消费端

❌ 错误示例:

@RocketMQMessageListener(topic = "order-topic", ...)
public class OrderConsumer implements RocketMQListener<OrderDTO> {
    
    @Override
    public void onMessage(OrderDTO orderDTO) {
        // ❌ 没有绑定 XID，不会注册分支事务
        inventoryService.decreaseStock();
    }
}

✅ 正确示例:

@RocketMQMessageListener(topic = "order-topic", ...)
public class OrderConsumer implements RocketMQListener<OrderDTO> {
    
    @Autowired
    private TransactionalTemplate transactionalTemplate;
    
    @Override
    public void onMessage(OrderDTO orderDTO) {
        // 从消息头获取 XID
        String xid = getMessageHeaderXid();
        
        // 使用事务模板执行
        transactionalTemplate.execute(new TransactionalExecutor() {
            @Override
            public Object execute() throws Throwable {
                inventoryService.decreaseStock();
                return null;
            }
            
            @Override
            public UserTransaction getTransactionInfo() {
                // 设置 XID
                return new UserTransaction(xid, 30000);
            }
        });
    }
    
    private String getMessageHeaderXid() {
        // 从消息头提取 XID
        return MessageConst.PROPERTY_TRANSACTION_ID;
    }
}

4.2 事务不回滚

现象: 异常抛出但数据未回滚

可能原因:

1. 异常类型不匹配

❌ 错误示例:

@GlobalTransactional(rollbackFor = Exception.class)
public void createOrder() {
    // 抛出 RuntimeException，会回滚
    throw new RuntimeException("失败");
}

@GlobalTransactional // 默认只回滚 Exception
public void createOrder2() {
    // 抛出 RuntimeException，不会回滚！
    throw new RuntimeException("失败");
}

✅ 正确示例:

// 明确指定 rollbackFor
@GlobalTransactional(rollbackFor = Exception.class)
public void createOrder() {
    throw new Exception("失败"); // 会回滚
}

// 或捕获后重新抛出
@GlobalTransactional
public void createOrder() throws Exception {
    try {
        // ...
    } catch (Exception e) {
        log.error("失败", e);
        throw e; // 重新抛出，会回滚
    }
}

2. 异常被吞掉

❌ 错误示例:

@GlobalTransactional
public void createOrder() {
    try {
        inventoryService.decreaseStock();
    } catch (Exception e) {
        log.error("扣减库存失败", e);
        // ❌ 异常被吞掉，事务不会回滚
    }
}

✅ 正确示例:

@GlobalTransactional
public void createOrder() {
    try {
        inventoryService.decreaseStock();
    } catch (Exception e) {
        log.error("扣减库存失败", e);
        throw new RuntimeException("扣减库存失败", e); // 重新抛出
    }
}

4.3 性能问题

现象: 事务执行慢，TPS 上不去

优化建议:

1. Undo Log 配置优化

seata:
  client:
    undo:
      # 仅记录更新的列（减少日志大小）
      only-care-update-columns: true
      # 压缩阈值
      compress:
        enable: true
        type: zip
        threshold: 64k

2. 批量提交优化

seata:
  client:
    rm:
      # 异步提交缓冲区大小
      async-commit-buffer-limit: 10000
      # 批量发送请求
      report-success-enable: false

3. 数据库索引优化

-- 确保 undo_log 表有合适的索引
ALTER TABLE undo_log 
ADD INDEX idx_xid_branch (xid, branch_id);

-- 定期清理历史 undo_log
DELETE FROM undo_log 
WHERE log_created < DATE_SUB(NOW(), INTERVAL 7 DAY);

4.4 脏写问题

现象: Phase 1 提交后，其他事务修改了数据

原因: AT 模式 Phase 1 提交后立即释放锁，存在脏写风险

解决方案:

方案 1: 使用全局锁

seata:
  client:
    rm:
      lock:
        # 开启全局锁
        retry-interval: 10
        retry-times: 30
        retry-policy-branch-rollback-on-conflict: true

方案 2: 升级 TCC 模式

对隔离性要求高的场景，考虑使用 TCC 模式

五、最佳实践

5.1 开发规范

✅ 推荐做法:

1. 事务粒度控制

// ✅ 好：事务范围小，执行快
@GlobalTransactional(timeoutMills = 60000)
public void createOrder(OrderDTO orderDTO) {
    // 仅包含必要的数据库操作
    orderMapper.insert(orderDTO);
    inventoryService.decreaseStock();
    accountService.decreaseBalance();
}

// ❌ 不好：事务范围过大，包含耗时操作
@GlobalTransactional(timeoutMills = 300000)
public void processEverything(OrderDTO orderDTO) {
    createOrder(orderDTO);  // 数据库操作
    sendEmail();            // ❌ 耗时的 HTTP 调用 (可能 5-10 秒)
    generateReport();       // ❌ 复杂的 Excel 生成 (可能 20 秒)
    uploadToOSS();          // ❌ 文件上传 (可能 30 秒)
}

原则:

事务内只包含必要的数据库操作
避免 HTTP 调用、文件处理等耗时操作
单个事务执行时间控制在 3 秒内

2. 超时时间设置

// 根据业务复杂度设置合理超时时间
@GlobalTransactional(timeoutMills = 300000) // 5 分钟
public void complexBusiness() {
    // 涉及 10+ 个微服务的复杂业务
}

@GlobalTransactional(timeoutMills = 60000) // 1 分钟
public void simpleBusiness() {
    // 涉及 2-3 个微服务的简单业务
}

@GlobalTransactional(timeoutMills = 10000) // 10 秒
public void quickBusiness() {
    // 简单的单表操作
}

建议:

默认超时时间：60 秒
复杂业务：不超过 5 分钟
简单查询：10-20 秒

3. 异常处理

@GlobalTransactional(rollbackFor = Exception.class)
public void businessProcess(OrderDTO orderDTO) {
    try {
        // 业务逻辑
        orderService.create(orderDTO);
        inventoryService.decrease();
    } catch (SpecificException e) {
        // 记录详细日志
        log.error("订单创建失败，orderId={}", orderDTO.getOrderId(), e);
        // 抛出运行时异常，触发回滚
        throw new BusinessException("订单创建失败", e);
    }
}

要点:

明确指定 rollbackFor = Exception.class
捕获异常后必须重新抛出
记录详细的错误日志便于排查

❌ 避免做法:

1. 避免在事务中执行耗时操作

@GlobalTransactional
public void badPractice(OrderDTO orderDTO) {
    // ✅ 快：数据库操作 (毫秒级)
    orderMapper.insert(orderDTO);
    
    // ❌ 慢：HTTP 调用 (秒级)
    restTemplate.postForObject(
        "http://external-service/api/notify", 
        orderDTO, 
        Response.class
    );  // 可能超时 5-10 秒
    
    // ❌ 慢：文件处理
    generateLargeExcel();  // 20-30 秒
    
    // ❌ 慢：第三方服务调用
    smsService.sendSms();  // 网络不稳定时可能 10+ 秒
}

改进方案:

@GlobalTransactional
public void goodPractice(OrderDTO orderDTO) {
    // 仅包含数据库操作
    orderMapper.insert(orderDTO);
    inventoryService.decrease();
    accountService.decrease();
}

// 异步执行耗时操作
@Component
public class OrderListener {
    
    @RocketMQMessageListener(topic = "order-created")
    public void onMessage(OrderDTO orderDTO) {
        // 异步发送通知
        sendEmailAsync(orderDTO);
        // 异步生成报表
        generateReportAsync(orderDTO);
    }
}

2. 避免事务嵌套

@GlobalTransactional
public void outerMethod() {
    // ❌ 内部方法也有 @GlobalTransactional
    innerMethod();  // 不会开启新事务，复用外层 XID
}

@GlobalTransactional(propagation = Propagation.REQUIRES_NEW)
public void innerMethod() {
    // ...
}

问题:

内层事务不会开启新事务
内层异常会导致外层整体回滚
性能损耗（两次 TC 交互）

正确做法:

@GlobalTransactional
public void businessProcess() {
    // 直接调用普通方法
    step1();
    step2();
    step3();
}

// 无需注解
private void step1() { /* ... */ }
private void step2() { /* ... */ }
private void step3() { /* ... */ }

3. 避免跨线程 XID 丢失

@GlobalTransactional
public void processWithThreads() {
    String xid = RootContext.getXID();
    
    // ❌ 错误：新线程没有 XID
    new Thread(() -> {
        serviceA.doSomething();  // 不会注册分支事务
    }).start();
    
    // ✅ 正确：手动传递 XID
    CompletableFuture.runAsync(() -> {
        try {
            RootContext.bind(xid);  // 绑定 XID
            serviceB.doSomething(); // 会注册分支事务
        } finally {
            RootContext.unbind();   // 清理
        }
    });
    
    // ✅ 更好：使用事务模板
    transactionalTemplate.execute(() -> {
        serviceC.doSomething();
        return null;
    });
}

5.2 监控告警

关键指标监控

Prometheus 配置:

seata:
  metrics:
    enabled: true
    exporter-list: prometheus
    exporter-prometheus-port: 9898

核心监控指标:

指标名称	说明	告警阈值
`global_transaction_tps`	全局事务 TPS	< 100 持续 5 分钟
`global_transaction_success_rate`	事务成功率	< 99%
`global_transaction_latency_avg`	平均耗时	> 3 秒
`global_transaction_rollback_count`	回滚数量	> 10/小时
`undo_log_size`	undo_log 表大小	> 100 万行

Grafana 看板

导入 Seata 官方 Dashboard:

Dashboard ID: 13393
监控内容:
- 全局事务状态分布
- 分支事务执行情况
- TP99/TP95 延迟
- 回滚趋势图

示例告警规则:

groups:
  - name: seata_alerts
    rules:
      # 事务成功率低于 99%
      - alert: SeataTransactionSuccessRateLow
        expr: rate(seata_global_transaction_success_total[5m]) / rate(seata_global_transaction_total[5m]) < 0.99
        for: 5m
        annotations:
          summary: "Seata 事务成功率过低"
          
      # 回滚事务过多
      - alert: SeataRollbackCountHigh
        expr: increase(seata_global_transaction_rollback_total[1h]) > 10
        for: 1h
        annotations:
          summary: "Seata 回滚事务过多"

5.3 性能调优

JVM 参数优化

#!/bin/bash
# seata-server.sh

JAVA_OPTS="-server \
  -Xms4g -Xmx4g \           # 堆内存 (根据服务器配置调整)
  -Xmn2g \                  # 新生代
  -XX:MetaspaceSize=256m \  # 元空间初始值
  -XX:MaxMetaspaceSize=512m \ # 元空间最大值
  -XX:+UseG1GC \            # G1 垃圾回收器
  -XX:MaxGCPauseMillis=100 \ # 最大 GC 停顿时间
  -XX:+HeapDumpOnOutOfMemoryError \ # OOM 时 dump 堆
  -XX:HeapDumpPath=/opt/seata/logs/heapdump.hprof"

export JAVA_OPTS

参数说明:

-Xms/-Xmx: 设置为相同值，避免内存动态伸缩
-Xmn: 新生代大小为堆的 1/2
-XX:+UseG1GC: G1 适合大堆低延迟场景
-XX:MaxGCPauseMillis: 控制 GC 停顿时间

数据库连接池优化

spring:
  datasource:
    hikari:
      # 最小空闲连接数
      minimum-idle: 10
      # 最大连接池大小
      maximum-pool-size: 30
      # 连接超时时间 (毫秒)
      connection-timeout: 30000
      # 空闲连接超时 (毫秒)
      idle-timeout: 600000
      # 连接最大生命周期 (毫秒)
      max-lifetime: 1800000
      # 连接测试查询
      connection-test-query: SELECT 1

优化建议:

maximum-pool-size: 根据并发量调整 (CPU 核数 * 2 + 1)
connection-timeout: 30 秒足够
定期监控连接池使用率 (建议 < 80%)

Undo Log 优化

seata:
  client:
    undo:
      # 仅记录更新的列 (减少日志大小)
      only-care-update-columns: true
      # 压缩配置
      compress:
        enable: true        # 启用压缩
        type: zip           # 压缩算法
        threshold: 64k      # 超过 64KB 才压缩

效果:

减少 undo_log 存储空间 50%-70%
降低网络传输开销
提升回滚速度

5.4 数据清理

定期清理策略

方案 1: MySQL 定时任务

-- 创建存储过程
DELIMITER $$

CREATE PROCEDURE cleanup_seata_undo_log()
BEGIN
    -- 清理 7 天前的 undo_log
    DELETE FROM undo_log 
    WHERE log_created < DATE_SUB(NOW(), INTERVAL 7 DAY)
    LIMIT 10000;  -- 分批删除，避免锁表
    
    -- 记录清理日志
    INSERT INTO cleanup_log (cleanup_time, deleted_count)
    VALUES (NOW(), ROW_COUNT());
END$$

DELIMITER ;

-- 创建定时任务，每天凌晨 2 点执行
CREATE EVENT IF NOT EXISTS cleanup_undo_log_event
ON SCHEDULE EVERY 1 DAY STARTS '2024-01-01 02:00:00'
DO CALL cleanup_seata_undo_log();

方案 2: Spring Boot 定时任务

@Component
public class UndoLogCleaner {
    
    @Autowired
    private JdbcTemplate jdbcTemplate;
    
    /**
     * 每天凌晨 2 点清理 7 天前的 undo_log
     */
    @Scheduled(cron = "0 0 2 * * ?")
    public void cleanUndoLog() {
        try {
            int days = 7;
            String sql = "DELETE FROM undo_log WHERE log_created < DATE_SUB(NOW(), INTERVAL ? DAY)";
            int count = jdbcTemplate.update(sql, days);
            log.info("清理 undo_log 完成，删除 {} 条记录", count);
        } catch (Exception e) {
            log.error("清理 undo_log 失败", e);
        }
    }
}

方案 3: Shell 脚本 + Crontab

#!/bin/bash
# clean_undo_log.sh

DB_HOST="localhost"
DB_USER="root"
DB_PASS="password"
DB_NAME="seata"
DAYS=7

mysql -h${DB_HOST} -u${DB_USER} -p${DB_PASS} ${DB_NAME} <<EOF
DELETE FROM undo_log 
WHERE log_created < DATE_SUB(NOW(), INTERVAL ${DAYS} DAY)
LIMIT 10000;
EOF

if [ $? -eq 0 ]; then
    echo "$(date): 清理 undo_log 成功" >> /var/log/cleanup.log
else
    echo "$(date): 清理 undo_log 失败" >> /var/log/cleanup.log
fi

Crontab 配置:

1 2	# 每天凌晨 2 点执行 0 2 * * * /opt/scripts/clean_undo_log.sh

5.5 安全防护

1. Nacos 鉴权

seata:
  registry:
    type: nacos
    nacos:
      server-addr: laosan.xin:8848
      username: seata_user      # 专用账号
      password: StrongP@ssw0rd  # 强密码
      namespace: dev-wdg

安全建议:

为 Seata 创建专用 Nacos 账号
密码长度 >= 12 位，包含大小写 + 数字 + 特殊字符
定期更换密码 (90 天)

2. 数据库权限最小化

-- 创建 Seata 专用数据库用户
CREATE USER 'seata'@'%' IDENTIFIED BY 'StrongP@ssw0rd';

-- 仅授予必要权限
GRANT SELECT, INSERT, UPDATE, DELETE ON seata.* TO 'seata'@'%';
GRANT CREATE TABLE, DROP, INDEX ON seata.* TO 'seata'@'%';

-- 撤销危险权限
REVOKE DROP ON *.* FROM 'seata'@'%';
REVOKE ALTER ON *.* FROM 'seata'@'%';
REVOKE GRANT OPTION ON *.* FROM 'seata'@'%';

FLUSH PRIVILEGES;

权限原则:

仅授予业务必需的权限
禁止 DROP、ALTER 等危险操作
禁止访问其他数据库

3. 敏感配置加密

# 使用 Jasypt 加密数据库密码
seata:
  store:
    db:
      user: root
      password: ENC(AbCdEfGhIjKlMnOpQrStUvWxYz)  # 加密后的密码

配置方法:

<!-- 添加依赖 -->
<dependency>
    <groupId>com.github.ulisesbocchio</groupId>
    <artifactId>jasypt-spring-boot-starter</artifactId>
    <version>3.0.5</version>
</dependency>

# 生成加密密码
java -cp jasypt-1.9.3.jar org.jasypt.intf.cli.JasyptPBEStringEncryptionCLI \
  input="my_password" \
  password="encryption_key" \
  algorithm=PBEWITHHMACSHA512ANDAES_256

5.6 故障处理

建立应急预案

预案 1: 事务卡住不提交

-- 1. 查看卡住的事务
SELECT xid, status, begin_time, timeout, gmt_modified
FROM global_table
WHERE status IN (2, 4)  -- Committing 或 Rollbacking
AND gmt_modified < NOW() - INTERVAL 5 MINUTE;

-- 2. 手动提交 (谨慎使用！)
UPDATE global_table 
SET status = 3, gmt_modified = NOW() 
WHERE xid = 'xxx' AND status = 2;

-- 3. 清理锁记录
DELETE FROM lock_table WHERE xid = 'xxx';

操作流程:

确认业务数据安全
记录操作日志
执行手动提交
通知业务方验证
事后分析根因

预案 2: 大量事务回滚

处理步骤:

立即停止相关服务
查看 Seata Server 日志
分析回滚原因
修复 Bug 后重启
数据对账修复

预案 3: Undo Log 表爆满

-- 紧急清理
DELETE FROM undo_log 
WHERE log_status = 1  -- 已处理状态
LIMIT 100000;

-- 优化表
OPTIMIZE TABLE undo_log;

故障演练

定期演练计划:

演练场景	频率	参与人员
事务超时回滚	每月 1 次	开发 + 运维
数据库宕机切换	每季度 1 次	运维 + DBA
Seata Server 故障	每季度 1 次	运维
数据不一致修复	每半年 1 次	开发 + 测试

演练流程:

制定演练方案
准备测试环境
执行故障注入
观察系统反应
记录问题清单
优化改进措施

六、总结

6.1 AT 模式优势

✅ 零侵入: 无需修改业务代码，仅添加注解
✅ 简单易用: 学习成本低，快速上手
✅ 高性能: 两阶段非阻塞提交，TPS 高
✅ 广泛支持: 兼容 MySQL、Oracle、PostgreSQL 等主流数据库
✅ 生态完善: 与 Spring Cloud、Dubbo 无缝集成

6.2 适用场景

✅ 推荐场景:

90% 的微服务分布式事务需求
对隔离性要求不苛刻的业务 (可接受短暂脏读)
追求快速上线、降低开发成本的项目
团队技术能力一般，希望降低维护成本

❌ 不推荐场景:

高并发秒杀系统 (考虑 TCC 模式)
强一致性要求 (考虑 XA 模式)
非关系型数据库场景 (Redis/MongoDB 等)
跨语言调用场景 (Go/Python 等)

6.3 与其他模式对比

特性	AT	TCC	Saga	XA
侵入性	无	高	中	无
性能	⭐⭐⭐⭐	⭐⭐⭐⭐⭐	⭐⭐⭐⭐	⭐⭐
隔离性	读未提交	业务隔离	无隔离	完全隔离
开发成本	低	高	中	低
适用场景	通用	高并发	长流程	强一致

6.4 快速选型指南

需要分布式事务吗？
├─ 否 → 使用本地事务
└─ 是 → 对隔离性要求高吗？
    ├─ 是 → 选择 TCC 模式
    └─ 否 → 是长流程编排吗？
        ├─ 是 → 选择 Saga 模式
        └─ 否 → 需要强一致性吗？
            ├─ 是 → 选择 XA 模式
            └─ 否 → ✅ 选择 AT 模式 (推荐)

6.5 学习资源

📖 官方文档: https://seata.apache.org/
💬 GitHub Issues: https://github.com/apache/incubator-seata/issues
📺 视频教程: Bilibili Seata 官方频道
📚 源码解读: 《Seata 源码解析》系列文章
💻 实战案例: GitHub seata-examples 仓库

作者: laosan
更新日期: 2024/8/7
版本: Seata 1.6.1
版权声明: 本文采用 CC BY-NC-SA 4.0 许可协议

# AT 模式, seata, 分布式事务

240807-分布式事务解决方案Seata之AT事务

Seata AT事务模式详解

一、AT 模式核心概念

1.1 全局事务与分支事务

1.2 两阶段提交模型

二、自动注册原理

2.1 数据源代理技术

2.2 自动注册流程

2.3 XID 传递机制

三、快速开始

3.1 数据库初始化

3.2 添加依赖

3.3 配置文件

3.4 开启全局事务

3.5 下游服务实现

3.6 测试验证

单元测试

查看 Seata Server 日志

四、常见问题排查

4.1 分支事务未自动注册

1. 手动创建了 DataSource

2. 多数据源场景

3. XID 丢失（最常见）

场景 1: 子线程中执行

场景 2: Feign/Dubbo 调用未透传 XID

场景 3: 消息队列消费端

4.2 事务不回滚

1. 异常类型不匹配

2. 异常被吞掉

4.3 性能问题

1. Undo Log 配置优化

2. 批量提交优化

3. 数据库索引优化

4.4 脏写问题

方案 1: 使用全局锁

方案 2: 升级 TCC 模式

五、最佳实践

5.1 开发规范

1. 事务粒度控制

2. 超时时间设置

3. 异常处理

1. 避免在事务中执行耗时操作

2. 避免事务嵌套

3. 避免跨线程 XID 丢失

5.2 监控告警

关键指标监控

Grafana 看板

5.3 性能调优

JVM 参数优化

数据库连接池优化

Undo Log 优化

5.4 数据清理

定期清理策略

方案 1: MySQL 定时任务

方案 2: Spring Boot 定时任务

方案 3: Shell 脚本 + Crontab

5.5 安全防护

1. Nacos 鉴权

2. 数据库权限最小化

3. 敏感配置加密

5.6 故障处理

建立应急预案

预案 1: 事务卡住不提交

预案 2: 大量事务回滚

预案 3: Undo Log 表爆满

故障演练

六、总结

6.1 AT 模式优势

6.2 适用场景

6.3 与其他模式对比

6.4 快速选型指南

6.5 学习资源

标签云

最新文章

标签

分类

归档

分类

归档

Your browser is out-of-date!