Spring Cloud微服务架构下Seata分布式事务解决方案：AT与TCC模式深度解析与实战指南

// 全局锁配置优化示例
seata:
  client:
    lock:
      retry-interval: 10ms    # 重试间隔优化
      retry-times: 30         # 重试次数调整
      lock-expire-time: 60000 # 锁过期时间

@GlobalTransactional(timeoutMills = 60000)
public void businessMethod() {
    // 业务逻辑
}

seata:
  client:
    tm:
      commit-retry-count: 3
      rollback-retry-count: 3
    lock:
      retry-interval: 10
      retry-times: 30

// 使用业务键分片减少锁竞争
@GlobalTransactional
public void processOrder(Order order) {
    String shardKey = order.getUserId() % 10; // 按用户ID分片
    // 业务处理
}

spring:
  datasource:
    hikari:
      maximum-pool-size: 20
      minimum-idle: 5
      connection-timeout: 30000

// 使用分布式锁保证幂等性
@Autowired
private RedissonClient redissonClient;

public boolean confirm(String xid, String businessKey) {
    RLock lock = redissonClient.getLock("tcc_confirm:" + xid + ":" + businessKey);
    if (lock.tryLock()) {
        try {
            // 检查是否已处理
            if (transactionLogRepository.existsByXidAndStatus(xid, "CONFIRMED")) {
                return true;
            }
            // 执行确认逻辑
            // ...
            // 记录处理状态
            transactionLogRepository.save(new TransactionLog(xid, "CONFIRMED"));
            return true;
        } finally {
            lock.unlock();
        }
    }
    return false;
}

# Istio VirtualService 配置示例
apiVersion: networking.istio.io/v1alpha3
kind: VirtualService
spec:
  hosts:
  - tcc-service
  http:
  - route:
    - destination:
        host: tcc-service
    timeout: 30s
    retries:
      attempts: 3
      perTryTimeout: 10s
      retryOn: gateway-error,connect-failure,refused-stream

<dependency>
    <groupId>com.alibaba.cloud</groupId>
    <artifactId>spring-cloud-starter-alibaba-seata</artifactId>
    <version>2023.0.1.0</version>
</dependency>

version: '3.8'
services:
  seata-server:
    image: seataio/seata-server:2.1.0
    container_name: seata-server
    ports:
      - "8091:8091"
      - "7091:7091"
    environment:
      - SEATA_IP=your_server_ip
      - STORE_MODE=db
      - SEATA_CONFIG_NAME=file:/root/seata-config/registry
    volumes:
      - ./seata/config:/root/seata-config

seata:
  application-id: order-service
  tx-service-group: my_tx_group
  enabled: true
  enable-auto-data-source-proxy: true
  config:
    type: nacos
    nacos:
      server-addr: 127.0.0.1:8848
  registry:
    type: nacos
    nacos:
      application: seata-server
      server-addr: 127.0.0.1:8848
      namespace: seata-cluster

@Service
@Slf4j
public class OrderService {
    @Autowired
    private OrderMapper orderMapper;
    @Autowired
    private StockService stockService;
    
    @GlobalTransactional(name = "create-order", timeoutMills = 60000, rollbackFor = Exception.class)
    public OrderDTO createOrder(OrderCreateRequest request) {
        log.info("全局事务开始，XID: {}", RootContext.getXID());
        
        // 1. 本地事务：插入订单
        Order order = convertToOrder(request);
        orderMapper.insert(order);
        
        // 2. 远程调用：扣减库存
        stockService.deductStock(request.getProductId(), request.getCount());
        
        // 3. 记录事务日志用于监控
        log.info("订单创建成功，订单ID: {}", order.getId());
        return convertToDTO(order);
    }
}

@Service
public class StockService {
    @Transactional(rollbackFor = Exception.class)
    public void deductStock(Long productId, Integer count) {
        // 使用乐观锁防止并发问题
        int rows = stockMapper.updateStockWithVersion(productId, count);
        if (rows == 0) {
            throw new RuntimeException("库存扣减失败，版本冲突");
        }
        log.info("库存扣减成功，产品ID: {}, 数量: {}", productId, count);
    }
}

CREATE TABLE undo_log (
  id BIGINT AUTO_INCREMENT PRIMARY KEY,
  branch_id BIGINT NOT NULL,
  xid VARCHAR(128) NOT NULL,
  context VARCHAR(128) NOT NULL,
  rollback_info LONGBLOB NOT NULL,
  log_status INT NOT NULL,
  log_created DATETIME(6) NOT NULL,
  log_modified DATETIME(6) NOT NULL,
  ext VARCHAR(100) DEFAULT NULL,
  UNIQUE KEY ux_undo_log (xid, branch_id)
) ENGINE=InnoDB AUTO_INCREMENT=1 DEFAULT CHARSET=utf8mb4;

