资源并发，允许每个给定资源访问一个或多个线程

Question

以下是我的问题。

我有一个 API 控制器，里面有一个 API 端点（资源）。

 api/myResource/{id}/do-something

我正在开发一个中间件，它将根据一些业务规则限制对该资源的访问。在这个中间件中，我正在匹配传入的请求，我正在解析 URI，并且我想允许访问它（让管道流动）或简单地返回 412 状态代码以防达到允许的线程数限制 FOR给定的资源（例如）

 api/myResource/1/do-something /// should allow 2 concurrent accesses.
 api/myResource/2/do-something /// should allow 10 concurrent accesses.
 api/myResource/3/do-something /// should allow 1 concurrent accesses.

为此，我已经开始实施我将附上的解决方案。

internal class AsyncLock<TKey>
{
    private static readonly ConcurrentDictionary<TKey, SemaphoreSlim> _safeSemaphores
        = new ConcurrentDictionary<TKey, SemaphoreSlim>();

    internal async Task<SemaphoreSlim> TryLockAsync(TKey key, int maxConcurrentCount)
    {
        if (!_safeSemaphores.TryGetValue(key, out SemaphoreSlim semaphore))
        {
            semaphore = new SemaphoreSlim(maxConcurrentCount, maxConcurrentCount);

            _safeSemaphores.TryAdd(key, semaphore);
        }

        await semaphore.WaitAsync();

        return semaphore;
    }

    internal SemaphoreSlim TryLock(TKey key, int maxConcurrentCount)
    {
        if (!_safeSemaphores.TryGetValue(key, out SemaphoreSlim semaphore))
        {
            semaphore = new SemaphoreSlim(maxConcurrentCount, maxConcurrentCount);

            _safeSemaphores.TryAdd(key, semaphore);
        }

        semaphore.Wait();

        return semaphore;
    }
}

这就是它的使用方式（它允许2个并发访问，当然这是这个问题的主题，它不会是硬编码的2而是在管道的早期确定）

AsyncLock<string> _lock = new AsyncLock<string>();
SemaphoreSlim semaphore = await _lock .TryLockAsync(key, 2);

if (semaphore.CurrentCount != 0)
{
    context.Response.StatusCode = 200;
    //await _next(context);
    semaphore.Release();
}
else
{
    context.Response.StatusCode = 412;
}

它的行为方式是不可预测的。我正在使用 4 个线程进行测试，有时它按预期工作，有时它们都返回 200，有时它们都卡住了，我的意思是每次都是组合。

如果能帮我解决这个问题，我将不胜感激。

Answer 1

使用Monitor.TryEnter 似乎最简单。

object _lock = new object();


void RunIfNotLocked()
{
    bool lockAcquired = false;

    Monitor.TryEnter(_lock, ref lockAcquired);
    if ( !lockAcquired ) 
    {
        //Skip
        return;
    }
    try
    {
        DoSomething();
    }
    finally
    {         
        Monitor.Exit(_lock);
    }
}

Answer 2

我已经提出了这个解决方案，它似乎按预期工作。

internal sealed class AsyncLock<TKey>
{
    public readonly Dictionary<TKey, SemaphoreSlim> _semaphores = new Dictionary<TKey, SemaphoreSlim>();

    internal IDisposable Lock(TKey key, int maxDegreeOfParallelism)
    {
        bool acquired = GetOrAdd(key, maxDegreeOfParallelism).Wait(0);

        return acquired
            ? new Releaser(key, this)
            : null;
    }

    internal async Task<IDisposable> LockAsync(TKey key, int maxDegreeOfParallelism = 1)
    {
        bool acquired = await GetOrAdd(key, maxDegreeOfParallelism).WaitAsync(0);

        return acquired
            ? new Releaser(key, this)
            : null;
    }

    private SemaphoreSlim GetOrAdd(TKey key, int maxConcurrencyCount = 1)
    {
        lock (_semaphores)
        {
            if (!_semaphores.TryGetValue(key, out SemaphoreSlim semaphore))
            {
                _semaphores[key] = semaphore = new SemaphoreSlim(maxConcurrencyCount, maxConcurrencyCount);
            }

            return semaphore;
        }
    }

    private sealed class Releaser : IDisposable
    {
        private AsyncLock<TKey> _parent;

        public void Dispose()
        {
            lock (_parent._semaphores)
            {
                if (_parent._semaphores.TryGetValue(Key, out SemaphoreSlim semaphore)) semaphore.Release();
            }
        }

        public TKey Key { get; }

        public Releaser(TKey key, AsyncLock<TKey> parent)
        {
            Key = key;
            _parent = parent;
        }
    }
}

使用它：

AsyncLock<string> _asyncLock = new AsyncLock<string>();

IDisposable disposable = _asyncLock.Lock(key, maxConcurrency: 2);

if (disposable is null)
{
    /// thread skipped
}
else
{
    ///  thread entered
    disposable.Dispose();
}

Answer 3

这是您可以使用的KeyedNamedSemaphore 类。它为每个密钥存储一个 named Semaphore。创建密钥后，关联的信号量将保留在字典中，直到释放 KeyedNamedSemaphore 实例。

public class KeyedNamedSemaphore<TKey> : IDisposable
{
    private readonly ConcurrentDictionary<TKey, Semaphore> _perKey;
    private readonly string _prefix;

    public KeyedNamedSemaphore(string prefix = null,
        IEqualityComparer<TKey> keyComparer = null)
    {
        _perKey = new ConcurrentDictionary<TKey, Semaphore>(keyComparer);
        _prefix = prefix ?? $@"Global\{System.Reflection.Assembly
            .GetExecutingAssembly().GetName().Name}-";
    }

    public bool TryLock(TKey key, int maximumConcurrency, out Semaphore semaphore)
    {
        if (!_perKey.TryGetValue(key, out semaphore))
        {
            var newSemaphore = new Semaphore(maximumConcurrency, maximumConcurrency,
                $"{_prefix}{key}");
            semaphore = _perKey.GetOrAdd(key, newSemaphore);
            if (semaphore != newSemaphore) newSemaphore.Dispose();
        }
        var acquired = semaphore.WaitOne(0);
        if (!acquired) semaphore = null;
        return acquired;
    }

    public void Dispose()
    {
        foreach (var key in _perKey.Keys)
            if (_perKey.TryRemove(key, out var semaphore)) semaphore.Dispose();
    }
}

使用示例：

var locker = new KeyedNamedSemaphore<string>();
if (locker.TryLock("api/myResource/1/do-something", 10, out var semaphore))
{
    try
    {
        await DoSomethingAsync();
    }
    finally { semaphore.Release(); }
}
else
{
    await DoSomethingElseAsync();
}