性能 - 数据库锁竞争

问题描述

在高并发场景下，幂等性实现中的数据库锁（唯一索引、行锁、表锁）可能成为性能瓶颈，导致：

1. 响应时间增加：请求等待锁释放
2. 吞吐量下降：并发能力受限
3. 死锁风险：多个事务互相等待
4. 超时错误：锁等待超时

典型场景

场景 1：高并发订单创建

// ❌ 性能问题：大量并发请求竞争同一个用户的订单锁
public async Task<Order> CreateOrder(Guid userId, OrderRequest request)
{
    var order = new Order
    {
        UserId = userId,
        OrderNumber = GenerateOrderNumber(),
        // ...
    };
    
    _dbContext.Orders.Add(order);
    await _dbContext.SaveChangesAsync(); // 可能在这里阻塞
    
    return order;
}

// 问题：
// - 同一用户的多个订单请求串行执行
// - 唯一索引检查导致行锁竞争
// - 高峰期响应时间从 50ms 增加到 2s+

场景 2：库存扣减

-- ❌ 性能问题：热门商品库存更新竞争激烈
UPDATE products 
SET stock = stock - 1 
WHERE id = 'popular-product-id' 
  AND stock > 0;

-- 问题：
-- - 同一商品的多个扣减请求串行
-- - 行锁竞争导致超时
-- - 可能产生死锁

解决方案

方案 1：使用 Redis 代替数据库锁（推荐）

public class OptimizedPaymentService
{
    private readonly IDatabase _redis;
    private readonly AppDbContext _dbContext;
    
    public async Task<Result<Payment>> CreatePaymentOptimized(
        CreatePaymentRequest request,
        string idempotencyKey)
    {
        // 1. 先在 Redis 中检查幂等性（快速）
        var cacheKey = $"payment:idempotency:{idempotencyKey}";
        var cached = await _redis.StringGetAsync(cacheKey);
        
        if (!cached.IsNullOrEmpty)
        {
            // 命中缓存，直接返回
            return Result<Payment>.Success(
                JsonSerializer.Deserialize<Payment>(cached!));
        }
        
        // 2. 尝试获取分布式锁
        var lockKey = $"payment:lock:{request.OrderId}";
        var lockValue = Guid.NewGuid().ToString();
        
        var acquired = await _redis.StringSetAsync(
            lockKey, 
            lockValue, 
            TimeSpan.FromSeconds(10),
            When.NotExists);
        
        if (!acquired)
        {
            return Result<Payment>.Failure("Request is being processed");
        }
        
        try
        {
            // 3. 再次检查（双重检查锁定）
            cached = await _redis.StringGetAsync(cacheKey);
            if (!cached.IsNullOrEmpty)
            {
                return Result<Payment>.Success(
                    JsonSerializer.Deserialize<Payment>(cached!));
            }
            
            // 4. 数据库操作
            var payment = await CreatePaymentInDatabase(request, idempotencyKey);
            
            // 5. 缓存结果
            await _redis.StringSetAsync(
                cacheKey, 
                JsonSerializer.Serialize(payment),
                TimeSpan.FromHours(24));
            
            return Result<Payment>.Success(payment);
        }
        finally
        {
            // 6. 释放锁（Lua 脚本保证原子性）
            var luaScript = @"
                if redis.call('get', KEYS[1]) == ARGV[1] then
                    return redis.call('del', KEYS[1])
                else
                    return 0
                end";
            
            await _redis.ScriptEvaluateAsync(luaScript, 
                new RedisKey[] { lockKey }, 
                new RedisValue[] { lockValue });
        }
    }
}

性能对比：

指标	数据库锁	Redis 锁	提升
平均响应时间	200ms	10ms	20x
P99 响应时间	2000ms	50ms	40x
并发 QPS	500	10000	20x
数据库连接数	高	低	-

方案 2：批量插入减少锁竞争

