分布式锁与一致性

一、分布式锁概述

分布式锁是在分布式系统中实现互斥访问共享资源的机制。在分布式环境下，多个进程或服务需要协调对共享资源的访问，分布式锁能够保证同一时间只有一个进程访问该资源。

二、分布式锁特性

2.1 分布式锁核心特性

三、Redis分布式锁

3.1 Redis锁原理

基于Redis的SET命令实现分布式锁：

3.2 Redis锁实现

public class RedisDistributedLock : IDistributedLock
{
    private readonly IDistributedCache _cache;
    private readonly string _lockKey;
    private readonly string _lockValue;
    private readonly TimeSpan _expiry;
    
    public RedisDistributedLock(IDistributedCache cache, string lockKey, TimeSpan expiry)
    {
        _cache = cache;
        _lockKey = lockKey;
        _lockValue = Guid.NewGuid().ToString();
        _expiry = expiry;
    }
    
    public async Task<bool> AcquireAsync()
    {
        var result = await _cache.StringSetAsync(
            _lockKey,
            _lockValue,
            _expiry,
            When.NotExists
        );
        return result;
    }
    
    public async Task ReleaseAsync()
    {
        var script = @"
            if redis.call('GET', KEYS[1]) == ARGV[1] then
                return redis.call('DEL', KEYS[1])
            else
                return 0
            end
        ";
        
        await _cache.ScriptEvaluateAsync(script, 
            new RedisKey[] { _lockKey }, 
            new RedisValue[] { _lockValue });
    }
    
    public async Task<bool> TryAcquireAsync(TimeSpan timeout)
    {
        var startTime = DateTime.Now;
        
        while (DateTime.Now - startTime < timeout)
        {
            if (await AcquireAsync())
                return true;
            
            await Task.Delay(100);
        }
        
        return false;
    }
}

3.3 Redis Redlock算法

public class RedlockDistributedLock : IDistributedLock
{
    private readonly List<IDistributedCache> _redisInstances;
    private readonly string _lockKey;
    private readonly string _lockValue;
    private readonly TimeSpan _expiry;
    private readonly int _quorum;
    
    public RedlockDistributedLock(List<IDistributedCache> redisInstances, 
        string lockKey, TimeSpan expiry)
    {
        _redisInstances = redisInstances;
        _lockKey = lockKey;
        _lockValue = Guid.NewGuid().ToString();
        _expiry = expiry;
        _quorum = (redisInstances.Count / 2) + 1;
    }
    
    public async Task<bool> AcquireAsync()
    {
        var tasks = _redisInstances.Select(async redis =>
        {
            return await redis.StringSetAsync(_lockKey, _lockValue, _expiry, When.NotExists);
        });
        
        var results = await Task.WhenAll(tasks);
        var successCount = results.Count(r => r);
        
        return successCount >= _quorum;
    }
    
    public async Task ReleaseAsync()
    {
        var tasks = _redisInstances.Select(async redis =>
        {
            var script = @"
                if redis.call('GET', KEYS[1]) == ARGV[1] then
                    return redis.call('DEL', KEYS[1])
                else
                    return 0
                end
            ";
            await redis.ScriptEvaluateAsync(script, 
                new RedisKey[] { _lockKey }, 
                new RedisValue[] { _lockValue });
        });
        
        await Task.WhenAll(tasks);
    }
}

四、ZooKeeper分布式锁

4.1 ZooKeeper锁原理

基于ZooKeeper的临时顺序节点实现分布式锁：

4.2 ZooKeeper锁实现

public class ZooKeeperDistributedLock : IDistributedLock
{
    private readonly IZooKeeperClient _zooKeeper;
    private readonly string _lockPath;
    private string _currentNodePath;
    private ManualResetEvent _waitEvent = new ManualResetEvent(false);
    
    public ZooKeeperDistributedLock(IZooKeeperClient zooKeeper, string lockPath)
    {
        _zooKeeper = zooKeeper;
        _lockPath = lockPath;
    }
    
