Spring事务在微服务架构中的实践与挑战

引言

微服务事务的挑战

1. 传统ACID事务的局限性

// 单体应用中的事务 - 在微服务中无法实现
@Service
@Transactional
public class MonolithicOrderService {
    
    public void createOrder(Order order) {
        // 所有操作在同一个数据库事务中
        orderRepository.save(order);           // 订单库
        inventoryService.deductStock(order);   // 库存库  
        accountService.deductBalance(order);   // 账户库
        notificationService.sendEmail(order);  // 通知服务
        // 要么全部成功，要么全部回滚
    }
}

微服务中的问题：

• 每个服务有独立的数据库
• 无法使用传统的数据库事务
• 网络调用可能失败
• 服务可能不可用

2. CAP定理的影响

在分布式系统中，只能同时满足以下两个：

• 一致性 (Consistency)：所有节点看到相同的数据
• 可用性 (Availability)：每个请求都能获得响应
• 分区容错性 (Partition Tolerance)：系统在网络分区时仍能工作

微服务架构通常选择AP + 最终一致性。

分布式事务模式

1. 两阶段提交 (2PC)

模式描述：协调者协调多个参与者，分准备和提交两个阶段。

// 2PC 协调者实现示例
@Service
@Slf4j
public class TwoPhaseCommitCoordinator {
    
    @Autowired
    private OrderService orderService;
    
    @Autowired
    private InventoryService inventoryService;
    
    @Autowired
    private AccountService accountService;
    
