数据一致性与事务管理

在微服务架构中，数据一致性是一个复杂而关键的挑战。由于每个服务管理自己的数据存储，跨服务的业务操作需要协调多个独立的数据库，这使得传统的ACID事务难以直接应用。本章将深入探讨微服务架构中的数据一致性问题、事务管理模式以及确保分布式系统数据完整性的最佳实践。

数据一致性挑战

分布式数据管理模型

在微服务架构中，每个服务拥有独立的数据存储，这种设计带来了诸多优势，如技术栈独立、可扩展性强等，但也引入了数据一致性的复杂性。

// 订单服务 - 管理订单数据
@Service
public class OrderService {
    @Autowired
    private OrderRepository orderRepository; // 独立的订单数据库
    
    public Order createOrder(OrderRequest request) {
        Order order = new Order(request);
        return orderRepository.save(order);
    }
}

// 库存服务 - 管理库存数据
@Service
public class InventoryService {
    @Autowired
    private InventoryRepository inventoryRepository; // 独立的库存数据库
    
    public boolean reserveInventory(String productId, int quantity) {
        Inventory inventory = inventoryRepository.findByProductId(productId);
        if (inventory.getAvailableQuantity() >= quantity) {
            inventory.setReservedQuantity(inventory.getReservedQuantity() + quantity);
            inventory.setAvailableQuantity(inventory.getAvailableQuantity() - quantity);
            inventoryRepository.save(inventory);
            return true;
        }
        return false;
    }
}

// 支付服务 - 管理支付数据
@Service
public class PaymentService {
    @Autowired
    private PaymentRepository paymentRepository; // 独立的支付数据库
    
    public Payment processPayment(String orderId, BigDecimal amount) {
        Payment payment = new Payment(orderId, amount);
        return paymentRepository.save(payment);
    }
}

一致性级别

在分布式系统中，不同业务场景对一致性的要求不同，需要根据具体需求选择合适的一致性模型。

// 一致性级别定义
public enum ConsistencyLevel {
    // 强一致性 - 数据更新后立即对所有后续访问可见
    STRONG("Strong", "数据更新后立即一致"),
    
    // 弱一致性 - 数据更新后不保证立即可见
    WEAK("Weak", "数据更新后可能延迟可见"),
    
    // 最终一致性 - 数据更新后经过一段时间最终达到一致状态
    EVENTUAL("Eventual", "数据最终会达到一致状态"),
    
    // 因果一致性 - 有因果关系的操作保证顺序一致性
    CAUSAL("Causal", "因果相关的操作保证顺序");
    
    private final String level;
    private final String description;
    
    ConsistencyLevel(String level, String description) {
        this.level = level;
        this.description = description;
    }
    
    // 根据业务场景选择一致性级别
    public static ConsistencyLevel selectForBusinessScenario(BusinessScenario scenario) {
        switch (scenario.getType()) {
            case FINANCIAL_TRANSACTION:
                return STRONG;  // 金融交易需要强一致性
            case USER_PROFILE_UPDATE:
                return EVENTUAL; // 用户资料更新可以接受最终一致性
            case ORDER_PROCESSING:
                return CAUSAL;   // 订单处理需要因果一致性
            default:
                return EVENTUAL;
        }
    }
}

分布式事务模式

两阶段提交（2PC）

两阶段提交是最经典的分布式事务协议，但在微服务架构中存在明显局限。

// 简化的两阶段提交实现
public class TwoPhaseCommitManager {
    private List<TransactionParticipant> participants;
    private TransactionLogger transactionLogger;
    
    public boolean executeDistributedTransaction(List<Operation> operations) {
        String transactionId = UUID.randomUUID().toString();
        transactionLogger.logTransactionStart(transactionId, operations);
        
        try {
            // 第一阶段：准备阶段
            PrepareResult prepareResult = preparePhase(transactionId, operations);
            if (!prepareResult.isSuccessful()) {
                rollbackPhase(transactionId);
                transactionLogger.logTransactionFailed(transactionId, prepareResult.getErrors());
                return false;
            }
            
            // 第二阶段：提交阶段
            CommitResult commitResult = commitPhase(transactionId);
            if (commitResult.isSuccessful()) {
                transactionLogger.logTransactionCommitted(transactionId);
                return true;
            } else {
                transactionLogger.logTransactionFailed(transactionId, commitResult.getErrors());
                return false;
            }
        } catch (Exception e) {
            transactionLogger.logTransactionError(transactionId, e);
            rollbackPhase(transactionId);
            return false;
        }
    }
    