// ❌ 低效：逐条插入，每次都要获取锁
foreach (var item in items)
{
    var entity = new Entity { /* ... */ };
    _dbContext.Entities.Add(entity);
    await _dbContext.SaveChangesAsync(); // N 次数据库往返
}

// ✅ 高效：批量插入，一次性获取锁
var entities = items.Select(item => new Entity { /* ... */ }).ToList();
_dbContext.Entities.AddRange(entities);
await _dbContext.SaveChangesAsync(); // 1 次数据库往返

方案 3：异步队列削峰

public class QueueBasedPaymentService
{
    private readonly Channel<PaymentRequest> _paymentChannel;
    
    public QueueBasedPaymentService()
    {
        // 创建有界通道，限制队列大小
        _paymentChannel = Channel.CreateBounded<PaymentRequest>(new BoundedChannelOptions(10000)
        {
            FullMode = BoundedChannelFullMode.Wait
        });
    }
    
    public async Task<string> SubmitPayment(PaymentRequest request)
    {
        var idempotencyKey = Guid.NewGuid().ToString();
        
        // 快速入队，立即返回
        await _paymentChannel.Writer.WriteAsync(new PaymentRequest
        {
            IdempotencyKey = idempotencyKey,
            Data = request
        });
        
        return idempotencyKey; // 返回键供后续查询
    }
    
    // 后台消费者
    public async Task StartProcessing(CancellationToken cancellationToken)
    {
        await foreach (var request in _paymentChannel.Reader.ReadAllAsync(cancellationToken))
        {
            try
            {
                await ProcessPaymentAsync(request);
            }
            catch (Exception ex)
            {
                _logger.LogError(ex, "Failed to process payment");
            }
        }
    }
}

优势：

• 客户端快速响应（< 10ms）
• 后端按处理能力消费
• 天然限流，避免雪崩

方案 4：分区策略减少冲突

public class PartitionedLockService
{
    private readonly int _partitionCount = 16;
    private readonly SemaphoreSlim[] _partitions;
    
    public PartitionedLockService()
    {
        _partitions = Enumerable.Range(0, _partitionCount)
            .Select(_ => new SemaphoreSlim(1, 1))
            .ToArray();
    }
    
    public async Task<T> ExecuteWithLockAsync<T>(
        string key, 
        Func<Task<T>> action,
        TimeSpan? timeout = null)
    {
        var partitionIndex = Math.Abs(key.GetHashCode()) % _partitionCount;
        var partition = _partitions[partitionIndex];
        
        var acquired = await partition.WaitAsync(timeout ?? TimeSpan.FromSeconds(5));
        
        if (!acquired)
        {
            throw new TimeoutException($"Failed to acquire lock for key: {key}");
        }
        
        try
        {
            return await action();
        }
        finally
        {
            partition.Release();
        }
    }
}

// 使用
var result = await _lockService.ExecuteWithLockAsync(
    $"order:{userId}",
    async () => await CreateOrder(userId, request));

原理：

• 将锁分散到多个分区
• 不同用户可能在不同分区，减少竞争
• 适合热点数据场景

方案 5：优化数据库索引

-- ❌ 低效：没有合适的索引，全表扫描
SELECT * FROM payments WHERE idempotency_key = 'xxx';

-- ✅ 高效：添加唯一索引
CREATE UNIQUE INDEX idx_payments_idempotency_key 
ON payments(idempotency_key);

-- ❌ 低效：复合索引顺序不当
CREATE INDEX idx_orders_user_created ON orders(created_at, user_id);

-- ✅ 高效：高频查询字段在前
CREATE INDEX idx_orders_user_created ON orders(user_id, created_at);

索引优化建议：

1. 覆盖索引：包含查询所需的所有字段
2. 部分索引：只对活跃数据建立索引
3. BRIN 索引：适合时间序列数据

-- PostgreSQL BRIN 索引（适合时间序列）
CREATE INDEX idx_payments_created_at_brin 
ON payments USING brin(created_at);

