时间戳

首页 > 解决方案 > Parallel.ForEach 中的多个异步等待链接

c# - Parallel.ForEach 中的多个异步等待链接

问题描述

我有一个 Parallel.ForEach 循环，它遍历一个集合。在循环内部，我进行了多次网络 I/O 调用。我使用了 Task.ContinueWith 并嵌套了随后的 async-await 调用。处理的顺序无关紧要，但是每个异步调用的数据应该以同步的方式处理。含义 - 对于每次迭代，从第一个异步调用检索到的数据应该传递给第二个异步调用。在第二个异步调用完成后，来自两个异步调用的数据应该一起处理。

Parallel.ForEach(someCollection, parallelOptions, async (item, state) =>
{
    Task<Country> countryTask = Task.Run(() => GetCountry(item.ID));

    //this is my first async call
    await countryTask.ContinueWith((countryData) =>
    {
        countries.Add(countryData.Result);

        Task<State> stateTask = Task.Run(() => GetState(countryData.Result.CountryID));

        //based on the data I receive in 'stateTask', I make another async call
        stateTask.ContinueWith((stateData) =>
        {
            states.Add(stateData.Result);

            // use data from both the async calls pass it to below function for some calculation
            // in a synchronized way (for a country, its corresponding state should be passed)

            myCollection.ConcurrentAddRange(SomeCalculation(countryData.Result, stateData.Result));
        });
    });
});

我在没有使用 continue await 的情况下尝试了上述方法，但它没有以同步方式工作。现在，上面的代码执行完成，但没有记录被处理。

请问有什么帮助吗？让我知道是否可以添加更多详细信息。

标签： c#asynchronousasync-awaittask-parallel-libraryparallel.foreach

解决方案

由于您的方法涉及 I/O，因此它们应该被编写为真正异步的，而不仅仅是使用Task.Run.

然后你可以Task.WhenAll结合使用Enumerable.Select：

var tasks = someCollection.Select(async item =>
{
    var country = await GetCountryAsync(item.Id);
    var state = await GetStateAsync(country.CountryID);
    var calculation = SomeCalculation(country, state);

    return (country, state, calculation);
});

foreach (var tuple in await Task.WhenAll(tasks))
{
    countries.Add(tuple.country);
    states.Add(tuple.state);
    myCollection.AddRange(tuple.calculation);
}

这将确保每个country> state>calculation顺序发生，但每个item都是同时和异步处理的。

根据评论更新

using var semaphore = new SemaphoreSlim(2);
using var cts = new CancellationTokenSource();

int failures = 0;

var tasks = someCollection.Select(async item =>
{
    await semaphore.WaitAsync(cts.Token);
    
    try
    {
        var country = await GetCountryAsync(item.Id);
        var state = await GetStateAsync(country.CountryID);
        var calculation = SomeCalculation(country, state);

        Interlocked.Exchange(ref failures, 0);

        return (country, state, calculation);
    {
    catch
    {
        if (Interlocked.Increment(ref failures) >= 10)
        {
            cts.Cancel();
        }
        throw;
    }
    finally
    {
        semaphore.Release();
    }
});

信号量保证最多 2 个并发异步操作，取消令牌将在连续 10 次异常后取消所有未完成的任务。

这些Interlocked方法确保failures以线程安全的方式访问。

进一步更新

使用 2 个信号量来防止多次迭代可能更有效。

将所有列表添加封装到一个方法中：

void AddToLists(Country country, State state, Calculation calculation)
{
    countries.Add(country);
    states.Add(state);
    myCollection.AddRange(calculation);
}

然后，您可以允许 2 个线程同时处理 Http 请求，并允许 1 个线程执行添加，从而使该操作线程安全：

using var httpSemaphore = new SemaphoreSlim(2);
using var listAddSemaphore = new SemaphoreSlim(1);
using var cts = new CancellationTokenSource();

int failures = 0;

await Task.WhenAll(someCollection.Select(async item =>
{
    await httpSemaphore.WaitAsync(cts.Token);
    
    try
    {
        var country = await GetCountryAsync(item.Id);
        var state = await GetStateAsync(country.CountryID);
        var calculation = SomeCalculation(country, state);

        await listAddSemaphore.WaitAsync(cts.Token);
        AddToLists(country, state, calculation);

        Interlocked.Exchange(ref failures, 0);
    {
    catch
    {
        if (Interlocked.Increment(ref failures) >= 10)
        {
            cts.Cancel();
        }
        throw;
    }
    finally
    {
        httpSemaphore.Release();
        listAddSemaphore.Release();
    }
}));

推荐阅读