    private PrepareResult preparePhase(String transactionId, List<Operation> operations) {
        List<PrepareResponse> responses = new ArrayList<>();
        
        for (TransactionParticipant participant : participants) {
            try {
                PrepareResponse response = participant.prepare(transactionId, operations);
                responses.add(response);
            } catch (Exception e) {
                return new PrepareResult(false, Arrays.asList(e.getMessage()));
            }
        }
        
        boolean allPrepared = responses.stream()
            .allMatch(PrepareResponse::isSuccessful);
            
        List<String> errors = responses.stream()
            .filter(response -> !response.isSuccessful())
            .flatMap(response -> response.getErrors().stream())
            .collect(Collectors.toList());
            
        return new PrepareResult(allPrepared, errors);
    }
    
    private CommitResult commitPhase(String transactionId) {
        List<CommitResponse> responses = new ArrayList<>();
        
        for (TransactionParticipant participant : participants) {
            try {
                CommitResponse response = participant.commit(transactionId);
                responses.add(response);
            } catch (Exception e) {
                return new CommitResult(false, Arrays.asList(e.getMessage()));
            }
        }
        
        boolean allCommitted = responses.stream()
            .allMatch(CommitResponse::isSuccessful);
            
        List<String> errors = responses.stream()
            .filter(response -> !response.isSuccessful())
            .flatMap(response -> response.getErrors().stream())
            .collect(Collectors.toList());
            
        return new CommitResult(allCommitted, errors);
    }
    
    private void rollbackPhase(String transactionId) {
        for (TransactionParticipant participant : participants) {
            try {
                participant.rollback(transactionId);
            } catch (Exception e) {
                transactionLogger.logRollbackError(transactionId, participant, e);
            }
        }
    }
}

Saga模式

Saga模式是微服务架构中处理长事务的主要模式，它将一个长事务分解为多个本地事务。

// Saga编排器
@Component
public class OrderSagaOrchestrator {
    @Autowired
    private OrderService orderService;
    
    @Autowired
    private InventoryService inventoryService;
    
    @Autowired
    private PaymentService paymentService;
    
    @Autowired
    private NotificationService notificationService;
    
    // 创建订单的Saga
    public OrderProcessResult createOrder(OrderRequest request) {
        String sagaId = UUID.randomUUID().toString();
        SagaContext context = new SagaContext(sagaId, request);
        
        try {
            // 步骤1：创建订单
            Order order = orderService.createOrder(request);
            context.setOrder(order);
            logSagaStep(sagaId, "CREATE_ORDER", "SUCCESS", order.getId());
            
            // 步骤2：预留库存
            boolean inventoryReserved = inventoryService.reserveInventory(
                request.getProductId(), request.getQuantity());
            context.setInventoryReserved(inventoryReserved);
            logSagaStep(sagaId, "RESERVE_INVENTORY", 
                       inventoryReserved ? "SUCCESS" : "FAILED");
            
            if (!inventoryReserved) {
                throw new InsufficientInventoryException();
            }
            
            // 步骤3：处理支付
            Payment payment = paymentService.processPayment(
                order.getId(), request.getAmount());
            context.setPayment(payment);
            logSagaStep(sagaId, "PROCESS_PAYMENT", "SUCCESS", payment.getId());
            
            // 步骤4：确认订单
            orderService.confirmOrder(order.getId());
            logSagaStep(sagaId, "CONFIRM_ORDER", "SUCCESS");
            
            // 步骤5：发送确认通知
            notificationService.sendOrderConfirmation(order);
            logSagaStep(sagaId, "SEND_NOTIFICATION", "SUCCESS");
            
            return new OrderProcessResult(true, order, null);
            
        } catch (Exception e) {
            // 执行补偿操作
            compensate(context);
            logSagaStep(sagaId, "SAGA_COMPENSATED", "SUCCESS");
            return new OrderProcessResult(false, null, e);
        }
    }
    
