如何同时完成对可观察集合的 HTTP 调用？答案

【问题标题】：How to concurrently complete HTTP calls on an observable collection?如何同时完成对可观察集合的 HTTP 调用？
【发布时间】：2020-09-28 15:31:34
【问题描述】：

在 WPF .net 核心应用中有以下内容：

可观察的项目集合 (itemObservCollection)。
一个静态的只读 HttpClient _httpclient
XML 响应

我正在对可观察集合中的每个项目（集合中的 0 到 1000 个项目）的 api 进行 URL 调用。返回是 XML。使用 XElement 解析 XML。可观察集合中的属性值是从 XML 更新的。

Task.Run 用于从 UI 线程运行操作。 Parallel.Foreach 用于进行并行调用。

我觉得我的解决方案过于复杂。有没有办法简化这个？通过单击按钮调用 UpdateItems()。

private async Task UpdateItems()
{
    try
    {
        await Task.Run(() => Parallel.ForEach(itemObservCollection, new ParallelOptions { MaxDegreeOfParallelism = 12 }, async item =>
        {
            try
            {
                var apiRequestString = $"http://localhost:6060/" + item.Name;
                HttpResponseMessage httpResponseMessage = await _httpclient.GetAsync(apiRequestString);
                var httpResponseStream = await httpResponseMessage.Content.ReadAsStreamAsync();
                StringBuilder sb = new StringBuilder(1024);
                XElement doc = XElement.Load(httpResponseStream);
                foreach (var elem in doc.Descendants())
                {
                    if (elem.Name == "ItemDetails")
                    {
                        var itemUpdate = itemObservCollection.FirstOrDefault(updateItem => updateItem.Name == item.Name);
                        if (itemUpdate != null)
                        {
                            itemUpdate.Price = decimal.Parse(elem.Attribute("Price").Value);
                            itemUpdate.Quantity = int.Parse(elem.Attribute("Quantity").Value);
                        }
                    }
                }
            }
            catch (Exception ex)
            {
                LoggerTextBlock.Text = ('\n' + ex.ToString());
            }
        }));
    }
    catch (Exception ex)
    {
        LoggerTextBlock.Text = ('\n' + ex.ToString());
    }
}

【问题讨论】：

严重错误是Parallel.ForEach 不处理异步操作，因为这样做没有任何意义。您传递给它的 lambda 本质上是无法等待的 async void 方法。这段代码几乎同时触发，从不等待它们完成
Task.Run is used to run the operation off the UI thread. 否，Task.Run 用于在后台线程上运行
@Fildor 答案只是“是”，这不是意见。问题只是标题。事实上，我怀疑有几个重复的问题可以回答真正的问题
共识似乎是Task.Run和Parallel.ForEach不正确。我将看看 TPL 数据流。感谢@TheodorZoulias、Fildor 和 Panagiotis Kanavos
一个错误 - DOP 参数不可见。我可能将值与另一个问题混淆了。 DOP 可以是网络或远程服务器允许的任何内容

标签： c# wpf asynchronous async-await task-parallel-library

【解决方案1】：

您可以创建一个任务数组并使用Task.WhenAll 等待它们。

以下示例代码启动 ObservableCollection<int> 中的每个项目的任务，然后异步等待所有任务完成：

ObservableCollection<int> itemObservCollection = 
    new ObservableCollection<int>(Enumerable.Range(1, 10));

async Task SendAsync()
{
    //query the HTTP API here...
    await Task.Delay(1000);
}

await Task.WhenAll(itemObservCollection.Select(x => SendAsync()).ToArray());

如果您想限制并发请求的数量，您可以遍历源集合的子集以批量发送请求或使用SemaphoreSlim 来限制实际并发请求的数量：

Task[] tasks = new Task[itemObservCollection.Count];
using (SemaphoreSlim semaphoreSlim = new SemaphoreSlim(12))
{
    for (int i = 0; i < itemObservCollection.Count; ++i)
    {
        async Task SendAsync()
        {
            //query the HTTP API here...
            try
            {
                await Task.Delay(5000);
            }
            finally
            {
                semaphoreSlim.Release();
            }
        }

        await semaphoreSlim.WaitAsync();
        tasks[i] = SendAsync();
    }
    await Task.WhenAll(tasks);
}

【讨论】：

这里没有节流。
像MaxDegreeOfParallelism 一样进行节流？这只是遍历源集合的子集的问题，例如使用Take 扩展方法：itemObservCollection.Take(10)...itemObservCollection.Skip(10) ...
OP 想要为集合中的每个项目发送请求，但同时限制为 12 个。 Take 不会做到这一点。
Take(x).Select(_ => SendAsync()) 将启动 x 任务。然后，您可以等待它们并开始下一个 x 任务，依此类推，具体取决于您的要求。这将限制并发请求数。
在这种情况下，您需要等待所有x 任务完成后才能开始下一批。信号量会是更好的解决方案。