第九章数据一致性与分布式事务

news/2026/1/22 21:44:59/文章来源:https://www.cnblogs.com/qxnm/p/19519098

第九章数据一致性与分布式事务

在微服务架构中，数据一致性是最具挑战性的问题之一。当你把一个单体应用拆分成多个服务时，原本简单的数据库事务变成了复杂的分布式事务。这一章，我想分享一些实战经验，帮助你在这个"没有银弹"的领域做出明智的决策。

9.1 分布式事务的现实挑战

9.1.1 为什么传统ACID事务不再适用

还记得我第一次做微服务系统时，天真地以为可以用分布式事务协调器来解决所有问题。结果证明，这是一个巨大的错误。

传统2PC（两阶段提交）的问题

// ❌ 这就是问题所在 - 阻塞和资源锁定
public class DistributedOrderService
{public async Task CreateOrder(CreateOrderRequest request){using (var transaction = new TransactionScope(TransactionScopeOption.Required,new TransactionOptions { IsolationLevel = IsolationLevel.Serializable })){// 1. 扣减库存 - 锁定库存表await _inventoryService.DeductStock(request.Items);// 2. 创建订单 - 锁定订单表var order = await _orderRepository.CreateAsync(request);// 3. 处理支付 - 锁定账户表await _paymentService.ProcessPayment(order.TotalAmount);// 在所有操作完成之前，所有相关的表都被锁定// 其他请求必须等待，这就是性能问题的根源transaction.Complete();}}
}

2PC的问题不在于它不能工作，而在于它违背了微服务的核心原则：

可用性牺牲：协调器故障会导致整个系统阻塞
性能问题：资源锁定时间过长
扩展性差：无法支持大规模分布式系统

9.1.2 接受最终一致性

在分布式系统中，我们需要接受一个现实：强一致性是有代价的。大多数情况下，最终一致性已经足够。

电商系统的最终一致性示例

// 订单创建流程 - 最终一致性版本
public class OrderApplicationService
{public async Task<CreateOrderResult> CreateOrder(CreateOrderCommand command){// 1. 本地事务：创建订单using var transaction = await _dbContext.Database.BeginTransactionAsync();try{var order = Order.Create(command.CustomerId, command.Items);await _orderRepository.AddAsync(order);await _dbContext.SaveChangesAsync();// 2. 发布领域事件（本地事务的一部分）await _eventPublisher.PublishAsync(new OrderCreatedEvent(order.Id,order.CustomerId,order.Items.Select(i => new OrderItemDto(i.ProductId, i.Quantity, i.Price)).ToList(),order.TotalPrice,DateTime.UtcNow));await transaction.CommitAsync();// 3. 返回成功，但此时库存可能还没扣减return CreateOrderResult.Success(order.Id);}catch (Exception ex){await transaction.RollbackAsync();return CreateOrderResult.Failure(ex.Message);}}
}// 库存服务监听订单创建事件
public class OrderCreatedEventHandler : IEventHandler<OrderCreatedEvent>
{public async Task HandleAsync(OrderCreatedEvent @event){try{// 1. 检查库存foreach (var item in @event.Items){var stockAvailable = await _inventoryRepository.CheckStockAsync(item.ProductId);if (stockAvailable < item.Quantity){// 库存不足，发布补偿事件await _eventBus.PublishAsync(new StockInsufficientEvent(@event.OrderId,item.ProductId,item.Quantity,stockAvailable));return;}}// 2. 预留库存var reservedItems = new Dictionary<Guid, int>();foreach (var item in @event.Items){await _inventoryService.ReserveStockAsync(item.ProductId, item.Quantity);reservedItems[item.ProductId] = item.Quantity;}// 3. 发布库存预留成功事件await _eventBus.PublishAsync(new StockReservedEvent(@event.OrderId,reservedItems,DateTime.UtcNow));}catch (Exception ex){_logger.LogError(ex, "Error handling OrderCreatedEvent for order {OrderId}", @event.OrderId);throw; // 重新抛出，让消息队列重试}}
}

9.2 Saga模式：分布式事务的实用方案

Saga模式是目前最实用的分布式事务解决方案。它将长事务分解为一系列本地事务，通过事件串联起来。

9.2.1 编排式Saga vs 协同式Saga

编排式Saga（Orchestration）

// 编排器 - 中央协调者
public class OrderProcessingSagaOrchestrator
{private readonly IOrderRepository _orderRepository;private readonly IInventoryServiceClient _inventoryService;private readonly IPaymentServiceClient _paymentService;private readonly IEventBus _eventBus;private readonly ILogger<OrderProcessingSagaOrchestrator> _logger;public async Task ProcessOrder(Guid orderId){var sagaState = await _sagaStateRepository.GetOrCreateAsync(orderId);try{switch (sagaState.CurrentStep){case OrderProcessingStep.OrderCreated:// 步骤1：预留库存var order = await _orderRepository.GetByIdAsync(orderId);var stockReserved = await _inventoryService.ReserveStockAsync(order.Items.Select(i => new StockReservationRequest(i.ProductId, i.Quantity)).ToList());if (!stockReserved){await CancelOrder(order, "Insufficient stock");return;}sagaState.MoveToStep(OrderProcessingStep.StockReserved);await _sagaStateRepository.SaveAsync(sagaState);// 继续下一步goto case OrderProcessingStep.StockReserved;case OrderProcessingStep.StockReserved:// 步骤2：处理支付var paymentResult = await _paymentService.ProcessPaymentAsync(orderId);if (!paymentResult.Success){await CompensateStockReservation(orderId);await CancelOrder(order, "Payment failed");return;}sagaState.MoveToStep(OrderProcessingStep.PaymentCompleted);await _sagaStateRepository.SaveAsync(sagaState);// 继续下一步goto case OrderProcessingStep.PaymentCompleted;case OrderProcessingStep.PaymentCompleted:// 步骤3：确认订单var orderToConfirm = await _orderRepository.GetByIdAsync(orderId);orderToConfirm.Confirm();await _orderRepository.UpdateAsync(orderToConfirm);sagaState.MoveToStep(OrderProcessingStep.Completed);await _sagaStateRepository.SaveAsync(sagaState);_logger.LogInformation("Order processing saga completed for order {OrderId}", orderId);break;}}catch (Exception ex){_logger.LogError(ex, "Error in order processing saga for order {OrderId}", orderId);await HandleSagaFailure(sagaState);}}private async Task CompensateStockReservation(Guid orderId){// 释放已预留的库存await _inventoryService.ReleaseStockAsync(orderId);_logger.LogInformation("Stock reservation compensated for order {OrderId}", orderId);}
}

协同式Saga（Choreography）

// 每个服务监听事件并决定下一步动作
public class PaymentService
{public async Task HandleStockReserved(StockReservedEvent @event){try{// 处理支付var paymentResult = await ProcessPayment(@event.OrderId, @event.TotalAmount);if (paymentResult.Success){// 支付成功，发布事件await _eventBus.PublishAsync(new PaymentCompletedEvent(@event.OrderId,paymentResult.TransactionId,DateTime.UtcNow));}else{// 支付失败，发布补偿事件await _eventBus.PublishAsync(new PaymentFailedEvent(@event.OrderId,paymentResult.ErrorMessage,DateTime.UtcNow));}}catch (Exception ex){_logger.LogError(ex, "Error processing payment for order {OrderId}", @event.OrderId);throw;}}
}public class OrderService
{public async Task HandlePaymentCompleted(PaymentCompletedEvent @event){try{// 确认订单var order = await _orderRepository.GetByIdAsync(@event.OrderId);order.Confirm();await _orderRepository.UpdateAsync(order);// 发布订单确认事件await _eventBus.PublishAsync(new OrderConfirmedEvent(@event.OrderId,@event.TransactionId,DateTime.UtcNow));}catch (Exception ex){_logger.LogError(ex, "Error confirming order {OrderId}", @event.OrderId);throw;}}public async Task HandlePaymentFailed(PaymentFailedEvent @event){try{// 取消订单var order = await _orderRepository.GetByIdAsync(@event.OrderId);order.Cancel(@event.ErrorMessage);await _orderRepository.UpdateAsync(order);_logger.LogWarning("Order {OrderId} cancelled due to payment failure: {Error}", @event.OrderId, @event.ErrorMessage);}catch (Exception ex){_logger.LogError(ex, "Error cancelling order {OrderId}", @event.OrderId);throw;}}
}

9.2.2 补偿事务的设计

补偿事务是Saga模式的核心，它需要仔细设计。

// 补偿事务接口
public interface ICompensatableAction
{Task ExecuteAsync();Task CompensateAsync();bool CanCompensate { get; }
}// 库存预留的补偿操作
public class ReserveStockAction : ICompensatableAction
{private readonly IInventoryService _inventoryService;private readonly Guid _orderId;private readonly Dictionary<Guid, int> _reservedItems;private bool _executed = false;public async Task ExecuteAsync(){foreach (var item in _reservedItems){await _inventoryService.ReserveStockAsync(item.Key, item.Value);}_executed = true;}public async Task CompensateAsync(){if (!_executed) return;foreach (var item in _reservedItems){await _inventoryService.ReleaseStockAsync(_orderId, item.Key, item.Value);}}public bool CanCompensate => _executed;
}// Saga步骤管理
public class SagaStep
{public string StepName { get; }public ICompensatableAction Action { get; }public bool IsCompleted { get; private set; }public bool IsCompensated { get; private set; }public async Task ExecuteAsync(){await Action.ExecuteAsync();IsCompleted = true;}public async Task CompensateAsync(){if (Action.CanCompensate && IsCompleted && !IsCompensated){await Action.CompensateAsync();IsCompensated = true;}}
}// Saga管理器
public class SagaManager
{private readonly List<SagaStep> _steps = new();private readonly ILogger<SagaManager> _logger;public void AddStep(string stepName, ICompensatableAction action){_steps.Add(new SagaStep(stepName, action));}public async Task<SagaExecutionResult> ExecuteAsync(){var completedSteps = new List<SagaStep>();try{foreach (var step in _steps){await step.ExecuteAsync();completedSteps.Add(step);_logger.LogInformation("Saga step completed: {StepName}", step.StepName);}return SagaExecutionResult.Success();}catch (Exception ex){_logger.LogError(ex, "Saga execution failed at step: {StepName}", completedSteps.LastOrDefault()?.StepName ?? "Unknown");// 补偿已完成的步骤（倒序）for (int i = completedSteps.Count - 1; i >= 0; i--){try{await completedSteps[i].CompensateAsync();_logger.LogInformation("Saga step compensated: {StepName}", completedSteps[i].StepName);}catch (Exception compensationEx){_logger.LogError(compensationEx, "Failed to compensate step: {StepName}", completedSteps[i].StepName);}}return SagaExecutionResult.Failure(ex.Message);}}
}

9.3 幂等性：分布式系统的守护神

在分布式系统中，幂等性是保证数据一致性的关键。

9.3.1 幂等性设计模式

// 幂等操作接口
public interface IdempotentOperation
{Guid OperationId { get; }DateTime CreatedAt { get; }
}// 幂等性处理器
public class IdempotencyHandler<T> where T : IdempotentOperation
{private readonly IDistributedCache _cache;private readonly ILogger<IdempotencyHandler<T>> _logger;public async Task<TResult> ExecuteAsync<TResult>(T operation, Func<Task<TResult>> operationFunc) where TResult : class{var cacheKey = $"idempotency:{typeof(T).Name}:{operation.OperationId}";// 检查是否已经执行过var cachedResult = await _cache.GetStringAsync(cacheKey);if (cachedResult != null){_logger.LogInformation("Operation {OperationId} already executed, returning cached result", operation.OperationId);return JsonSerializer.Deserialize<TResult>(cachedResult);}// 执行操作var result = await operationFunc();// 缓存结果var serializedResult = JsonSerializer.Serialize(result);await _cache.SetStringAsync(cacheKey, serializedResult, TimeSpan.FromHours(24));return result;}
}// 在API中使用幂等性
[ApiController]
public class OrdersController : ControllerBase
{private readonly IdempotencyHandler<CreateOrderOperation> _idempotencyHandler;[HttpPost]public async Task<ActionResult<OrderDto>> CreateOrder([FromHeader(Name = "Idempotency-Key")] string idempotencyKey,[FromBody] CreateOrderRequest request){if (string.IsNullOrEmpty(idempotencyKey)){return BadRequest("Idempotency-Key header is required");}var operation = new CreateOrderOperation{OperationId = Guid.Parse(idempotencyKey),Request = request,CreatedAt = DateTime.UtcNow};var result = await _idempotencyHandler.ExecuteAsync(operation,async () => await _orderService.CreateOrderAsync(request));return Ok(result);}
}

9.3.2 消息幂等性

// 消息去重处理器
public class MessageDeduplicationHandler
{private readonly IDistributedCache _cache;private readonly ILogger<MessageDeduplicationHandler> _logger;public async Task<bool> ProcessMessageAsync<T>(string messageId, Func<Task> messageHandler) where T : class{var cacheKey = $"message:{typeof(T).Name}:{messageId}";// 使用分布式锁确保并发安全var lockKey = $"lock:{cacheKey}";var lockId = Guid.NewGuid().ToString();try{// 尝试获取锁var lockAcquired = await _cache.SetStringAsync(lockKey, lockId, TimeSpan.FromSeconds(30), When.NotExists);if (!lockAcquired){_logger.LogWarning("Message {MessageId} is being processed by another instance", messageId);return false;}// 检查是否已经处理过var processed = await _cache.GetStringAsync(cacheKey);if (processed != null){_logger.LogInformation("Message {MessageId} already processed", messageId);return false;}// 处理消息await messageHandler();// 标记为已处理await _cache.SetStringAsync(cacheKey, "processed", TimeSpan.FromDays(7));return true;}finally{// 释放锁var currentLockId = await _cache.GetStringAsync(lockKey);if (currentLockId == lockId){await _cache.RemoveAsync(lockKey);}}}
}// 使用消息去重的事件处理器
public class IdempotentOrderCreatedHandler : IEventHandler<OrderCreatedEvent>
{private readonly MessageDeduplicationHandler _deduplicationHandler;private readonly IOrderService _orderService;public async Task HandleAsync(OrderCreatedEvent @event){var processed = await _deduplicationHandler.ProcessMessageAsync<OrderCreatedEvent>(@event.Id.ToString(),async () =>{// 实际的消息处理逻辑await _orderService.ProcessNewOrderAsync(@event);});if (!processed){_logger.LogInformation("OrderCreatedEvent {EventId} was not processed (duplicate)", @event.Id);}}
}

9.4 监控和追踪

在分布式系统中，监控和追踪是保证数据一致性的重要工具。

9.4.1 分布式追踪

// 使用OpenTelemetry进行分布式追踪
public class DistributedTracingConfiguration
{public static void ConfigureTracing(IServiceCollection services, IConfiguration configuration){services.AddOpenTelemetry().WithTracing(builder =>{builder.AddAspNetCoreInstrumentation().AddHttpClientInstrumentation().AddEntityFrameworkCoreInstrumentation().AddSource("OrderService", "PaymentService", "InventoryService").SetResourceBuilder(ResourceBuilder.CreateDefault().AddService("ecommerce-system")).AddJaegerExporter(options =>{options.AgentHost = configuration["Jaeger:AgentHost"];options.AgentPort = int.Parse(configuration["Jaeger:AgentPort"]);});});}
}// 在业务代码中添加追踪
public class OrderService
{private readonly ActivitySource _activitySource;public async Task<Order> CreateOrderAsync(CreateOrderRequest request){using var activity = _activitySource.StartActivity("CreateOrder");activity?.SetTag("order.customer_id", request.CustomerId);activity?.SetTag("order.item_count", request.Items.Count);try{var order = Order.Create(request.CustomerId, request.Items);activity?.SetTag("order.id", order.Id);activity?.SetTag("order.total_amount", order.TotalPrice.Amount);await _orderRepository.AddAsync(order);return order;}catch (Exception ex){activity?.SetStatus(ActivityStatusCode.Error, ex.Message);activity?.RecordException(ex);throw;}}
}

9.4.2 一致性监控

// 一致性检查服务
public class ConsistencyMonitor
{private readonly IServiceProvider _serviceProvider;private readonly ILogger<ConsistencyMonitor> _logger;public async Task CheckOrderConsistencyAsync(){// 检查订单状态与支付状态的一致性var inconsistentOrders = await _orderRepository.FindInconsistentOrdersAsync();foreach (var order in inconsistentOrders){_logger.LogWarning("Inconsistent order found: {OrderId}, Status: {Status}, PaymentStatus: {PaymentStatus}",order.Id, order.Status, order.PaymentStatus);// 发送告警或自动修复await _alertingService.SendAlertAsync(new ConsistencyAlert{OrderId = order.Id,Issue = "Order status inconsistent with payment status",Severity = AlertSeverity.High});}}public async Task CheckInventoryConsistencyAsync(){// 检查库存数据的一致性var inconsistentProducts = await _inventoryRepository.FindInconsistentStockAsync();foreach (var product in inconsistentProducts){_logger.LogWarning("Inconsistent inventory found: {ProductId}, Available: {Available}, Total: {Total}",product.Id, product.AvailableQuantity, product.TotalQuantity);}}
}