    private void compensate(SagaContext context) {
        try {
            // 逆向执行补偿操作（按相反顺序）
            
            // 补偿通知发送（如果已发送）
            if (context.isNotificationSent()) {
                notificationService.sendOrderCancellation(context.getOrder());
            }
            
            // 补偿订单确认
            if (context.isOrderConfirmed()) {
                orderService.cancelOrder(context.getOrder().getId());
            }
            
            // 补偿支付处理
            if (context.getPayment() != null) {
                paymentService.refund(context.getPayment().getId());
            }
            
            // 补偿库存预留
            if (context.isInventoryReserved() && context.getOrder() != null) {
                inventoryService.releaseInventory(
                    context.getOrder().getProductId(), 
                    context.getOrder().getQuantity());
            }
            
            // 补偿订单创建
            if (context.getOrder() != null) {
                orderService.deleteOrder(context.getOrder().getId());
            }
            
        } catch (Exception e) {
            log.error("补偿操作失败", e);
            // 记录补偿失败，可能需要人工干预
        }
    }
    
    private void logSagaStep(String sagaId, String step, String status) {
        logSagaStep(sagaId, step, status, null);
    }
    
    private void logSagaStep(String sagaId, String step, String status, String referenceId) {
        SagaLogEntry logEntry = new SagaLogEntry(sagaId, step, status, referenceId);
        sagaLogRepository.save(logEntry);
    }
}

// 基于事件的Saga模式
@Component
public class EventDrivenSagaManager {
    @Autowired
    private EventPublisher eventPublisher;
    
    @Autowired
    private SagaRepository sagaRepository;
    
    // 启动Saga
    public void startOrderSaga(OrderRequest request) {
        String sagaId = UUID.randomUUID().toString();
        OrderSaga saga = new OrderSaga(sagaId, request);
        sagaRepository.save(saga);
        
        // 发布创建订单事件
        OrderCreatedEvent event = new OrderCreatedEvent(sagaId, request);
        eventPublisher.publish("order-created", event);
    }
    
    // 处理订单创建完成事件
    @EventListener
    public void handleOrderCreated(OrderCreatedEvent event) {
        updateSagaState(event.getSagaId(), SagaState.ORDER_CREATED);
        
        // 发布预留库存事件
        ReserveInventoryEvent reserveEvent = new ReserveInventoryEvent(
            event.getSagaId(), 
            event.getRequest().getProductId(), 
            event.getRequest().getQuantity());
        eventPublisher.publish("reserve-inventory", reserveEvent);
    }
    
    // 处理库存预留完成事件
    @EventListener
    public void handleInventoryReserved(InventoryReservedEvent event) {
        updateSagaState(event.getSagaId(), SagaState.INVENTORY_RESERVED);
        
        // 发布处理支付事件
        ProcessPaymentEvent paymentEvent = new ProcessPaymentEvent(
            event.getSagaId(), 
            event.getOrderId(), 
            event.getAmount());
        eventPublisher.publish("process-payment", paymentEvent);
    }
    
    // 处理支付完成事件
    @EventListener
    public void handlePaymentProcessed(PaymentProcessedEvent event) {
        updateSagaState(event.getSagaId(), SagaState.PAYMENT_PROCESSED);
        
        // 发布确认订单事件
        ConfirmOrderEvent confirmEvent = new ConfirmOrderEvent(
            event.getSagaId(), 
            event.getOrderId());
        eventPublisher.publish("confirm-order", confirmEvent);
    }
    
    // 处理Saga完成事件
    @EventListener
    public void handleSagaCompleted(SagaCompletedEvent event) {
        completeSaga(event.getSagaId());
    }
    
    // 处理失败事件并执行补偿
    @EventListener
    public void handleSagaFailed(SagaFailedEvent event) {
        OrderSaga saga = sagaRepository.findById(event.getSagaId());
        compensateSaga(saga, event.getFailedStep(), event.getErrorMessage());
    }
    
    private void updateSagaState(String sagaId, SagaState newState) {
        OrderSaga saga = sagaRepository.findById(sagaId);
        saga.updateState(newState);
        sagaRepository.save(saga);
    }
}

事件驱动的一致性保证

事件溯源与CQRS

事件溯源和CQRS模式可以有效解决数据一致性问题。

// 事件发布确保一致性
@Service
@Transactional
public class OrderService {
    @Autowired
    private OrderRepository orderRepository;
    
    @Autowired
    private EventPublisher eventPublisher;
    
