登录 注册

微服务架构下分布式事务最佳实践：Saga模式、TCC模式与消息队列补偿机制

算法架构师
算法架构师 2025-12-20T23:13:00+08:00
0 0 0

引言

在微服务架构日益普及的今天，如何保障分布式环境下的事务一致性成为了系统设计的核心挑战之一。传统的单体应用中，事务管理相对简单，但在拆分为多个独立的服务后，跨服务的事务处理变得异常复杂。一个典型的业务场景可能涉及多个微服务的协调操作，任何一个环节的失败都可能导致数据不一致的问题。

分布式事务的本质是在分布式系统环境下，保证一组操作要么全部成功，要么全部失败的一致性机制。在微服务架构中，由于服务间的通信是基于网络的异步调用，传统的ACID事务难以直接应用，因此需要引入专门的分布式事务解决方案。

本文将深入探讨微服务架构下分布式事务的核心解决方案，包括Saga模式、TCC模式以及消息队列补偿机制，并结合实际代码示例和最佳实践，为开发者提供一套完整的事务一致性保障策略。

分布式事务问题分析

什么是分布式事务

分布式事务是指涉及多个独立服务节点的事务操作。在微服务架构中，一个业务操作可能需要同时调用多个服务来完成，这些服务可能部署在不同的服务器上，使用不同的数据库。当其中一个环节出现故障时，如何保证整个业务流程的一致性就成为了一个复杂的问题。

分布式事务的核心挑战

  1. 网络不可靠性：服务间通信可能因网络问题导致超时或失败
  2. 数据不一致性：不同服务的数据存储可能处于不同状态
  3. 事务隔离性：需要保证在事务执行过程中其他操作不会干扰
  4. 性能开销：分布式事务通常会带来额外的网络延迟和处理开销

传统解决方案的局限性

传统的数据库事务（ACID）主要适用于单体应用，无法直接应用于微服务架构。在分布式环境中，需要考虑：

  • 跨服务的数据一致性
  • 网络故障的容错能力
  • 服务间的解耦需求
  • 高可用性和可扩展性要求

Saga模式：长事务的优雅解决方案

Saga模式概述

Saga模式是一种经典的分布式事务解决方案，它将一个长事务拆分为多个短事务，每个短事务都有对应的补偿操作。当某个步骤失败时，通过执行之前已完成步骤的补偿操作来回滚整个流程。

核心思想与原理

Saga模式的核心思想是最终一致性而非强一致性。它通过将复杂的业务流程分解为一系列可独立执行的步骤，并为每个步骤提供相应的撤销机制来实现事务的一致性。

在Saga模式中：

  • 每个服务负责执行自己的业务逻辑
  • 当前服务失败时，通过执行之前步骤的补偿操作来回滚
  • 整个流程通过协调器来管理各个步骤的执行顺序

Saga模式实现方式

1. 基于状态机的实现

public class SagaManager {
    private List<SagaStep> steps;
    private List<SagaStep> executedSteps;
    
    public void executeSaga() {
        try {
            for (SagaStep step : steps) {
                step.execute();
                executedSteps.add(step);
            }
        } catch (Exception e) {
            // 发生异常时执行补偿操作
            rollback();
            throw new RuntimeException("Saga execution failed", e);
        }
    }
    
    private void rollback() {
        // 逆序执行补偿操作
        for (int i = executedSteps.size() - 1; i >= 0; i--) {
            executedSteps.get(i).compensate();
        }
    }
}

public abstract class SagaStep {
    public abstract void execute() throws Exception;
    public abstract void compensate() throws Exception;
}

2. 基于事件驱动的实现

@Component
public class OrderSagaCoordinator {
    
    @Autowired
    private EventPublisher eventPublisher;
    
    @Autowired
    private OrderRepository orderRepository;
    
