我有一个 Azure 服务总线队列,我正在其中接收 1 到 10 条具有相同“密钥”的消息。其中一条消息需要通过长时间运行的操作来处理。完成后,数据库将被更新,其他消息将对其进行检查。但是,与此同时,其他消息将重新排队,以便进程不会丢失。
但要点是这个长时间运行的操作CANNOT对于同一个密钥同时运行,并且不应运行多次。
这是我到目前为止所得到的:
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}");
}
}
它没有按预期工作,因为它会创建多个信号量,突然我的长时间运行操作使用相同的密钥运行两次。我认为问题在于整个操作不是原子的。
解决这个问题的最佳方法是什么?