    public Order createOrder(OrderRequest request) {
        // 1. 创建订单实体
        Order order = new Order(request);
        order = orderRepository.save(order);
        
        // 2. 发布订单创建事件（在同一个事务中）
        OrderCreatedEvent event = new OrderCreatedEvent(order);
        eventPublisher.publish("order-events", event);
        
        return order;
    }
    
    // 通过事件重建状态
    public Order rebuildOrder(String orderId) {
        List<OrderEvent> events = eventStore.getEventsForAggregate(orderId);
        Order order = new Order();
        for (OrderEvent event : events) {
            order.apply(event);
        }
        return order;
    }
}

// CQRS读写分离
@Component
public class OrderCommandHandler {
    @Autowired
    private OrderWriteRepository orderWriteRepository;
    
    @Autowired
    private EventPublisher eventPublisher;
    
    public void handle(CreateOrderCommand command) {
        Order order = new Order(command);
        orderWriteRepository.save(order);
        
        OrderCreatedEvent event = new OrderCreatedEvent(order);
        eventPublisher.publish("order-created", event);
    }
}

@Component
public class OrderQueryHandler {
    @Autowired
    private OrderReadRepository orderReadRepository;
    
    public OrderDetailView handle(GetOrderQuery query) {
        return orderReadRepository.findOrderDetailView(query.getOrderId());
    }
}

// 读模型投影器
@Component
public class OrderDetailViewProjector {
    @Autowired
    private OrderReadRepository orderReadRepository;
    
    @EventListener
    public void handle(OrderCreatedEvent event) {
        OrderDetailView view = new OrderDetailView();
        view.setId(event.getOrderId());
        view.setCustomerId(event.getCustomerId());
        view.setAmount(event.getAmount());
        view.setStatus(OrderStatus.CREATED);
        view.setCreatedAt(event.getTimestamp());
        
        orderReadRepository.save(view);
    }
    
    @EventListener
    public void handle(OrderConfirmedEvent event) {
        OrderDetailView view = orderReadRepository.findById(event.getOrderId());
        view.setStatus(OrderStatus.CONFIRMED);
        view.setUpdatedAt(event.getTimestamp());
        
        orderReadRepository.save(view);
    }
}

幂等性处理

确保事件处理的幂等性是保证数据一致性的关键。

@Component
public class IdempotentEventHandler {
    @Autowired
    private OrderRepository orderRepository;
    
    @Autowired
    private ProcessedEventRepository processedEventRepository;
    
    @EventListener
    public void handleOrderCreated(OrderCreatedEvent event) {
        // 幂等性检查
        if (processedEventRepository.existsByEventId(event.getEventId())) {
            // 事件已处理，跳过处理
            log.info("Event already processed: {}", event.getEventId());
            return;
        }
        
        try {
            // 创建订单
            Order order = new Order(event);
            orderRepository.save(order);
            
            // 记录已处理的事件
            ProcessedEvent processedEvent = new ProcessedEvent(
                event.getEventId(), 
                event.getClass().getSimpleName(), 
                System.currentTimeMillis());
            processedEventRepository.save(processedEvent);
            
        } catch (Exception e) {
            log.error("Failed to process order created event: {}", event.getEventId(), e);
            // 记录处理失败的事件，便于后续重试
            FailedEvent failedEvent = new FailedEvent(
                event.getEventId(), 
                event.getClass().getSimpleName(), 
                e.getMessage(), 
                System.currentTimeMillis());
            failedEventRepository.save(failedEvent);
            throw e;
        }
    }
    
    @EventListener
    public void handleOrderConfirmed(OrderConfirmedEvent event) {
        // 检查订单是否存在
        Order order = orderRepository.findById(event.getOrderId());
        if (order == null) {
            throw new OrderNotFoundException(event.getOrderId());
        }
        
        // 幂等性检查
        if (order.getStatus() == OrderStatus.CONFIRMED) {
            // 订单已确认，跳过处理
            log.info("Order already confirmed: {}", event.getOrderId());
            return;
        }
        
        order.setStatus(OrderStatus.CONFIRMED);
        order.setConfirmedAt(event.getTimestamp());
        orderRepository.save(order);
        
        // 记录已处理的事件
        ProcessedEvent processedEvent = new ProcessedEvent(
            event.getEventId(), 
            event.getClass().getSimpleName(), 
            System.currentTimeMillis());
        processedEventRepository.save(processedEvent);
    }
}