    // 订单创建步骤
    public void createOrder(OrderRequest request) {
        Order order = new Order();
        order.setId(UUID.randomUUID().toString());
        order.setStatus("CREATED");
        
        // 保存订单状态
        orderRepository.save(order);
        
        // 发布订单创建事件
        eventPublisher.publish(new OrderCreatedEvent(order.getId(), 
            request.getCustomerId(), request.getAmount()));
    }
    
    // 处理订单创建成功的回调
    @EventListener
    public void handleOrderCreated(OrderCreatedEvent event) {
        // 调用库存服务
        inventoryService.reserveStock(event.getProductId(), event.getQuantity());
        
        // 更新订单状态为已确认
        orderRepository.updateStatus(event.getOrderId(), "CONFIRMED");
    }
    
    // 处理库存预留失败的补偿
    @EventListener
    public void handleStockReservationFailed(StockReservationFailedEvent event) {
        // 发布订单取消事件，触发补偿流程
        eventPublisher.publish(new OrderCancelledEvent(event.getOrderId()));
    }
}

Saga模式的优势与局限

优势：

  1. 高可用性：每个步骤独立执行，单点故障不影响整体流程
  2. 可扩展性强：支持水平扩展和分布式部署
  3. 灵活性高：可以根据业务需求调整流程结构
  4. 性能较好：避免了长时间的锁等待

局限性：

  1. 实现复杂度高：需要为每个步骤设计补偿逻辑
  2. 数据一致性风险：在执行过程中可能出现状态不一致
  3. 调试困难：分布式环境下的问题排查较为复杂

TCC模式：两阶段提交的微服务实践

TCC模式详解

TCC（Try-Confirm-Cancel）模式是另一种重要的分布式事务解决方案，它通过将业务操作分为三个阶段来实现分布式一致性：

  1. Try阶段：尝试执行业务操作，进行资源预留
  2. Confirm阶段：确认执行业务操作，正式提交事务
  3. Cancel阶段：取消执行业务操作，释放预留资源

TCC模式的核心组件

public interface TccService {
    /**
     * Try阶段 - 预留资源
     */
    boolean tryExecute(String businessId, Map<String, Object> params);
    
    /**
     * Confirm阶段 - 确认执行
     */
    boolean confirmExecute(String businessId);
    
    /**
     * Cancel阶段 - 取消执行
     */
    boolean cancelExecute(String businessId);
}

@Component
public class AccountTccService implements TccService {
    
    @Autowired
    private AccountRepository accountRepository;
    
    @Override
    public boolean tryExecute(String businessId, Map<String, Object> params) {
        String accountId = (String) params.get("accountId");
        BigDecimal amount = (BigDecimal) params.get("amount");
        
        // Try阶段：检查账户余额并预留资金
        Account account = accountRepository.findById(accountId);
        if (account.getBalance().compareTo(amount) < 0) {
            return false; // 余额不足，预留失败
        }
        
        // 预留资金（冻结部分金额）
        account.setReservedAmount(account.getReservedAmount().add(amount));
        accountRepository.save(account);
        
        return true;
    }
    
    @Override
    public boolean confirmExecute(String businessId) {
        // Confirm阶段：正式扣款
        Account account = accountRepository.findByBusinessId(businessId);
        if (account != null && account.getReservedAmount().compareTo(BigDecimal.ZERO) > 0) {
            account.setBalance(account.getBalance().subtract(account.getReservedAmount()));
            account.setReservedAmount(BigDecimal.ZERO);
            accountRepository.save(account);
            return true;
        }
        return false;
    }
    
    @Override
    public boolean cancelExecute(String businessId) {
        // Cancel阶段：释放预留资金
        Account account = accountRepository.findByBusinessId(businessId);
        if (account != null && account.getReservedAmount().compareTo(BigDecimal.ZERO) > 0) {
            account.setReservedAmount(BigDecimal.ZERO);
            accountRepository.save(account);
            return true;
        }
        return false;
    }
}

