分布式系统幂等性设计与.NET实战

## 1. 异步请求中的幂等性危机：你可能正在重复造轮子 上周排查一个线上订单重复支付问题时，发现某个.NET服务接口在10秒内被同一个客户端连续调用了7次。更可怕的是——由于缺乏幂等性处理，这7次调用全部执行成功，直接导致用户被重复扣款。这种场景在分布式系统中几乎每天都在上演，但90%的.NET开发者对幂等性的认知还停留在"加个数据库唯一索引"的层面。 幂等性的本质是"无论调用多少次，结果都等同于第一次调用"。在同步请求中我们可能用个锁就能解决，但面对以下异步场景时传统方案全部失效： - 消息队列的消费者重试 - 前端按钮重复提交 - 微服务超时重试 - 分布式事务补偿机制 ## 2. 幂等性设计的三个死亡陷阱 ### 2.1 陷阱一：用SessionID作为幂等键 我见过最典型的错误代码是这样的： ```csharp // 错误示范：用SessionID防重复提交 public ActionResult SubmitOrder(OrderRequest request) { var key = $"order_{Session.SessionID}_{request.ProductId}"; if (Cache.Exists(key)) return Json("请勿重复提交"); Cache.Set(key, 1, TimeSpan.FromMinutes(5)); // 业务处理... }

这种方案有三大致命伤：

1. 用户刷新页面会生成新SessionID
2. 集群环境下缓存可能不同步
3. 无法防御不同设备的并发请求

2.2 陷阱二：依赖数据库事务隔离级别

很多团队认为"把操作放到事务里就安全了"，但不同隔离级别的表现：

即使使用最高级别的Serializable，也无法解决以下场景：

• 跨库操作
• 非数据库操作（如调用外部API）
• 分布式环境下的时钟漂移

2.3 陷阱三：没有考虑业务补偿

某电商平台的库存扣减逻辑：

public bool DeductStock(int productId, int quantity) { var stock = GetStockFromDB(productId); if (stock >= quantity) { UpdateStock(productId, stock - quantity); return true; } return false; }

当这个接口被重复调用时，会出现：

1. 第一次调用：库存100→80（成功）
2. 第二次调用：库存80→60（本应拒绝却继续执行）

3. 工业级幂等方案实战

3.1 请求指纹技术：生成全局唯一业务ID

推荐采用客户端生成的SnowflakeID：

// 使用Zookeeper协调WorkerID public class SnowflakeGenerator { private const int TIMESTAMP_BITS = 41; private const int WORKER_BITS = 10; private const int SEQUENCE_BITS = 12; private long _lastTimestamp = -1L; private long _sequence = 0L; public long NextId() { lock (this) { var timestamp = DateTimeOffset.UtcNow.ToUnixTimeMilliseconds(); if (timestamp < _lastTimestamp) { throw new Exception("时钟回拨异常"); } if (_lastTimestamp == timestamp) { _sequence = (_sequence + 1) & ((1 << SEQUENCE_BITS) - 1); if (_sequence == 0) { timestamp = WaitNextMillis(_lastTimestamp); } } else { _sequence = 0; } _lastTimestamp = timestamp; return (timestamp << (WORKER_BITS + SEQUENCE_BITS)) | (WorkerId << SEQUENCE_BITS) | _sequence; } } }

在ASP.NET Core中通过过滤器实现：

public class IdempotencyAttribute : Attribute, IActionFilter { public void OnActionExecuting(ActionExecutingContext context) { var request = context.HttpContext.Request; var idempotencyKey = request.Headers["X-Idempotency-Key"]; if (string.IsNullOrEmpty(idempotencyKey)) { context.Result = new BadRequestObjectResult("缺少幂等键"); return; } var cache = context.HttpContext.RequestServices.GetService<IDistributedCache>(); if (cache.GetString($"idempotency:{idempotencyKey}") != null) { context.Result = new ConflictObjectResult("重复请求"); return; } cache.SetString($"idempotency:{idempotencyKey}", "1", new DistributedCacheEntryOptions { AbsoluteExpirationRelativeToNow = TimeSpan.FromDays(1) }); } }