management:
  endpoints:
    web:
      exposure:
        include: metrics,health,info
  metrics:
    export:
      prometheus:
        enabled: true

@Configuration
public class SeataDataSourceConfig {
    
    @Bean
    @ConfigurationProperties(prefix = "spring.datasource")
    public DruidDataSource druidDataSource() {
        return new DruidDataSource();
    }
    
    @Primary
    @Bean("dataSource")
    public DataSource dataSource(DruidDataSource druidDataSource) {
        return new DataSourceProxy(druidDataSource);
    }
}

@SpringBootTest
@TestMethodOrder(MethodOrderer.OrderAnnotation.class)
class OrderServiceATTest {
    
    @Autowired
    private OrderService orderService;
    
    @Autowired
    private MeterRegistry meterRegistry;
    
    @Test
    @Order(1)
    void testCreateOrderSuccess() {
        // 正常下单：订单和库存均更新
        OrderCreateRequest request = new OrderCreateRequest(1001L, 10, 1L);
        OrderDTO result = orderService.createOrder(request);
        
        Assertions.assertNotNull(result.getId());
        Assertions.assertEquals(90, stockService.getStock(1001L));
        
        // 验证事务指标
        Counter successCounter = meterRegistry.counter("seata.transaction.success");
        assertTrue(successCounter.count() > 0);
    }
    
    @Test
    @Order(2)
    void testCreateOrderRollback() {
        // 模拟库存不足异常：全局事务回滚
        stockService.setStock(1001L, 5);
        
        OrderCreateRequest request = new OrderCreateRequest(1001L, 10, 1L);
        assertThrows(RuntimeException.class, () -> orderService.createOrder(request));
        
        // 验证回滚：订单未创建，库存未变化
        Assertions.assertEquals(0, orderService.getOrderCount(1L));
        Assertions.assertEquals(5, stockService.getStock(1001L));
        
        // 验证回滚指标
        Counter rollbackCounter = meterRegistry.counter("seata.transaction.rollback");
        assertTrue(rollbackCounter.count() > 0);
    }
    
    @Test
    @Order(3)
    void testConcurrentOrderCreation() throws InterruptedException {
        // 并发测试：验证全局锁机制
        int threadCount = 10;
        CountDownLatch latch = new CountDownLatch(threadCount);
        
        for (int i = 0; i < threadCount; i++) {
            new Thread(() -> {
                try {
                    orderService.createOrder(new OrderCreateRequest(1002L, 1, 1L));
                } catch (Exception e) {
                    log.info("并发冲突处理: {}", e.getMessage());
                } finally {
                    latch.countDown();
                }
            }).start();
        }
        
        latch.await(30, TimeUnit.SECONDS);
        // 验证最终一致性
        assertTrue(stockService.getStock(1002L) >= 0);
    }
}

<dependency>
    <groupId>io.seata</groupId>
    <artifactId>seata-spring-boot-starter</artifactId>
    <version>2.1.0</version>
</dependency>
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-data-jpa</artifactId>
</dependency>
<dependency>
    <groupId>org.springframework.cloud</groupId>
    <artifactId>spring-cloud-starter-kubernetes</artifactId>
    <version>3.1.0</version>
</dependency>

seata:
  enabled: true
  application-id: order-service
  tx-service-group: my_test_tx_group_2025
  service:
    vgroup-mapping:
      my_test_tx_group_2025: default
    grouplist:
      default: seata-server.${NAMESPACE:default}.svc.cluster.local:8091
  config:
    type: nacos
    nacos:
      server-addr: ${NACOS_SERVER:localhost:8848}

@LocalTCC
public interface InventoryTccAction {
    
    @TwoPhaseBusinessAction(name = "inventoryTccAction", commitMethod = "commit", 
                           rollbackMethod = "rollback")
    boolean prepare(BusinessActionContext actionContext,
                   @BusinessActionContextParameter(paramName = "productId") String productId,
                   @BusinessActionContextParameter(paramName = "count") Integer count);
    
    boolean commit(BusinessActionContext actionContext);
    
    boolean rollback(BusinessActionContext actionContext);
}

@Service
@Slf4j
public class InventoryTccActionImpl implements InventoryTccAction {
    
    @Autowired
    private InventoryRepository inventoryRepository;
    
    @Autowired
    private TransactionStateRepository stateRepository;
    