TCC模式的完整流程实现

@Component
public class TransferTccCoordinator {
    
    @Autowired
    private AccountTccService accountTccService;
    
    @Autowired
    private TransactionRepository transactionRepository;
    
    public void transfer(String fromAccountId, String toAccountId, BigDecimal amount) {
        String businessId = UUID.randomUUID().toString();
        
        try {
            // 1. Try阶段 - 预留资源
            Map<String, Object> fromParams = new HashMap<>();
            fromParams.put("accountId", fromAccountId);
            fromParams.put("amount", amount);
            
            if (!accountTccService.tryExecute(bbusinessId, fromParams)) {
                throw new RuntimeException("Failed to reserve funds from source account");
            }
            
            Map<String, Object> toParams = new HashMap<>();
            toParams.put("accountId", toAccountId);
            toParams.put("amount", amount);
            
            if (!accountTccService.tryExecute(businessId, toParams)) {
                // 如果目标账户预留失败，需要取消源账户的预留
                accountTccService.cancelExecute(businessId);
                throw new RuntimeException("Failed to reserve funds for target account");
            }
            
            // 2. Confirm阶段 - 确认执行
            accountTccService.confirmExecute(businessId);
            
            // 3. 记录事务完成状态
            transactionRepository.recordSuccess(businessId, "TRANSFER_COMPLETED");
            
        } catch (Exception e) {
            // 如果任何步骤失败，需要执行补偿操作
            try {
                accountTccService.cancelExecute(businessId);
            } catch (Exception cancelEx) {
                log.error("Failed to cancel transaction: " + businessId, cancelEx);
            }
            throw new RuntimeException("Transfer failed", e);
        }
    }
}

TCC模式的事务管理器

@Component
public class TccTransactionManager {
    
    private final Map<String, TccTransaction> activeTransactions = new ConcurrentHashMap<>();
    
    public void startTransaction(String businessId) {
        TccTransaction transaction = new TccTransaction();
        transaction.setId(businessId);
        transaction.setStatus(TransactionStatus.STARTED);
        transaction.setStartTime(System.currentTimeMillis());
        activeTransactions.put(businessId, transaction);
    }
    
    public void completeTryPhase(String businessId) {
        TccTransaction transaction = activeTransactions.get(businessId);
        if (transaction != null) {
            transaction.setStatus(TransactionStatus.TRY_COMPLETED);
            transaction.setTryCompleteTime(System.currentTimeMillis());
        }
    }
    
    public void completeConfirmPhase(String businessId) {
        TccTransaction transaction = activeTransactions.get(businessId);
        if (transaction != null) {
            transaction.setStatus(TransactionStatus.CONFIRMED);
            transaction.setCompleteTime(System.currentTimeMillis());
            activeTransactions.remove(businessId);
        }
    }
    
    public void handleFailure(String businessId) {
        TccTransaction transaction = activeTransactions.get(businessId);
        if (transaction != null && transaction.getStatus() == TransactionStatus.TRY_COMPLETED) {
            // 执行补偿操作
            compensateTransaction(businessId);
            transaction.setStatus(TransactionStatus.FAILED);
        }
    }
    
    private void compensateTransaction(String businessId) {
        // 实现补偿逻辑
        // 这里可以调用各个服务的cancel方法
    }
}

TCC模式的优势与适用场景

优势：

  1. 强一致性：通过两阶段提交保证数据一致性
  2. 业务侵入性小：只需要在原有业务代码基础上增加Try、Confirm、Cancel方法
  3. 可扩展性强：支持复杂的业务流程
  4. 性能较好：相比Saga模式，事务锁定时间较短

适用场景：

  1. 需要强一致性的业务场景
  2. 业务逻辑相对固定的场景
  3. 对数据一致性要求极高的系统
  4. 可以接受一定复杂度的系统架构

消息队列补偿机制：异步处理的优雅方案

