c# - .Net 中的 key 锁

Question

我有一个 Azure 服务总线队列，我正在其中接收 1 到 10 条具有相同“ key ”的消息。其中一条消息需要通过长时间运行的操作来处理。完成后，数据库将被更新，其他消息将对其进行检查。但是，与此同时，其他消息将重新排队，以便进程不会丢失。

但要点是，这个长时间运行的操作不能针对同一个 key 同时运行，并且不应运行多次。

这是我到目前为止所得到的:

void Main()
{
    Enumerable.Range(1, 1000)
              .AsParallel()
              .ForAll(async i => await ManageConcurrency(i % 2, async () => await Task.Delay(TimeSpan.FromSeconds(10)))); 
}

private readonly ConcurrentDictionary<int, SemaphoreSlim> _taskLocks = new ConcurrentDictionary<int, SemaphoreSlim>();

private async Task<bool> ManageConcurrency(int taskId, Func<Task> task)
{
    SemaphoreSlim taskLock = null;

    try
    {
        if (_taskLocks.TryGetValue(taskId, out taskLock))
        {
            if (taskLock.CurrentCount == 0)
            {
                Console.WriteLine($"{DateTime.Now.ToString("hh:mm:ss.ffffff")},  {taskId}, I found. No available.. Thread Id: {Thread.CurrentThread.ManagedThreadId}");
                return false;
            }

            taskLock.Wait();

            Console.WriteLine($"{DateTime.Now.ToString("hh:mm:ss.ffffff")},  {taskId}, I found and took. Thread Id: {System.Threading.Thread.CurrentThread.ManagedThreadId}");
        }
        else
        {
            taskLock = new SemaphoreSlim(1, 1);
            taskLock = _taskLocks.GetOrAdd(taskId, taskLock);
            if (taskLock.CurrentCount == 0)
            {
                Console.WriteLine($"{DateTime.Now.ToString("hh:mm:ss.ffffff")},  {taskId}, I didn't find, and then found/created. None available.. Thread Id: {System.Threading.Thread.CurrentThread.ManagedThreadId}");
                return false;
            }
            else
            {
                taskLock.Wait(TimeSpan.FromSeconds(1));

                Console.WriteLine($"{DateTime.Now.ToString("hh:mm:ss.ffffff")},  {taskId}, I didn't find, then found/created, and took. Thread Id: {System.Threading.Thread.CurrentThread.ManagedThreadId}");
            }
        }

        Console.WriteLine($"{DateTime.Now.ToString("hh:mm:ss.ffffff")},  {taskId}, Lock pulled for TaskId {taskId}, Thread Id: {System.Threading.Thread.CurrentThread.ManagedThreadId}");

        await task.Invoke();

        return true;
    }
    catch (Exception e)
    {
        ;
        return false;
    }
    finally
    {
        //taskLock?.Release();

        _taskLocks.TryRemove(taskId, out taskLock);

        //Console.WriteLine($"I removed. Thread Id: {System.Threading.Thread.CurrentThread.ManagedThreadId}");
    }
}

它没有按预期工作，因为它会创建多个信号量，突然我的长时间运行操作使用相同的 key 运行两次。我认为问题在于整个操作不是原子的。

解决此问题的最佳方法是什么？

Answer 1

您正确地认识到您需要确保每个键仅创建一个信号量。标准习惯用法是:

var dict = new ConcurrentDictionary<TKey, Lazy<SemaphoreSlim>>();
...
var sem = dict.GetOrAdd( , _ => new new Lazy<SemaphoreSlim>(() => SemaphoreSlim(1, 1))).Value;

可能会创建多个懒惰，但只有其中一个会被揭示和具体化。

此外，依赖内存状态是一种值得怀疑的做法。如果您的队列处理应用程序回收并且所有信号量都丢失怎么办？您最好使用持久存储来跟踪此锁定信息。