介绍

在单机部署的应用中，可以通过 lock、Monitor、Mutex 等机制实现并发控制。但在分布式系统中，应用部署在多台机器的多个进程中，本地锁无法跨进程生效，需要一种跨机器的互斥机制来控制共享资源的访问，这就是分布式锁。

分布式锁需具备以下条件：

  1. 在分布式环境下，同一时间只能有一台机器的一个线程执行某个方法
  2. 高可用的获取锁与释放锁
  3. 高性能的获取锁与释放锁
  4. 具备可重入特性
  5. 具备锁失效机制，防止死锁
  6. 具备非阻塞锁特性，即没有获取到锁将直接返回失败

silky 框架的分布式锁

silky 框架使用 DistributedLock 实现分布式锁，并内置了基于 Redis 的分布式锁实现（Silky.DistributedLock.Redis 包，已包含在 Silky.Agent.Host 中）。

注意：silky 框架在服务注册过程中内部使用了分布式锁，防止多个服务实例并发注册同一服务条目时出现数据不一致的问题。因此，即使业务代码不使用分布式锁，也必须正确配置 Redis 服务。

Redis 配置（必须）

distributedCache:
  redis:
    isEnabled: true
    configuration: 127.0.0.1:6379,defaultDatabase=0,password=qwe!P4ss

在业务中使用分布式锁

silky 通过 IDistributedLockFactory 接口提供分布式锁能力，开发者通过构造注入即可使用。

基本用法

using Medallion.Threading;

public class OrderDomainService : IOrderDomainService, IScopedDependency
{
    private readonly IDistributedLockFactory _lockFactory;

    public OrderDomainService(IDistributedLockFactory lockFactory)
    {
        _lockFactory = lockFactory;
    }

    public async Task CreateOrderAsync(CreateOrderInput input)
    {
        // 以用户 ID 为粒度加锁，防止同一用户并发下单
        var lockKey = $"order:create:user:{input.UserId}";
        var @lock = _lockFactory.CreateLock(lockKey);

        await using (await @lock.AcquireAsync())
        {
            // 此处为加锁区域，同一时间只有一个实例执行
            var existingOrder = await CheckDuplicateOrderAsync(input);
            if (existingOrder != null)
            {
                throw new BusinessException("请勿重复下单");
            }

            await SaveOrderAsync(input);
        }
        // 锁在 using 块结束时自动释放
    }
}

非阻塞锁（TryAcquire）

public async Task<bool> TryProcessAsync(string resourceId)
{
    var @lock = _lockFactory.CreateLock($"resource:{resourceId}");

    // 尝试获取锁，如果获取失败立即返回 null（非阻塞）
    await using var handle = await @lock.TryAcquireAsync(timeout: TimeSpan.Zero);
    if (handle == null)
    {
        // 未获取到锁，直接返回
        return false;
    }

    // 获取到锁，执行业务逻辑
    await DoProcessAsync(resourceId);
    return true;
}

带超时的锁获取

public async Task ProcessWithTimeoutAsync(string resourceId)
{
    var @lock = _lockFactory.CreateLock($"resource:{resourceId}");

    // 最多等待 5 秒获取锁
    await using var handle = await @lock.TryAcquireAsync(timeout: TimeSpan.FromSeconds(5));
    if (handle == null)
    {
        throw new TimeoutException($"获取资源 {resourceId} 的锁超时");
    }

    await DoProcessAsync(resourceId);
}

注意事项

注意

  1. 分布式锁的粒度要合理，粒度过粗会降低并发性能，粒度过细可能无法保护临界资源。
  2. 加锁区域的代码应尽量简短，避免长时间持有锁导致其他实例长期等待。
  3. 分布式锁依赖 Redis 服务的可用性，生产环境建议使用 Redis 集群或哨兵模式以提高可用性。
  4. 分布式锁不能替代数据库事务，对于数据一致性要求严格的场景，应结合数据库乐观锁或悲观锁使用。