分布式事务最佳实践

事务边界设计

合理设计事务边界是实现数据一致性的基础。

// 正确的业务聚合设计
public class OrderAggregate {
    private String orderId;
    private String customerId;
    private List<OrderItem> items;
    private OrderStatus status;
    private BigDecimal totalAmount;
    private LocalDateTime createdAt;
    private LocalDateTime updatedAt;
    
    // 订单相关的所有操作都在同一个聚合内
    public void addItem(Product product, int quantity) {
        // 添加商品项
        OrderItem item = new OrderItem(product, quantity);
        items.add(item);
        // 更新总金额
        totalAmount = totalAmount.add(item.getSubtotal());
        // 更新时间戳
        updatedAt = LocalDateTime.now();
    }
    
    public void applyDiscount(Discount discount) {
        // 应用折扣
        BigDecimal discountAmount = totalAmount.multiply(discount.getRate());
        totalAmount = totalAmount.subtract(discountAmount);
        updatedAt = LocalDateTime.now();
    }
    
    public void confirm() {
        // 确认订单
        this.status = OrderStatus.CONFIRMED;
        this.updatedAt = LocalDateTime.now();
    }
    
    public void cancel() {
        // 取消订单
        this.status = OrderStatus.CANCELLED;
        this.updatedAt = LocalDateTime.now();
    }
}

// 聚合根约束
public class OrderAggregateConstraints {
    // 业务不变量检查
    public void validateInvariants(OrderAggregate aggregate) {
        // 检查订单项数量限制
        if (aggregate.getItems().size() > 100) {
            throw new BusinessRuleViolationException("Order cannot have more than 100 items");
        }
        
        // 检查订单金额限制
        if (aggregate.getTotalAmount().compareTo(new BigDecimal("100000")) > 0) {
            throw new BusinessRuleViolationException("Order amount cannot exceed 100,000");
        }
        
        // 检查订单状态转换合法性
        if (aggregate.getStatus() == OrderStatus.CANCELLED && 
            aggregate.getUpdatedAt().isBefore(aggregate.getCreatedAt().plusHours(24))) {
            throw new BusinessRuleViolationException("Order cannot be cancelled after 24 hours");
        }
    }
}

补偿事务设计

设计完善的补偿事务机制是Saga模式成功的关键。

// 补偿事务管理器
@Component
public class CompensationManager {
    private Map<String, CompensatingAction> compensatingActions;
    
    public interface CompensatingAction {
        void compensate(String transactionId, Object context);
    }
    
    // 注册补偿操作
    public void registerCompensation(String step, CompensatingAction action) {
        compensatingActions.put(step, action);
    }
    
    // 执行补偿
    public void executeCompensation(List<CompensationStep> steps) {
        // 逆序执行补偿操作
        Collections.reverse(steps);
        for (CompensationStep step : steps) {
            CompensatingAction action = compensatingActions.get(step.getStepName());
            if (action != null) {
                try {
                    action.compensate(step.getTransactionId(), step.getContext());
                } catch (Exception e) {
                    log.error("Compensation failed for step: " + step.getStepName(), e);
                    // 记录补偿失败，可能需要人工干预
                    recordCompensationFailure(step, e);
                }
            }
        }
    }
    
    // 补偿操作实现示例
    @Component
    public class InventoryReservationCompensator implements CompensatingAction {
        @Autowired
        private InventoryService inventoryService;
        
        @Override
        public void compensate(String transactionId, Object context) {
            if (context instanceof InventoryReservationContext) {
                InventoryReservationContext reservationContext = 
                    (InventoryReservationContext) context;
                inventoryService.releaseInventory(
                    reservationContext.getProductId(), 
                    reservationContext.getQuantity());
            }
        }
    }
    
    @Component
    public class PaymentProcessingCompensator implements CompensatingAction {
        @Autowired
        private PaymentService paymentService;
        
        @Override
        public void compensate(String transactionId, Object context) {
            if (context instanceof PaymentProcessingContext) {
                PaymentProcessingContext paymentContext = 
                    (PaymentProcessingContext) context;
                paymentService.refund(paymentContext.getPaymentId());
            }
        }
    }
}

超时与重试机制