    public async Task AcquireAsync()
    {
        await EnsureLockPathExists();
        
        _currentNodePath = await _zooKeeper.CreateAsync(
            $"{_lockPath}/lock-",
            data: null,
            CreateMode.EphemeralSequential
        );
        
        var children = await _zooKeeper.GetChildrenAsync(_lockPath);
        children.Sort();
        
        var currentIndex = children.IndexOf(Path.GetFileName(_currentNodePath));
        
        if (currentIndex == 0)
            return;
        
        var previousNode = children[currentIndex - 1];
        var previousNodePath = $"{_lockPath}/{previousNode}";
        
        await _zooKeeper.SubscribeDataChange(previousNodePath, (s, e) =>
        {
            if (e.ChangeType == Watcher.Event.EventType.NodeDeleted)
            {
                _waitEvent.Set();
            }
        });
        
        _waitEvent.WaitOne();
    }
    
    public async Task ReleaseAsync()
    {
        if (!string.IsNullOrEmpty(_currentNodePath))
        {
            await _zooKeeper.DeleteAsync(_currentNodePath);
            _currentNodePath = null;
        }
    }
    
    private async Task EnsureLockPathExists()
    {
        if (!await _zooKeeper.ExistsAsync(_lockPath))
        {
            await _zooKeeper.CreateAsync(_lockPath, data: null, CreateMode.Persistent);
        }
    }
}

五、etcd分布式锁

5.1 etcd锁原理

基于etcd的KV存储和watch机制实现分布式锁：

5.2 etcd锁实现

public class EtcdDistributedLock : IDistributedLock
{
    private readonly IEtcdClient _etcdClient;
    private readonly string _lockKey;
    private readonly string _lockValue;
    private long _leaseId;
    
    public EtcdDistributedLock(IEtcdClient etcdClient, string lockKey)
    {
        _etcdClient = etcdClient;
        _lockKey = lockKey;
        _lockValue = Guid.NewGuid().ToString();
    }
    
    public async Task<bool> AcquireAsync(TimeSpan expiry)
    {
        var leaseResponse = await _etcdClient.LeaseGrantAsync(expiry);
        _leaseId = leaseResponse.ID;
        
        var putResponse = await _etcdClient.PutAsync(_lockKey, _lockValue, 
            new PutOptions { Lease = _leaseId, PrevExist = false });
        
        if (!putResponse.Succeeded)
        {
            await _etcdClient.LeaseRevokeAsync(_leaseId);
            return false;
        }
        
        return true;
    }
    
    public async Task ReleaseAsync()
    {
        if (_leaseId > 0)
        {
            await _etcdClient.LeaseRevokeAsync(_leaseId);
            _leaseId = 0;
        }
    }
    
    public async Task<bool> TryAcquireAsync(TimeSpan expiry, TimeSpan timeout)
    {
        var startTime = DateTime.Now;
        
        while (DateTime.Now - startTime < timeout)
        {
            if (await AcquireAsync(expiry))
                return true;
            
            await WaitForLockRelease();
        }
        
        return false;
    }
    
    private async Task WaitForLockRelease()
    {
        using var watcher = _etcdClient.Watch(_lockKey);
        
        await foreach (var response in watcher)
        {
            foreach (var @event in response.Events)
            {
                if (@event.Type == Mvccpb.Event.Types.EventType.Delete)
                    return;
            }
        }
    }
}

六、分布式锁对比

6.1 分布式锁方案对比

七、分布式锁与数据一致性

7.1 分布式锁保障一致性

7.2 锁粒度选择

public class LockGranularityService
{
    public IDistributedLock GetLock(LockType type, string resourceId)
    {
        return type switch
        {
            LockType.Global => new RedisDistributedLock(_cache, "global-lock", TimeSpan.FromSeconds(30)),
            LockType.Product => new RedisDistributedLock(_cache, $"product:{resourceId}", TimeSpan.FromSeconds(10)),
            LockType.Order => new RedisDistributedLock(_cache, $"order:{resourceId}", TimeSpan.FromSeconds(15)),
            LockType.User => new RedisDistributedLock(_cache, $"user:{resourceId}", TimeSpan.FromSeconds(20)),
            _ => throw new ArgumentOutOfRangeException(nameof(type))
        };
    }
}

public enum LockType
{
    Global,
    Product,
    Order,
    User
}

7.3 锁与事务结合

