问题描述
在 asp.net core 3.x 中创建了一个 web API,它负责将数据保存到 Azure 服务总线队列中,我们将处理这些数据以进行报告。
API 负载太高,因此我们决定将每个请求的数据保存在内存中。一旦数据限制增加到一定限制(> 50计数)下一个请求(51)将从内存中获取所有数据并一次性保存到服务总线并清除内存缓存。
对于顺序请求,所有逻辑都可以正常工作,但是当并行加载时,很少有数据丢失,我认为这是因为一个批处理请求需要一些时间,然后所有数据问题就开始了。
我做了一些研究并找到了文章并使用了 SemaphoreSlim。它工作正常,但这是好方法吗?正如您在下面的代码中看到的,我正在阻止每个请求,但实际上我想在处理批处理时锁定。我试图在 if 条件 内移动锁,但它不起作用。 https://medium.com/swlh/async-lock-mechanism-on-asynchronous-programing-d43f15ad0b3
using (await lockThread.LockAsync())
{
var topVisitedTiles = _service.GetFromCache(CacheKey);
if (topVisitedTiles?.Count >= 50)
{
topVisitedTiles?.Add(link);
await _service.AddNewQuickLinkAsync(topVisitedTiles);
_service.SetToCache(CacheKey,new List<TopVisitedTilesItem>());
return Ok(link.Title);
}
topVisitedTiles?.Add(link);
_service.SetToCache(CacheKey,topVisitedTiles);
}
return Ok(link.Title);
我从研究中得到一些 concurrentbag 和 blockingcollection 的帮助,但我不知道如何在我的情况下使用。你的小指导会帮助我。
解决方法
如果您不想深入研究包或队列的并行实现,可以使用 Task Parallel Library。
在你的情况下可以使用这样的东西
// Define a buffer block with size = 10
var batchBlock = new BatchBlock<string>(10);
// Define an ActionBlock that processes batches received from BatchBlock
var processingBlock = new ActionBlock<string[]>((messages) =>
{
Console.WriteLine("-------------");
Console.WriteLine($"Number of messages: {messages.Length}");
Console.WriteLine($"Messages: {string.Join(",",messages)}");
});
// Link processing block to batchBloack.
batchBlock.LinkTo(processingBlock);
batchBlock.Completion.ContinueWith((t) =>
{
processingBlock.Complete();
});
var task1 = Task.Run(async () =>
{
for (int i = 0; i < 50; i++)
{
await batchBlock.SendAsync($"Message {i}");
}
});
var task2 = Task.Run(async () =>
{
for (int i = 50; i < 100; i++)
{
await batchBlock.SendAsync($"Message {i}");
}
});
await Task.WhenAll(task1,task2);
// Complete pipeline. You can leave it as active if you want.
batchBlock.Complete();
processingBlock.Completion.Wait();