    public void createOrderDistributed(Order order) {
        String transactionId = generateTransactionId();
        List participants = Arrays.asList(
            new Participant(\"order\", () -> orderService.prepareCreate(order, transactionId)),
            new Participant(\"inventory\", () -> inventoryService.prepareDeduct(order, transactionId)),
            new Participant(\"account\", () -> accountService.prepareDeduct(order, transactionId))
        );
        
        try {
            // 阶段一：准备阶段
            boolean allPrepared = preparePhase(participants, transactionId);
            
            if (allPrepared) {
                // 阶段二：提交阶段
                commitPhase(participants, transactionId);
                log.info(\"分布式事务提交成功: {}\", transactionId);
            } else {
                // 阶段二：回滚阶段
                rollbackPhase(participants, transactionId);
                log.warn(\"分布式事务回滚: {}\", transactionId);
            }
        } catch (Exception e) {
            rollbackPhase(participants, transactionId);
            throw new DistributedTransactionException(\"分布式事务执行失败\", e);
        }
    }
    
    private boolean preparePhase(List participants, String transactionId) {
        for (Participant participant : participants) {
            try {
                boolean prepared = participant.prepare();
                if (!prepared) {
                    log.warn(\"参与者准备失败: {}, 事务: {}\", participant.getName(), transactionId);
                    return false;
                }
            } catch (Exception e) {
                log.error(\"参与者准备异常: {}, 事务: {}\", participant.getName(), transactionId, e);
                return false;
            }
        }
        return true;
    }
    
    private void commitPhase(List participants, String transactionId) {
        for (Participant participant : participants) {
            try {
                participant.commit();
            } catch (Exception e) {
                // 提交阶段失败需要人工干预
                log.error(\"参与者提交失败: {}, 事务: {}\", participant.getName(), transactionId, e);
                // 记录异常，需要补偿
                recordCompensationTask(participant.getName(), transactionId, \"COMMIT\");
            }
        }
    }
    
    private void rollbackPhase(List participants, String transactionId) {
        for (Participant participant : participants) {
            try {
                participant.rollback();
            } catch (Exception e) {
                log.error(\"参与者回滚失败: {}, 事务: {}\", participant.getName(), transactionId, e);
                // 记录异常，需要补偿
                recordCompensationTask(participant.getName(), transactionId, \"ROLLBACK\");
            }
        }
    }
}

// 参与者接口
public interface Participant {
    String getName();
    boolean prepare();
    void commit();
    void rollback();
}

2. Saga模式

模式描述：将分布式事务拆分为一系列本地事务，每个事务都有对应的补偿操作。

// Saga 协调者实现
@Service
@Slf4j
public class SagaCoordinator {
    
    @Autowired
    private SagaStepExecutor stepExecutor;
    
    @Autowired
    private SagaStateRepository stateRepository;
    
    public void executeOrderSaga(Order order) {
        String sagaId = generateSagaId();
        SagaState sagaState = new SagaState(sagaId, order);
        
        try {
            // 定义Saga步骤
            List steps = Arrays.asList(
                new SagaStep(\"create_order\", 
                    () -> stepExecutor.createOrder(order, sagaId),
                    () -> stepExecutor.compensateOrder(order, sagaId)),
                    
                new SagaStep(\"deduct_inventory\",
                    () -> stepExecutor.deductInventory(order, sagaId), 
                    () -> stepExecutor.compensateInventory(order, sagaId)),
                    
                new SagaStep(\"deduct_balance\",
                    () -> stepExecutor.deductBalance(order, sagaId),
                    () -> stepExecutor.compensateBalance(order, sagaId)),
                    
                new SagaStep(\"send_notification\",
                    () -> stepExecutor.sendNotification(order, sagaId),
                    null) // 最后一步不需要补偿
            );
            
            // 执行Saga
            executeSagaSteps(steps, sagaState);
            
        } catch (Exception e) {
            log.error(\"Saga执行失败: {}\", sagaId, e);
            throw new SagaException(\"订单创建失败\", e);
        }
    }
    
    private void executeSagaSteps(List steps, SagaState sagaState) {
        for (int i = 0; i < steps.size(); i++) {
            SagaStep step = steps.get(i);
            sagaState.setCurrentStep(step.getName());
            
            try {
                // 执行正向操作
                step.execute();
                sagaState.recordSuccess(step.getName());
                
            } catch (Exception e) {
                log.error(\"Saga步骤执行失败: {}, 步骤: {}\", sagaState.getSagaId(), step.getName(), e);
                sagaState.recordFailure(step.getName(), e.getMessage());
                
                // 执行补偿操作
                compensatePreviousSteps(steps, i, sagaState);
                break;
            }
        }
        
        // 保存Saga状态
        stateRepository.save(sagaState);
    }
    
    private void compensatePreviousSteps(List steps, int failedIndex, SagaState sagaState) {
        // 从失败步骤的前一步开始反向补偿
        for (int i = failedIndex - 1; i >= 0; i--) {
            SagaStep step = steps.get(i);
            if (step.getCompensate() != null) {
                try {
                    step.compensate();
                    sagaState.recordCompensation(step.getName());
                } catch (Exception e) {
                    log.error(\"Saga补偿操作失败: {}, 步骤: {}\", sagaState.getSagaId(), step.getName(), e);
                    sagaState.recordCompensationFailure(step.getName(), e.getMessage());
                    // 补偿失败需要人工干预
                }
            }
        }
    }
}

// Saga步骤定义
@Data
@AllArgsConstructor
class SagaStep {
    private String name;
    private Runnable execute;
    private Runnable compensate;
}

3. TCC模式 (Try-Confirm-Cancel)

模式描述：每个服务提供Try、Confirm、Cancel三个接口。

// TCC 订单服务实现
@Service
@Slf4j
public class TccOrderService {
    
    @Autowired
    private OrderRepository orderRepository;
    
    @Autowired
    private TccTransactionRepository tccRepository;
    
    // Try阶段：预留资源
    @Transactional
    public boolean tryCreateOrder(Order order, String transactionId) {
        try {
            // 检查业务约束
            validateOrder(order);
            
            // 创建预订单状态
            Order pendingOrder = order.toPending(transactionId);
            orderRepository.save(pendingOrder);
            
            // 记录TCC状态
            TccTransaction tcc = new TccTransaction(transactionId, \"order\", \"CREATE\");
            tccRepository.save(tcc);
            
            log.info(\"订单Try阶段成功: {}\", transactionId);
            return true;
            
        } catch (Exception e) {
            log.error(\"订单Try阶段失败: {}\", transactionId, e);
            return false;
        }
    }
    
    // Confirm阶段：确认操作
    @Transactional 
    public void confirmCreateOrder(String transactionId) {
        TccTransaction tcc = tccRepository.findByTransactionIdAndService(transactionId, \"order\");
        if (tcc == null || !\"CREATE\".equals(tcc.getAction())) {
            throw new TccException(\"无效的TCC事务: \" + transactionId);
        }
        
        // 更新订单状态为确认
        Order pendingOrder = orderRepository.findByTransactionId(transactionId);
        if (pendingOrder != null) {
            pendingOrder.confirm();
            orderRepository.save(pendingOrder);
            
            // 删除TCC记录
            tccRepository.delete(tcc);
        }
        
        log.info(\"订单Confirm阶段成功: {}\", transactionId);
    }
    
    // Cancel阶段：取消操作
    @Transactional
    public void cancelCreateOrder(String transactionId) {
        TccTransaction tcc = tccRepository.findByTransactionIdAndService(transactionId, \"order\");
        if (tcc == null || !\"CREATE\".equals(tcc.getAction())) {
            throw new TccException(\"无效的TCC事务: \" + transactionId);
        }
        
        // 删除预订单
        Order pendingOrder = orderRepository.findByTransactionId(transactionId);
        if (pendingOrder != null) {
            orderRepository.delete(pendingOrder);
            
            // 删除TCC记录
            tccRepository.delete(tcc);
        }
        
        log.info(\"订单Cancel阶段成功: {}\", transactionId);
    }
}

Spring Cloud分布式事务实践

1. 使用Seata框架

Seata配置：

# application.yml
seata:
  enabled: true
  application-id: order-service
  tx-service-group: my_tx_group
  service:
    vgroup-mapping:
      my_tx_group: default
    disable-global-transaction: false
  config:
    type: nacos
    nacos:
      server-addr: 127.0.0.1:8848
  registry:
    type: nacos
    nacos:
      server-addr: 127.0.0.1:8848

全局事务使用：

// 订单服务
@Service
@Slf4j
public class SeataOrderService {
    
    @Autowired
    private OrderMapper orderMapper;
    
    @Autowired
    private InventoryFeignClient inventoryFeignClient;
    
    @Autowired
    private AccountFeignClient accountFeignClient;
    
    // 开启全局事务
    @GlobalTransactional
    public void createOrderWithSeata(Order order) {
        // 1. 创建本地订单
        orderMapper.insert(order);
        log.info(\"本地订单创建成功\");
        
        // 2. 调用库存服务（远程服务）
        Boolean inventoryResult = inventoryFeignClient.deductStock(order.getItems());
        if (!Boolean.TRUE.equals(inventoryResult)) {
            throw new RuntimeException(\"库存扣减失败\");
        }
        log.info(\"库存扣减成功\");
        
        // 3. 调用账户服务（远程服务）
        Boolean accountResult = accountFeignClient.deductBalance(order.getUserId(), order.getAmount());
        if (!Boolean.TRUE.equals(accountResult)) {
            throw new RuntimeException(\"余额扣减失败\");
        }
        log.info(\"余额扣减成功\");
        
        // 4. 更新订单状态
        order.complete();
        orderMapper.updateStatus(order);
        log.info(\"订单状态更新完成\");
    }
}

// Feign客户端配置
@FeignClient(name = \"inventory-service\", path = \"/api/inventory\")
public interface InventoryFeignClient {
    
    @PostMapping(\"/deduct\")
    Boolean deductStock(@RequestBody List items);
}

@FeignClient(name = \"account-service\", path = \"/api/account\")  
public interface AccountFeignClient {
    
    @PostMapping(\"/deduct\")
    Boolean deductBalance(@RequestParam Long userId, @RequestParam BigDecimal amount);
}

2. 事务消息模式

基于RocketMQ的事务消息：

@Service
@Slf4j
public class TransactionMessageService {
    
    @Autowired
    private RocketMQTemplate rocketMQTemplate;
    
    @Autowired
    private OrderService orderService;
    
    // 发送事务消息
    public void createOrderWithTransactionMessage(Order order) {
        // 执行本地事务
        Order savedOrder = orderService.createOrderLocal(order);
        
        // 发送事务消息
        TransactionSendResult result = rocketMQTemplate.sendMessageInTransaction(
            \"order-topic\", 
            MessageBuilder.withPayload(order).build(),
            savedOrder.getId() // 业务参数
        );
        
        log.info(\"事务消息发送结果: {}\", result.getSendStatus());
    }
    
    // 本地事务执行器
    @Transactional
    public void executeLocalTransaction(String transactionId, Long orderId) {
        // 查询订单状态
        Order order = orderService.findById(orderId);
        if (order != null && order.isPending()) {
            // 更新订单为处理中
            order.processing();
            orderService.updateOrder(order);
            log.info(\"本地事务执行成功: {}\", transactionId);
        } else {
            throw new RuntimeException(\"订单状态异常: \" + orderId);
        }
    }
    
    // 事务回查
    public LocalTransactionState checkLocalTransaction(Message message) {
        String transactionId = message.getTransactionId();
        Long orderId = Long.valueOf(new String(message.getBody()));
        
        try {
            Order order = orderService.findById(orderId);
            if (order == null) {
                return LocalTransactionState.ROLLBACK_MESSAGE;
            }
            
            if (order.isCompleted() || order.isProcessing()) {
                return LocalTransactionState.COMMIT_MESSAGE;
            } else if (order.isFailed()) {
                return LocalTransactionState.ROLLBACK_MESSAGE;
            } else {
                return LocalTransactionState.UNKNOW;
            }
        } catch (Exception e) {
            log.error(\"事务回查异常: {}\", transactionId, e);
            return LocalTransactionState.UNKNOW;
        }
    }
}

事件驱动架构与最终一致性

1. 基于事件的Saga实现

// 事件发布服务
@Service
@Slf4j
public class DomainEventPublisher {
    
    @Autowired
    private ApplicationEventPublisher eventPublisher;
    
    @Autowired
    private EventStoreRepository eventStoreRepository;
    
    @Transactional
    public void publishWithTransaction(DomainEvent event) {
        // 存储事件（与业务操作在同一个事务中）
        eventStoreRepository.save(event.toEventStore());
        
        // 发布事件
        eventPublisher.publishEvent(event);
        
        log.info(\"领域事件发布: {}\", event.getEventType());
    }
}

// 订单创建事件
@Data
public class OrderCreatedEvent extends DomainEvent {
    private Long orderId;
    private Long userId;
    private BigDecimal amount;
    private List items;
    
    public OrderCreatedEvent(Order order) {
        super(\"ORDER_CREATED\", order.getId().toString());
        this.orderId = order.getId();
        this.userId = order.getUserId();
        this.amount = order.getAmount();
        this.items = order.getItems();
    }
}

// 事件处理器
@Component
@Slf4j
public class OrderCreatedEventHandler {
    
    @Autowired
    private InventoryService inventoryService;
    
    @Autowired
    private AccountService accountService;
    
    @EventListener
    @Async
    public void handleOrderCreated(OrderCreatedEvent event) {
        log.info(\"处理订单创建事件: {}\", event.getOrderId());
        
        try {
            // 扣减库存
            inventoryService.deductStock(event.getItems());
            log.info(\"库存扣减成功: {}\", event.getOrderId());
            
            // 扣减余额
            accountService.deductBalance(event.getUserId(), event.getAmount());
            log.info(\"余额扣减成功: {}\", event.getOrderId());
            
            // 发布订单完成事件
            eventPublisher.publish(new OrderCompletedEvent(event.getOrderId()));
            
        } catch (Exception e) {
            log.error(\"订单处理失败: {}\", event.getOrderId(), e);
            
            // 发布订单失败事件
            eventPublisher.publish(new OrderFailedEvent(event.getOrderId(), e.getMessage()));
        }
    }
}

2. 事件溯源模式

// 事件存储
@Entity
@Table(name = \"event_store\")
@Data
public class EventStore {
    @Id
    private String eventId;
    private String aggregateId;
    private String eventType;
    private String eventData;
    private LocalDateTime timestamp;
    private int version;
}

// 聚合根基类
public abstract class AggregateRoot {
    protected String id;
    protected List changes = new ArrayList();
    protected int version = 0;
    
    public void apply(DomainEvent event) {
        changes.add(event);
        handle(event);
    }
    
    protected abstract void handle(DomainEvent event);
    
    public void markChangesAsCommitted() {
        changes.clear();
    }
    
    public List getUncommittedChanges() {
        return new ArrayList(changes);
    }
}

// 订单聚合根
public class OrderAggregate extends AggregateRoot {
    private OrderStatus status;
    private BigDecimal amount;
    private Long userId;
    
    public OrderAggregate(String orderId, Long userId, BigDecimal amount) {
        this.id = orderId;
        this.userId = userId;
        this.amount = amount;
        this.status = OrderStatus.CREATED;
        
        apply(new OrderCreatedEvent(orderId, userId, amount));
    }
    
    // 重建聚合根
    public OrderAggregate(String orderId, List events) {
        this.id = orderId;
        for (DomainEvent event : events) {
            handle(event);
            this.version++;
        }
    }
    
    public void complete() {
        if (this.status != OrderStatus.CREATED) {
            throw new IllegalStateException(\"订单状态异常\");
        }
        this.status = OrderStatus.COMPLETED;
        apply(new OrderCompletedEvent(this.id));
    }
    
    public void fail(String reason) {
        this.status = OrderStatus.FAILED;
        apply(new OrderFailedEvent(this.id, reason));
    }
    
    @Override
    protected void handle(DomainEvent event) {
        if (event instanceof OrderCreatedEvent) {
            handleOrderCreated((OrderCreatedEvent) event);
        } else if (event instanceof OrderCompletedEvent) {
            handleOrderCompleted((OrderCompletedEvent) event);
        } else if (event instanceof OrderFailedEvent) {
            handleOrderFailed((OrderFailedEvent) event);
        }
    }
    
    private void handleOrderCreated(OrderCreatedEvent event) {
        this.userId = event.getUserId();
        this.amount = event.getAmount();
        this.status = OrderStatus.CREATED;
    }
    
    private void handleOrderCompleted(OrderCompletedEvent event) {
        this.status = OrderStatus.COMPLETED;
    }
    
    private void handleOrderFailed(OrderFailedEvent event) {
        this.status = OrderStatus.FAILED;
    }
}

微服务事务监控

1. 分布式链路追踪

// 事务链路追踪
@Aspect
@Component
@Slf4j
public class DistributedTransactionTracingAspect {
    
    @Autowired
    private Tracer tracer;
    
    @Around(\"@annotation(globalTransactional)\")
    public Object traceGlobalTransaction(ProceedingJoinPoint joinPoint, 
                                       GlobalTransactional globalTransactional) throws Throwable {
        
        Span transactionSpan = tracer.nextSpan().name(\"global-transaction\").start();
        
        try (Tracer.SpanInScope ws = tracer.withSpanInScope(transactionSpan)) {
            // 添加事务标签
            transactionSpan.tag(\"transaction.type\", \"global\");
            transactionSpan.tag(\"transaction.timeout\", 
                String.valueOf(globalTransactional.timeoutMills()));
            
            // 记录事务开始
            log.info(\"全局事务开始: {}\", transactionSpan.context().traceId());
            
            Object result = joinPoint.proceed();
            
            // 记录事务成功
            transactionSpan.finish();
            log.info(\"全局事务完成: {}\", transactionSpan.context().traceId());
            
            return result;
            
        } catch (Exception e) {
            // 记录事务失败
            transactionSpan.error(e);
            transactionSpan.finish();
            log.error(\"全局事务失败: {}\", transactionSpan.context().traceId(), e);
            throw e;
        }
    }
}

// 在Feign调用中传播追踪上下文
@Configuration
public class FeignTracingConfig {
    
    @Bean
    public Feign.Builder feignBuilder(Tracer tracer) {
        return Feign.builder()
            .client(new feign.Client.Default(null, null))
            .requestInterceptor(new ForwardedHeaderInterceptor(tracer));
    }
    
    static class ForwardedHeaderInterceptor implements RequestInterceptor {
        private final Tracer tracer;
        
        ForwardedHeaderInterceptor(Tracer tracer) {
            this.tracer = tracer;
        }
        
        @Override
        public void apply(RequestTemplate template) {
            Span currentSpan = tracer.currentSpan();
            if (currentSpan != null) {
                template.header(\"X-B3-TraceId\", currentSpan.context().traceIdString());
                template.header(\"X-B3-SpanId\", currentSpan.context().spanIdString());
                template.header(\"X-B3-ParentSpanId\", 
                    currentSpan.context().parentIdString());
            }
        }
    }
}

2. 事务健康检查

@Component
public class TransactionHealthIndicator implements HealthIndicator {
    
    @Autowired
    private TransactionMetrics transactionMetrics;
    
    @Autowired
    private DataSource dataSource;
    
    @Override
    public Health health() {
        try {
            // 检查数据库连接
            checkDatabaseConnection();
            
            // 检查事务指标
            Map details = new HashMap();
            details.put(\"activeTransactions\", transactionMetrics.getActiveTransactionCount());
            details.put(\"rollbackRate\", transactionMetrics.getRollbackRate());
            details.put(\"averageDuration\", transactionMetrics.getAverageDuration());
            
            // 判断健康状态
            if (transactionMetrics.getRollbackRate() > 0.1) {
                return Health.down()
                    .withDetails(details)
                    .withException(new RuntimeException(\"事务回滚率过高\"))
                    .build();
            }
            
            if (transactionMetrics.getAverageDuration() > 5000) {
                return Health.down()
                    .withDetails(details)
                    .withException(new RuntimeException(\"事务执行时间过长\"))
                    .build();
            }
            
            return Health.up().withDetails(details).build();
            
        } catch (Exception e) {
            return Health.down(e).build();
        }
    }
    
    private void checkDatabaseConnection() throws SQLException {
        try (Connection conn = dataSource.getConnection()) {
            if (!conn.isValid(5)) {
                throw new SQLException(\"数据库连接无效\");
            }
        }
    }
}

最佳实践与模式选择

1. 模式选择指南

2. 降级和容错策略

@Service
@Slf4j
public class CircuitBreakerTransactionService {
    
    @Autowired
    private InventoryService inventoryService;
    
    @Autowired
    private AccountService accountService;
    
    // 使用断路器保护远程调用
    @CircuitBreaker(name = \"inventoryService\", fallbackMethod = \"fallbackDeductInventory\")
    @Retry(name = \"inventoryService\")
    @TimeLimiter(name = \"inventoryService\")
    public CompletableFuture deductInventoryWithProtection(Order order) {
        return CompletableFuture.supplyAsync(() -> 
            inventoryService.deductStock(order.getItems()));
    }
    
    // 降级策略
    public CompletableFuture fallbackDeductInventory(Order order, Exception e) {
        log.warn(\"库存服务降级，订单进入待处理状态: {}\", order.getId());
        
        // 将订单标记为待处理，后续补偿
        order.markAsPending();
        orderService.updateOrder(order);
        
        // 发布待处理事件
        eventPublisher.publish(new OrderPendingEvent(order.getId()));
        
        return CompletableFuture.completedFuture(true);
    }
    
    // 异步补偿任务
    @Async
    @Scheduled(fixedDelay = 30000) // 每30秒执行一次
    public void compensatePendingOrders() {
        List pendingOrders = orderService.findPendingOrders();
        
        for (Order order : pendingOrders) {
            try {
                // 重试库存扣减
                Boolean result = inventoryService.deductStock(order.getItems());
                if (Boolean.TRUE.equals(result)) {
                    order.complete();
                    orderService.updateOrder(order);
                    log.info(\"补偿处理成功: {}\", order.getId());
                }
            } catch (Exception e) {
                log.error(\"补偿处理失败: {}\", order.getId(), e);
            }
        }
    }
}

总结

微服务架构下的事务管理是一个复杂但重要的话题：

核心要点：

1. 放弃强一致性，拥抱最终一致性
2. 选择合适的模式基于业务需求
3. 设计补偿机制处理失败场景
4. 实施全面监控确保系统可靠性

技术选择：

• 简单场景：事务消息、本地消息表
• 复杂业务：Saga模式、TCC模式
• Spring Cloud生态：Seata框架
• 事件驱动：事件溯源、CQRS

成功关键：

1. 业务分析：理解业务真正的一致性需求
2. 渐进式实施：从简单模式开始，逐步复杂化
3. 监控告警：建立完善的监控体系
4. 团队协作：跨团队的事务协议设计

记住，在微服务架构中，没有银弹解决方案。最重要的是根据具体业务场景选择最适合的事务策略。

下期预告：《Spring 6 & Spring Boot 3新特性：事务管理的革新》

如果觉得本文对你有帮助，请点赞、收藏、关注！欢迎在评论区分享你在微服务事务方面的实战经验和挑战。