消息队列补偿机制概述

消息队列补偿机制是基于消息中间件实现的分布式事务解决方案。通过将事务操作分解为多个步骤，每一步都通过消息队列进行异步通信，当某个步骤失败时，可以通过消息队列的重试机制或补偿机制来保证最终一致性。

核心架构设计

@Component
public class MessageCompensationService {
    
    @Autowired
    private RabbitTemplate rabbitTemplate;
    
    @Autowired
    private TransactionRepository transactionRepository;
    
    // 发送事务消息
    public void sendTransactionMessage(TransactionEvent event) {
        try {
            // 1. 记录事务状态
            transactionRepository.save(event);
            
            // 2. 发送消息到消息队列
            rabbitTemplate.convertAndSend("transaction.exchange", 
                "transaction." + event.getEventType(), event);
                
        } catch (Exception e) {
            log.error("Failed to send transaction message: " + event.getId(), e);
            throw new RuntimeException("Transaction message sending failed", e);
        }
    }
    
    // 处理事务消息
    @RabbitListener(queues = "transaction.queue")
    public void handleTransactionMessage(TransactionEvent event) {
        try {
            // 1. 执行业务逻辑
            executeBusinessLogic(event);
            
            // 2. 更新事务状态为成功
            transactionRepository.updateStatus(event.getId(), "SUCCESS");
            
        } catch (Exception e) {
            log.error("Failed to process transaction: " + event.getId(), e);
            // 3. 发送补偿消息或触发重试机制
            handleCompensation(event, e);
        }
    }
    
    private void executeBusinessLogic(TransactionEvent event) {
        switch (event.getEventType()) {
            case "ORDER_CREATED":
                orderService.createOrder(event.getOrderData());
                break;
            case "INVENTORY_RESERVED":
                inventoryService.reserveStock(event.getInventoryData());
                break;
            case "PAYMENT_PROCESSED":
                paymentService.processPayment(event.getPaymentData());
                break;
        }
    }
    
    private void handleCompensation(TransactionEvent event, Exception exception) {
        // 1. 记录失败信息
        transactionRepository.updateStatus(event.getId(), "FAILED");
        transactionRepository.recordFailure(event.getId(), exception.getMessage());
        
        // 2. 发送补偿消息
        CompensationEvent compensation = new CompensationEvent();
        compensation.setOriginalEventId(event.getId());
        compensation.setCompensationType("REVERSE_" + event.getEventType());
        compensation.setRetryCount(0);
        
        rabbitTemplate.convertAndSend("compensation.exchange", 
            "compensation." + compensation.getCompensationType(), compensation);
    }
}

消息队列补偿机制实现

@Component
public class CompensationProcessor {
    
    @Autowired
    private RabbitTemplate rabbitTemplate;
    
    @Autowired
    private TransactionRepository transactionRepository;
    
    // 补偿消息处理器
    @RabbitListener(queues = "compensation.queue")
    public void handleCompensationMessage(CompensationEvent event) {
        try {
            log.info("Processing compensation for event: " + event.getOriginalEventId());
            
            // 1. 检查是否需要补偿
            if (shouldCompensate(event)) {
                // 2. 执行补偿操作
                executeCompensation(event);
                
                // 3. 更新补偿状态
                transactionRepository.updateCompensationStatus(event.getOriginalEventId(), "COMPLETED");
            }
            
        } catch (Exception e) {
            log.error("Failed to process compensation: " + event.getOriginalEventId(), e);
            // 4. 处理补偿失败，进行重试或报警
            handleCompensationFailure(event, e);
        }
    }
    
    private boolean shouldCompensate(CompensationEvent event) {
        // 检查是否已经执行过补偿
        Transaction transaction = transactionRepository.findById(event.getOriginalEventId());
        return transaction != null && 
               "FAILED".equals(transaction.getStatus()) && 
               !"COMPLETED".equals(transaction.getCompensationStatus());
    }
    