    @Override
    @Retryable(value = Exception.class, maxAttempts = 3, backoff = @Backoff(delay = 1000))
    public boolean prepare(BusinessActionContext actionContext, String productId, Integer count) {
        String xid = actionContext.getXid();
        
        // 检查是否已执行过prepare，确保幂等性
        if (stateRepository.existsByXidAndPhase(xid, "prepare")) {
            log.info("Prepare操作已执行，直接返回成功，xid: {}", xid);
            return true;
        }
        
        Inventory inventory = inventoryRepository.findByProductIdWithLock(productId);
        if (inventory.getAvailable() < count) {
            throw new InventoryInsufficientException("库存不足，当前可用库存：" + inventory.getAvailable());
        }
        
        // 冻结部分库存
        inventory.setFrozen(inventory.getFrozen() + count);
        inventory.setAvailable(inventory.getAvailable() - count);
        inventoryRepository.save(inventory);
        
        // 记录事务状态
        stateRepository.save(new TransactionState(xid, "prepare", "SUCCESS"));
        
        return true;
    }
    
    @Override
    public boolean commit(BusinessActionContext actionContext) {
        String xid = actionContext.getXid();
        
        // 幂等性检查
        if (stateRepository.existsByXidAndPhase(xid, "commit")) {
            log.info("Commit操作已执行，直接返回成功，xid: {}", xid);
            return true;
        }
        
        String productId = (String) actionContext.getActionContext("productId");
        Integer count = (Integer) actionContext.getActionContext("count");
        
        try {
            Inventory inventory = inventoryRepository.findByProductId(productId);
            inventory.setFrozen(inventory.getFrozen() - count);
            inventoryRepository.save(inventory);
            
            stateRepository.save(new TransactionState(xid, "commit", "SUCCESS"));
            return true;
        } catch (Exception e) {
            log.error("Commit操作失败，xid: {}", xid, e);
            stateRepository.save(new TransactionState(xid, "commit", "FAILED"));
            throw new TccCommitException("库存确认扣减失败", e);
        }
    }
    
    @Override
    public boolean rollback(BusinessActionContext actionContext) {
        String xid = actionContext.getXid();
        
        if (stateRepository.existsByXidAndPhase(xid, "rollback")) {
            log.info("Rollback操作已执行，直接返回成功，xid: {}", xid);
            return true;
        }
        
        String productId = (String) actionContext.getActionContext("productId");
        Integer count = (Integer) actionContext.getActionContext("count");
        
        try {
            Inventory inventory = inventoryRepository.findByProductId(productId);
            inventory.setFrozen(inventory.getFrozen() - count);
            inventory.setAvailable(inventory.getAvailable() + count);
            inventoryRepository.save(inventory);
            
            stateRepository.save(new TransactionState(xid, "rollback", "SUCCESS"));
            return true;
        } catch (Exception e) {
            log.error("Rollback操作失败，需要人工干预，xid: {}", xid, e);
            stateRepository.save(new TransactionState(xid, "rollback", "FAILED"));
            // 触发告警，需要人工干预
            alertService.sendManualInterventionAlert(xid, "库存回滚失败");
            throw new TccRollbackException("库存回滚失败，需要人工干预", e);
        }
    }
}

@Service
@Slf4j
public class OrderService {
    
    @Autowired
    private InventoryTccAction inventoryTccAction;
    
    @Autowired
    private AccountTccAction accountTccAction;
    
    @Autowired
    private CircuitBreakerFactory circuitBreakerFactory;
    
    @GlobalTransactional(timeoutMills = 300000, name = "create-order-tx-2025")
    public Order createOrder(OrderRequest request) {
        CircuitBreaker circuitBreaker = circuitBreakerFactory.create("inventoryService");
        
        // 1. 创建订单记录
        Order order = createOrderRecord(request);
        
        // 2. TCC模式扣减库存（支持熔断降级）
        boolean inventoryResult = circuitBreaker.run(() -> {
            BusinessActionContext inventoryContext = new BusinessActionContext();
            return inventoryTccAction.prepare(inventoryContext, 
                request.getProductId(), request.getCount());
        }, throwable -> {
            log.warn("库存服务熔断降级，订单创建进入人工审核流程");
            order.setStatus(OrderStatus.PENDING_REVIEW);
            orderRepository.save(order);
            return false;
        });
        
        // 3. TCC模式扣减账户余额
        BusinessActionContext accountContext = new BusinessActionContext();
        boolean accountResult = accountTccAction.prepare(accountContext,
            request.getUserId(), request.getTotalAmount());
        
        if (!inventoryResult || !accountResult) {
            throw new OrderCreationException("订单创建过程中部分服务异常");
        }
        
        return order;
    }
    
    // 网络分区场景下的补偿接口
    @PostMapping("/orders/{orderId}/compensate")
    public ResponseEntity<String> manualCompensate(@PathVariable String orderId) {
        try {
            Order order = orderRepository.findById(orderId)
                .orElseThrow(() -> new OrderNotFoundException("订单不存在"));
            
            if (order.getStatus() == OrderStatus.PENDING_REVIEW) {
                // 执行手动补偿逻辑
                compensateService.manualCompensate(order);
                return ResponseEntity.ok("补偿操作执行成功");
            }
            
            return ResponseEntity.badRequest().body("订单状态不支持补偿操作");
        } catch (Exception e) {
            log.error("手动补偿操作失败，orderId: {}", orderId, e);
            return ResponseEntity.internalServerError().body("补偿操作执行失败");
        }
    }
}