-- 部分索引（只索引未完成的支付）
CREATE INDEX idx_payments_pending 
ON payments(status) 
WHERE status = 'pending';

监控和诊断

1. 检测锁竞争

-- PostgreSQL：查看当前锁等待
SELECT 
    blocked_locks.pid AS blocked_pid,
    blocked_activity.usename AS blocked_user,
    blocking_locks.pid AS blocking_pid,
    blocking_activity.usename AS blocking_user,
    blocked_activity.query AS blocked_statement,
    blocking_activity.query AS current_statement_in_blocking_process,
    now() - blocked_activity.query_start AS wait_duration
FROM pg_catalog.pg_locks blocked_locks
JOIN pg_catalog.pg_stat_activity blocked_activity ON blocked_activity.pid = blocked_locks.pid
JOIN pg_catalog.pg_locks blocking_locks 
    ON blocking_locks.locktype = blocked_locks.locktype
    AND blocking_locks.database IS NOT DISTINCT FROM blocked_locks.database
    AND blocking_locks.relation IS NOT DISTINCT FROM blocked_locks.relation
    AND blocking_locks.page IS NOT DISTINCT FROM blocked_locks.page
    AND blocking_locks.tuple IS NOT DISTINCT FROM blocked_locks.tuple
    AND blocking_locks.virtualxid IS NOT DISTINCT FROM blocked_locks.virtualxid
    AND blocking_locks.transactionid IS NOT DISTINCT FROM blocked_locks.transactionid
    AND blocking_locks.classid IS NOT DISTINCT FROM blocked_locks.classid
    AND blocking_locks.objid IS NOT DISTINCT FROM blocked_locks.objid
    AND blocking_locks.objsubid IS NOT DISTINCT FROM blocked_locks.objsubid
    AND blocking_locks.pid != blocked_locks.pid
JOIN pg_catalog.pg_stat_activity blocking_activity ON blocking_activity.pid = blocking_locks.pid
WHERE NOT blocked_locks.granted;

2. C# 性能监控

public class LockPerformanceMonitor
{
    private readonly Histogram<double> _lockWaitTime;
    private readonly Counter<long> _lockTimeouts;
    private readonly Gauge<int> _activeLocks;
    
    public async Task<T> MonitorLockAsync<T>(
        string lockName,
        Func<Task<T>> action)
    {
        var stopwatch = Stopwatch.StartNew();
        _activeLocks.Add(1);
        
        try
        {
            var result = await action();
            stopwatch.Stop();
            
            _lockWaitTime.Record(stopwatch.ElapsedMilliseconds);
            
            return result;
        }
        catch (TimeoutException)
        {
            _lockTimeouts.Add(1);
            throw;
        }
        finally
        {
            _activeLocks.Add(-1);
        }
    }
}

最佳实践总结

1. 设计原则

✅ 优先使用 Redis：比数据库锁快 10-100 倍
✅ 缩短临界区：锁内只做必要操作
✅ 设置超时：避免无限等待
✅ 异步处理：队列削峰填谷
✅ 分区策略：分散热点

2. 避免的陷阱

❌ 长事务：尽快提交事务
❌ 嵌套锁：容易导致死锁
❌ 全表扫描：确保有合适的索引
❌ 过度加锁：评估是否真的需要
❌ 忽略监控：及时发现性能问题

3. 性能基准

目标指标：
- 平均响应时间：< 50ms
- P99 响应时间：< 200ms
- 并发 QPS：> 5000
- 锁等待超时率：< 0.1%
- 死锁发生率：< 0.01%

总结

数据库锁竞争是幂等性实现中的常见性能瓶颈。通过以下方案可以有效优化：

1. Redis 分布式锁：替代数据库锁，提升性能
2. 批量操作：减少数据库往返
3. 异步队列：削峰填谷，平滑负载
4. 分区策略：分散热点，减少冲突
5. 索引优化：提升查询效率

关键是要根据业务场景选择合适的方案，并做好监控和告警。