    private void executeCompensation(CompensationEvent event) {
        switch (event.getCompensationType()) {
            case "REVERSE_ORDER_CREATED":
                orderService.cancelOrder(event.getOrderData().getId());
                break;
            case "REVERSE_INVENTORY_RESERVED":
                inventoryService.releaseStock(event.getInventoryData());
                break;
            case "REVERSE_PAYMENT_PROCESSED":
                paymentService.refund(event.getPaymentData().getTransactionId());
                break;
        }
    }
    
    private void handleCompensationFailure(CompensationEvent event, Exception exception) {
        // 1. 增加重试次数
        int retryCount = transactionRepository.getRetryCount(event.getOriginalEventId());
        if (retryCount < 3) {
            // 2. 重新发送补偿消息（延时队列）
            event.setRetryCount(retryCount + 1);
            rabbitTemplate.convertAndSend("compensation.delay.exchange", 
                "compensation.delay." + event.getCompensationType(), event, 
                message -> {
                    message.getMessageProperties().setDelay(5000); // 延迟5秒
                    return message;
                });
        } else {
            // 3. 发送报警通知
            notifyAlert(event, exception);
            transactionRepository.updateCompensationStatus(event.getOriginalEventId(), "FAILED");
        }
    }
    
    private void notifyAlert(CompensationEvent event, Exception exception) {
        // 发送告警通知到监控系统
        AlertMessage alert = new AlertMessage();
        alert.setEventId(event.getOriginalEventId());
        alert.setCompensationType(event.getCompensationType());
        alert.setErrorMessage(exception.getMessage());
        alert.setTimestamp(System.currentTimeMillis());
        
        rabbitTemplate.convertAndSend("alert.exchange", "alert.compensation.failed", alert);
    }
}

消息队列补偿机制的可靠性保障

@Component
public class ReliableMessageService {
    
    @Autowired
    private RabbitTemplate rabbitTemplate;
    
    @Autowired
    private MessageRepository messageRepository;
    
    // 可靠消息发送
    public void sendReliableMessage(String routingKey, Object message, 
                                  MessageCallback callback) {
        try {
            // 1. 持久化消息到数据库
            MessageEntity messageEntity = new MessageEntity();
            messageEntity.setId(UUID.randomUUID().toString());
            messageEntity.setRoutingKey(routingKey);
            messageEntity.setMessageContent(JsonUtils.toJson(message));
            messageEntity.setStatus("PENDING");
            messageEntity.setCreateTime(System.currentTimeMillis());
            
            messageRepository.save(messageEntity);
            
            // 2. 发送消息到消息队列
            rabbitTemplate.convertAndSend("message.exchange", routingKey, message);
            
            // 3. 更新消息状态为已发送
            messageEntity.setStatus("SENT");
            messageRepository.updateStatus(messageEntity.getId(), "SENT");
            
            // 4. 注册回调处理
            if (callback != null) {
                callback.onSuccess(messageEntity);
            }
            
        } catch (Exception e) {
            log.error("Failed to send reliable message", e);
            if (callback != null) {
                callback.onError(e);
            }
        }
    }
    
    // 消息确认机制
    @RabbitListener(queues = "message.confirm.queue")
    public void handleMessageConfirm(MessageConfirm confirm) {
        try {
            // 1. 更新消息状态为已确认
            messageRepository.updateStatus(confirm.getMessageId(), "CONFIRMED");
            
            // 2. 执行后续处理逻辑
            processConfirmedMessage(confirm);
            
        } catch (Exception e) {
            log.error("Failed to process confirmed message: " + confirm.getMessageId(), e);
            // 3. 发送重试消息或报警
            retryOrAlert(confirm, e);
        }
    }
    
    private void processConfirmedMessage(MessageConfirm confirm) {
        // 处理确认后的业务逻辑
        log.info("Message confirmed: " + confirm.getMessageId());
    }
    