@Component
public class RetryableEventHandler {
    @EventListener
    @Retryable(
        value = {Exception.class}, 
        maxAttempts = 3, 
        backoff = @Backoff(delay = 1000, multiplier = 2)
    )
    public void handleOrderCreated(OrderCreatedEvent event) {
        try {
            // 处理订单创建事件
            processOrderCreation(event);
        } catch (Exception e) {
            log.error("Failed to process order creation event", e);
            // 发布失败事件触发补偿
            SagaFailedEvent failedEvent = new SagaFailedEvent(
                event.getSagaId(), 
                "order-creation", 
                e.getMessage());
            eventPublisher.publish("saga-failed", failedEvent);
            throw e;
        }
    }
    
    @Recover
    public void recover(Exception e, OrderCreatedEvent event) {
        log.error("Failed to process order creation after retries", e);
        // 发布永久失败事件
        SagaPermanentlyFailedEvent failedEvent = new SagaPermanentlyFailedEvent(
            event.getSagaId(), 
            "order-creation", 
            e.getMessage());
        eventPublisher.publish("saga-permanently-failed", failedEvent);
        
        // 通知管理员进行人工处理
        alertService.sendAlert("Order Creation Permanently Failed", 
                              "Order: " + event.getOrder().getId() + 
                              ", Error: " + e.getMessage());
    }
    
    // 死信队列处理
    @RabbitListener(queues = "order-events-dlq")
    public void handleDeadLetter(OrderEvent event, 
                                @Header("x-death") List<Map<String, Object>> deathHeaders) {
        // 记录死信事件
        log.error("Dead letter event received: " + event);
        
        // 分析失败原因
        String failureReason = analyzeFailureReason(deathHeaders);
        
        // 记录到专门的失败表
        failedEventRepository.save(new FailedEvent(event, failureReason));
        
        // 根据失败类型决定处理方式
        if (isRetryable(failureReason)) {
            // 重新入队处理
            requeueEvent(event);
        } else {
            // 需要人工干预
            notifyAdminForManualHandling(event, failureReason);
        }
    }
}

监控与故障处理

一致性监控

@Component
public class ConsistencyMonitor {
    private MeterRegistry meterRegistry;
    
    public void recordInconsistency(String service, String type) {
        Counter.builder("data.inconsistency")
               .tag("service", service)
               .tag("type", type)
               .register(meterRegistry)
               .increment();
    }
    
    @Scheduled(fixedRate = 300000) // 每5分钟检查一次
    public void checkDataConsistency() {
        // 检查各服务间数据一致性
        List<Inconsistency> inconsistencies = consistencyChecker.checkAll();
        
        for (Inconsistency inconsistency : inconsistencies) {
            recordInconsistency(inconsistency.getService(), 
                              inconsistency.getType());
            
            // 发送告警
            alertService.sendAlert("Data Inconsistency Detected", 
                                 inconsistency.toString());
        }
    }
    
    // 数据一致性检查器
    @Component
    public class DataConsistencyChecker {
        public List<Inconsistency> checkAll() {
            List<Inconsistency> inconsistencies = new ArrayList<>();
            
            // 检查订单与支付数据一致性
            inconsistencies.addAll(checkOrderPaymentConsistency());
            
            // 检查库存与订单数据一致性
            inconsistencies.addAll(checkInventoryOrderConsistency());
            
            // 检查用户与订单数据一致性
            inconsistencies.addAll(checkUserOrderConsistency());
            
            return inconsistencies;
        }
        
        private List<Inconsistency> checkOrderPaymentConsistency() {
            List<Inconsistency> inconsistencies = new ArrayList<>();
            
            // 查找已创建但未支付的订单
            List<Order> unpaidOrders = orderRepository.findUnpaidOrders(
                LocalDateTime.now().minusHours(1));
                
            for (Order order : unpaidOrders) {
                Payment payment = paymentRepository.findByOrderId(order.getId());
                if (payment == null) {
                    inconsistencies.add(new Inconsistency(
                        "order-service", 
                        "payment-service", 
                        "missing-payment", 
                        "Order " + order.getId() + " has no payment record"));
                }
            }
            
            return inconsistencies;
        }
    }
}

通过合理运用这些数据一致性与事务管理技术，可以在保持微服务架构优势的同时，确保分布式系统的数据完整性。关键是要根据具体业务场景选择合适的一致性模型和事务处理模式，并建立完善的监控和故障处理机制。