Asp.net core Web API 并行调用，批处理丢失一些数据

如何解决Asp.net core Web API 并行调用，批处理丢失一些数据

在 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();