    private void retryOrAlert(MessageConfirm confirm, Exception exception) {
        MessageEntity message = messageRepository.findById(confirm.getMessageId());
        if (message != null && message.getRetryCount() < 3) {
            // 重试逻辑
            message.setRetryCount(message.getRetryCount() + 1);
            messageRepository.updateRetryCount(message.getId(), message.getRetryCount());
            
            // 重新发送消息
            rabbitTemplate.convertAndSend("message.exchange", 
                message.getRoutingKey(), JsonUtils.fromJson(message.getMessageContent(), Object.class));
        } else {
            // 发送告警
            log.error("Message processing failed after retries: " + confirm.getMessageId());
        }
    }
}

三种模式的对比分析与选型建议

性能对比

特性 Saga模式 TCC模式 消息队列补偿
响应时间 较快 中等 较慢
并发性能 高 中等 高
实现复杂度 中等 高 低
数据一致性 最终一致 强一致 最终一致
网络依赖 中等 高 高

适用场景分析

Saga模式适用于：

  1. 业务流程相对简单且固定
  2. 对强一致性要求不是特别严格
  3. 需要较高的系统吞吐量
  4. 服务间依赖关系清晰

TCC模式适用于：

  1. 对数据一致性要求极高
  2. 业务流程复杂且需要精确控制
  3. 可以接受较高的实现复杂度
  4. 事务涉及的资源需要精确预留和释放

消息队列补偿适用于：

  1. 异步处理需求强烈
  2. 需要高可用性和容错能力
  3. 系统架构中已存在消息中间件
  4. 对实时性要求不高的场景

实际应用建议

@Configuration
public class DistributedTransactionConfig {
    
    @Bean
    @Primary
    public TransactionStrategyFactory transactionStrategyFactory() {
        return new TransactionStrategyFactory() {
            @Override
            public TransactionStrategy getStrategy(String businessType) {
                switch (businessType) {
                    case "TRANSFER":
                        return new TccTransactionStrategy();
                    case "ORDER_PROCESSING":
                        return new SagaTransactionStrategy();
                    case "INVENTORY_MANAGEMENT":
                        return new MessageCompensationStrategy();
                    default:
                        return new DefaultTransactionStrategy();
                }
            }
        };
    }
    
    // 策略模式实现
    public interface TransactionStrategy {
        void execute(TransactionContext context) throws Exception;
        void compensate(TransactionContext context) throws Exception;
    }
    
    public class TccTransactionStrategy implements TransactionStrategy {
        @Override
        public void execute(TransactionContext context) throws Exception {
            // TCC执行逻辑
            new TccTransactionManager().execute(context);
        }
        
        @Override
        public void compensate(TransactionContext context) throws Exception {
            // TCC补偿逻辑
            new TccTransactionManager().compensate(context);
        }
    }
    
    public class SagaTransactionStrategy implements TransactionStrategy {
        @Override
        public void execute(TransactionContext context) throws Exception {
            // Saga执行逻辑
            new SagaManager().execute(context);
        }
        
        @Override
        public void compensate(TransactionContext context) throws Exception {
            // Saga补偿逻辑
            new SagaManager().rollback(context);
        }
    }
}

生产环境最佳实践

1. 容错机制设计

@Component
public class FaultTolerantTransactionManager {
    
    private static final int MAX_RETRY_COUNT = 3;
    private static final long RETRY_DELAY_MS = 5000;
    
    public void executeWithRetry(TransactionContext context) {
        int retryCount = 0;
        Exception lastException = null;
        
        while (retryCount < MAX_RETRY_COUNT) {
            try {
                executeTransaction(context);
                return; // 成功执行，退出循环
            } catch (Exception e) {
                lastException = e;
                retryCount++;
                
                if (retryCount >= MAX_RETRY_COUNT) {
                    throw new RuntimeException("Transaction execution failed after " + 
                        MAX_RETRY_COUNT + " retries", e);
                }
                
                log.warn("Transaction execution failed, retrying... (attempt: " + retryCount + ")", e);
                sleep(RETRY_DELAY_MS * retryCount); // 指数退避
            }
        }
    }
    