@Service
public class InventoryTccActionImpl implements InventoryTccAction {
    
    @Override
    public boolean commit(BusinessActionContext actionContext) {
        String xid = actionContext.getXid();
        
        // 增强的幂等性检查，支持分布式锁
        DistributedLock lock = distributedLockService.getLock("tcc_commit_" + xid);
        try {
            if (lock.tryLock(5, TimeUnit.SECONDS)) {
                if (isActionCompleted(xid, "commit")) {
                    log.info("Commit操作已执行，直接返回成功，xid: {}", xid);
                    return true;
                }
                
                // 执行实际的commit逻辑
                return doCommit(actionContext);
            } else {
                throw new TccOperationTimeoutException("获取分布式锁超时");
            }
        } finally {
            lock.unlock();
        }
    }
    
    private boolean doCommit(BusinessActionContext actionContext) {
        String xid = actionContext.getXid();
        String productId = (String) actionContext.getActionContext("productId");
        Integer count = (Integer) actionContext.getActionContext("count");
        
        try {
            // 模拟网络分区场景下的重试机制
            return executeWithRetry(() -> {
                Inventory inventory = inventoryRepository.findByProductId(productId);
                if (inventory == null) {
                    throw new InventoryNotFoundException("库存记录不存在");
                }
                
                inventory.setFrozen(inventory.getFrozen() - count);
                inventoryRepository.save(inventory);
                
                // 记录执行状态到多个存储，确保可靠性
                recordActionCompletion(xid, "commit", "SUCCESS");
                metricService.incrementCounter("tcc.commit.success");
                
                return true;
            }, 3, 1000);
        } catch (Exception e) {
            log.error("Commit操作最终失败，xid: {}", xid, e);
            recordActionCompletion(xid, "commit", "FAILED");
            metricService.incrementCounter("tcc.commit.failure");
            alertService.sendTccFailureAlert(xid, "commit", e.getMessage());
            throw new TccCommitException("库存确认扣减最终失败", e);
        }
    }
    
    private <T> T executeWithRetry(Callable<T> operation, int maxAttempts, long delayMs) {
        // 实现带指数退避的重试机制
    }
}

# istio-virtual-service.yaml
apiVersion: networking.istio.io/v1alpha3
kind: VirtualService
metadata:
  name: tcc-transaction-service
spec:
  hosts:
  - tcc-service.default.svc.cluster.local
  http:
  - match:
    - headers:
        x-tcc-transaction-id:
          regex: ".+"
    route:
    - destination:
        host: tcc-service.default.svc.cluster.local
      headers:
        request:
          set:
            x-b3-traceid: "%REQ(x-tcc-transaction-id)%"
    timeout: 30s
    retries:
      attempts: 3
      perTryTimeout: 10s

// 未来版本的TCC Service Mesh适配器
@Configuration
@EnableTccMesh
public class TccMeshConfiguration {
    
    @Bean
    public TccMeshInterceptor tccMeshInterceptor() {
        return new TccMeshInterceptor();
    }
    
    @Bean
    public TccSidecarProxy tccSidecarProxy() {
        return new TccSidecarProxy();
    }
}

seata:
  client:
    tm:
      commit-retry-count: 5
      rollback-retry-count: 5
    rm:
      report-retry-count: 3
  transport:
    enable-client-batch-send-request: true
    compression: gzip

@Configuration
@EnableCaching
public class SeataConfig2025 {
    
    @Bean
    public GlobalTransactionScanner globalTransactionScanner() {
        return new GlobalTransactionScanner("order-service", "my_test_tx_group_2025")
            .setMode(TransactionMode.TCC)
            .setEnableDegrade(false);
    }
    
    @Bean
    public CacheManager transactionStateCacheManager() {
        return new RedisCacheManager(redisCacheWriter, cacheConfiguration);
    }
}

@Component
public class TccTransactionMonitor2025 {
    
    @EventListener
    public void onTransactionEvent(TransactionEvent event) {
        // 集成OpenTelemetry进行分布式追踪
        Span span = tracer.spanBuilder("tcc-transaction")
            .setParent(Context.current().with(traceContext))
            .startSpan();
        
        try (Scope scope = span.makeCurrent()) {
            span.setAttribute("xid", event.getXid());
            span.setAttribute("phase", event.getPhase());
            span.setAttribute("status", event.getStatus());
            
            monitorService.recordTransactionStatus(event);
            metricService.recordTransactionMetrics(event);
        } finally {
            span.end();
        }
    }
}