3.2 状态机模式：让业务操作具备自检能力

以订单状态流转为例：

stateDiagram-v2 [*] --> Pending Pending --> Processing : 开始处理 Processing --> Completed : 处理成功 Processing --> Failed : 处理失败 Failed --> Processing : 重试 Completed --> [*]

对应的C#实现：

public class OrderService { public async Task<Result> ProcessOrderAsync(string orderId) { using var transaction = await _dbContext.Database.BeginTransactionAsync(); try { var order = await _dbContext.Orders .Where(o => o.Id == orderId) .FirstOrDefaultAsync(); if (order == null) return Result.Fail("订单不存在"); switch (order.Status) { case OrderStatus.Completed: return Result.Ok("订单已完成"); case OrderStatus.Failed: return Result.Fail("订单已失败"); case OrderStatus.Processing: var lockKey = $"order_lock:{orderId}"; if (await _distributedLock.AcquireAsync(lockKey, TimeSpan.FromSeconds(30))) { try { // 检查是否已被其他实例处理完成 var currentStatus = await _dbContext.Orders .Where(o => o.Id == orderId) .Select(o => o.Status) .FirstOrDefaultAsync(); if (currentStatus == OrderStatus.Completed) { return Result.Ok("订单已完成"); } // 继续处理... } finally { await _distributedLock.ReleaseAsync(lockKey); } } break; default: order.Status = OrderStatus.Processing; await _dbContext.SaveChangesAsync(); // 业务处理... order.Status = OrderStatus.Completed; await _dbContext.SaveChangesAsync(); await transaction.CommitAsync(); return Result.Ok(); } } catch { await transaction.RollbackAsync(); throw; } } }

3.3 补偿事务设计：最终一致性的保障

对于支付场景的补偿方案：

public class PaymentService { private readonly IMessageBus _messageBus; public async Task<PaymentResult> ProcessPaymentAsync(PaymentRequest request) { var paymentId = GenerateSnowflakeId(); var payment = new Payment { Id = paymentId, Status = PaymentStatus.Pending, Amount = request.Amount, CreatedAt = DateTime.UtcNow }; await _dbContext.Payments.AddAsync(payment); await _dbContext.SaveChangesAsync(); try { // 调用银行接口 var bankResponse = await _bankClient.ProcessAsync(new BankRequest { PaymentId = paymentId, Amount = request.Amount, CardNumber = request.CardNumber }); payment.Status = PaymentStatus.Completed; payment.TransactionId = bankResponse.TransactionId; await _dbContext.SaveChangesAsync(); // 发送领域事件 await _messageBus.PublishAsync(new PaymentCompletedEvent { PaymentId = paymentId, UserId = request.UserId }); return PaymentResult.Success(paymentId); } catch (Exception ex) { payment.Status = PaymentStatus.Failed; payment.ErrorMessage = ex.Message; await _dbContext.SaveChangesAsync(); // 启动补偿流程 _ = Task.Run(async () => { await Task.Delay(TimeSpan.FromMinutes(5)); await CompensatePaymentAsync(paymentId); }); return PaymentResult.Failed("支付处理中"); } } private async Task CompensatePaymentAsync(string paymentId) { var payment = await _dbContext.Payments .FirstOrDefaultAsync(p => p.Id == paymentId); if (payment?.Status == PaymentStatus.Failed) { // 检查银行实际状态 var status = await _bankClient.QueryStatusAsync(paymentId); if (status == BankTransactionStatus.Completed) { payment.Status = PaymentStatus.Completed; await _dbContext.SaveChangesAsync(); await _messageBus.PublishAsync(new PaymentCompletedEvent { PaymentId = paymentId }); } } } }

4. 高并发场景下的优化技巧

4.1 分布式锁的正确打开方式

常见锁方案对比：

推荐使用RedLock算法的改进版：

public class RedLock : IAsyncDisposable { private readonly List<IDistributedLock> _locks; private readonly TimeSpan _expiry; public static async Task<RedLock> CreateAsync( string resource, TimeSpan expiry, params IDistributedLockProvider[] providers) { var locks = new List<IDistributedLock>(); try { foreach (var provider in providers) { var @lock = await provider.AcquireAsync(resource, expiry); if (@lock != null) { locks.Add(@lock); if (locks.Count >= providers.Length / 2 + 1) { return new RedLock(locks, expiry); } } } throw new Exception("获取锁失败"); } catch { foreach (var @lock in locks) { await @lock.DisposeAsync(); } throw; } } public ValueTask DisposeAsync() { return new ValueTask(Task.WhenAll( _locks.Select(lock => lock.DisposeAsync().AsTask()))); } }

4.2 幂等存储的索引优化

针对SQL Server的幂等表设计：

CREATE TABLE IdempotencyRecords ( IdempotencyKey NVARCHAR(128) NOT NULL, ResourceType NVARCHAR(64) NOT NULL, CreatedAt DATETIME2 NOT NULL DEFAULT SYSUTCDATETIME(), ResponseBody VARBINARY(MAX), ResponseCode INT, CONSTRAINT PK_Idempotency PRIMARY KEY NONCLUSTERED (IdempotencyKey), INDEX IX_ResourceType_CreatedAt CLUSTERED (ResourceType, CreatedAt) ) WITH (MEMORY_OPTIMIZED = ON, DURABILITY = SCHEMA_ONLY);

内存优化表配合以下策略：

1. 每日定时归档旧记录
2. 使用Temporal Table跟踪历史
3. 对热点资源启用分区表

5. 真实案例：电商秒杀系统的幂等改造

某平台在618大促期间出现的典型问题：

1. 00:00:00 用户点击"立即购买"
2. 00:00:01 前端收到504超时（实际后端正在处理）
3. 00:00:02 用户再次点击
4. 00:00:05 两个请求同时完成，创建了重复订单

改造后的架构：

客户端 → API网关 → 幂等校验 → 消息队列 → 订单服务 → 数据库 ↑ ↓ Redis缓存 补偿Worker

关键代码实现：

// 网关层的幂等拦截器 public class IdempotencyInterceptor : DelegatingHandler { protected override async Task<HttpResponseMessage> SendAsync( HttpRequestMessage request, CancellationToken cancellationToken) { if (request.Method == HttpMethod.Post) { var idempotencyKey = request.Headers.GetValues("X-Idempotency-Key").FirstOrDefault(); if (!string.IsNullOrEmpty(idempotencyKey)) { var cache = request.GetService<IDistributedCache>(); var cachedResponse = await cache.GetAsync($"idempotency:{idempotencyKey}"); if (cachedResponse != null) { return JsonConvert.DeserializeObject<HttpResponseMessage>( Encoding.UTF8.GetString(cachedResponse)); } var response = await base.SendAsync(request, cancellationToken); if (response.IsSuccessStatusCode) { await cache.SetAsync( $"idempotency:{idempotencyKey}", Encoding.UTF8.GetBytes(JsonConvert.SerializeObject(response)), new DistributedCacheEntryOptions { AbsoluteExpirationRelativeToNow = TimeSpan.FromDays(1) }); } return response; } } return await base.SendAsync(request, cancellationToken); } }

性能优化前后的对比数据：

这个案例告诉我们：牺牲5%的性能换取99.9%的数据一致性是完全值得的。在实际编码中，我建议用Polly库实现自动重试策略：

var retryPolicy = Policy<HttpResponseMessage> .HandleResult(r => r.StatusCode == HttpStatusCode.Conflict) .RetryAsync(3, (result, retryCount) => { _logger.LogWarning($"检测到冲突，第{retryCount}次重试"); });

最后分享一个排查幂等性问题的小技巧：在NLog配置中加入TraceId：

<target name="jsonFile" xsi:type="File" fileName="logs/app.json"> <layout xsi:type="JsonLayout"> <attribute name="time" layout="${date:format=yyyy-MM-ddTHH\:mm\:ss.fff}" /> <attribute name="level" layout="${level:upperCase=true}"/> <attribute name="traceId" layout="${activityid}"/> <attribute name="operation" layout="${mdlc:item=operation}"/> <attribute name="message" layout="${message}" /> </layout> </target>

这样在ELK中可以通过trace_id快速串联整个调用链的日志。