public async Task ProcessOrderAsync(int orderId)
{
    var lockKey = $"order:{orderId}";
    
    using var distributedLock = new RedisDistributedLock(_cache, lockKey, TimeSpan.FromSeconds(30));
    
    if (!await distributedLock.TryAcquireAsync(TimeSpan.FromSeconds(10)))
    {
        throw new TimeoutException("获取锁超时");
    }
    
    try
    {
        using var transaction = await _dbContext.Database.BeginTransactionAsync();
        
        var order = await _dbContext.Orders.FindAsync(orderId);
        if (order == null)
            throw new NotFoundException("订单不存在");
        
        order.Status = OrderStatus.Processing;
        await _dbContext.SaveChangesAsync();
        
        await _cache.SetStringAsync($"order:{orderId}", JsonSerializer.Serialize(order));
        
        await transaction.CommitAsync();
    }
    finally
    {
        await distributedLock.ReleaseAsync();
    }
}

八、分布式锁最佳实践

8.1 设置合理的超时时间

超时时间应大于业务执行时间，避免锁提前释放。

8.2 使用唯一标识释放锁

使用唯一标识防止误释放其他客户端的锁。

8.3 锁的续期机制

public class RedisDistributedLockWithRenewal : RedisDistributedLock
{
    private Timer _renewalTimer;
    private readonly TimeSpan _renewalInterval;
    
    public RedisDistributedLockWithRenewal(IDistributedCache cache, 
        string lockKey, TimeSpan expiry) : base(cache, lockKey, expiry)
    {
        _renewalInterval = expiry / 3;
    }
    
    public override async Task<bool> AcquireAsync()
    {
        var acquired = await base.AcquireAsync();
        
        if (acquired)
        {
            _renewalTimer = new Timer(async _ =>
            {
                await RenewLockAsync();
            }, null, _renewalInterval, _renewalInterval);
        }
        
        return acquired;
    }
    
    private async Task RenewLockAsync()
    {
        var script = @"
            if redis.call('GET', KEYS[1]) == ARGV[1] then
                return redis.call('PEXPIRE', KEYS[1], ARGV[2])
            else
                return 0
            end
        ";
        
        await _cache.ScriptEvaluateAsync(script, 
            new RedisKey[] { _lockKey }, 
            new RedisValue[] { _lockValue, _expiry.TotalMilliseconds });
    }
    
    public override async Task ReleaseAsync()
    {
        _renewalTimer?.Dispose();
        await base.ReleaseAsync();
    }
}

8.4 锁的降级策略

public async Task<T> ExecuteWithLockAsync<T>(
    string lockKey, 
    Func<Task<T>> operation,
    Func<Task<T>> fallbackOperation)
{
    using var distributedLock = new RedisDistributedLock(_cache, lockKey, TimeSpan.FromSeconds(30));
    
    try
    {
        if (await distributedLock.TryAcquireAsync(TimeSpan.FromSeconds(5)))
        {
            return await operation();
        }
    }
    catch (Exception ex)
    {
        _logger.LogError(ex, "获取锁失败");
    }
    
    return await fallbackOperation();
}

8.5 锁的监控与告警

public class LockMonitor
{
    public async Task MonitorLockUsage()
    {
        var lockStats = await _cache.GetLockStatisticsAsync();
        
        foreach (var stat in lockStats)
        {
            if (stat.WaitTime > TimeSpan.FromSeconds(30))
            {
                await _alertService.SendAlert("锁等待时间过长", 
                    $"锁: {stat.LockKey}, 等待时间: {stat.WaitTime}");
            }
            
            if (stat.FailedAttempts > 100)
            {
                await _alertService.SendAlert("锁获取失败次数过多", 
                    $"锁: {stat.LockKey}, 失败次数: {stat.FailedAttempts}");
            }
        }
    }
}

九、分布式锁常见问题

9.1 死锁问题

通过超时释放机制避免死锁。

9.2 锁提前释放

通过锁续期机制解决。

9.3 锁竞争激烈

通过细粒度锁和排队机制缓解。

9.4 锁服务故障

通过多节点部署和降级策略应对。

十、总结

分布式锁是分布式系统中保证数据一致性的关键机制。Redis锁性能高适合高频场景，ZooKeeper锁一致性强适合严格场景，etcd锁适合K8s生态。合理选择锁方案、设置超时时间、实现锁续期、监控锁状态，能够构建可靠的分布式锁系统，保障数据一致性。