    private void executeTransaction(TransactionContext context) throws Exception {
        // 实际的事务执行逻辑
        if (context.getStrategy() == TransactionStrategy.TCC) {
            executeTccTransaction(context);
        } else if (context.getStrategy() == TransactionStrategy.SAGA) {
            executeSagaTransaction(context);
        }
    }
    
    private void sleep(long millis) {
        try {
            Thread.sleep(millis);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            throw new RuntimeException("Interrupted during sleep", e);
        }
    }
}

2. 监控与告警

@Component
public class TransactionMonitor {
    
    private final MeterRegistry meterRegistry;
    private final Counter transactionSuccessCounter;
    private final Counter transactionFailureCounter;
    private final Timer transactionTimer;
    
    public TransactionMonitor(MeterRegistry meterRegistry) {
        this.meterRegistry = meterRegistry;
        
        this.transactionSuccessCounter = Counter.builder("transaction.success")
            .description("Successful transactions")
            .register(meterRegistry);
            
        this.transactionFailureCounter = Counter.builder("transaction.failure")
            .description("Failed transactions")
            .register(meterRegistry);
            
        this.transactionTimer = Timer.builder("transaction.duration")
            .description("Transaction execution duration")
            .register(meterRegistry);
    }
    
    public void recordSuccess(String transactionId, long duration) {
        transactionSuccessCounter.increment();
        transactionTimer.record(duration, TimeUnit.MILLISECONDS);
        
        log.info("Transaction completed successfully: {}", transactionId);
    }
    
    public void recordFailure(String transactionId, Exception exception) {
        transactionFailureCounter.increment();
        
        // 发送告警
        sendAlert(transactionId, exception);
        
        log.error("Transaction failed: {}", transactionId, exception);
    }
    
    private void sendAlert(String transactionId, Exception exception) {
        // 实现告警逻辑，可以发送到监控系统、邮件、短信等
        AlertService.sendAlert(AlertType.TRANSACTION_FAILURE, 
            "Transaction " + transactionId + " failed: " + exception.getMessage());
    }
}

3. 数据一致性保障

@Component
public class ConsistencyChecker {
    
    @Autowired
    private TransactionRepository transactionRepository;
    
    @Autowired
    private EventPublisher eventPublisher;
    
    // 定期检查事务一致性
    @Scheduled(fixedRate = 300000) // 每5分钟检查一次
    public void checkConsistency() {
        List<Transaction> pendingTransactions = 
            transactionRepository.findPendingTransactions();
            
        for (Transaction transaction : pendingTransactions) {
            try {
                if (isTransactionTimeout(transaction)) {
                    handleTimeoutTransaction(transaction);
                } else if (!isTransactionConsistent(transaction)) {
                    triggerCompensation(transaction);
                }
            } catch (Exception e) {
                log.error("Error checking consistency for transaction: " + 
                    transaction.getId(), e);
            }
        }
    }
    
    private boolean isTransactionTimeout(Transaction transaction) {
        long currentTime = System.currentTimeMillis();
        return currentTime - transaction.getCreateTime() > 3600000; // 1小时超时
    }
    
    private void handleTimeoutTransaction(Transaction transaction) {
        log.warn("Transaction timeout detected: {}", transaction.getId());
        
        // 发送超时告警
        eventPublisher.publish(new TransactionTimeoutEvent(transaction.getId()));
        
        // 触发补偿机制
        triggerCompensation(transaction);
    }
    
    private boolean isTransactionConsistent
# 微服务 # 分布式事务 # Saga模式
相关推荐
广告位招租

相似文章

    评论 (0)

    0/2000