C#：使用 Parallel.ForEach 和异步操作限制最大并发操作

Question

我正在尝试使用 asp.net core 2.1 实现自托管 Web 服务，但遇到了实现后台长时间执行任务的问题。

由于每个ProcessSingle方法（在下面的代码sn-p中）的高CPU负载和时间消耗，我想限制同时执行的任务的数量。但是我几乎可以立即看到Parallel.ForEachstart 中的所有任务，尽管我设置了MaxDegreeOfParallelism = 3

我的代码是（简化版）：

public static async Task<int> Work()
{
    var id = await CreateIdInDB() // async create record in DB

    // run background task, don't wait when it finishes
    Task.Factory.StartNew(async () => {
        Parallel.ForEach(
            listOfData,
            new ParallelOptions { CancellationToken = token, MaxDegreeOfParallelism = 3 },
            async x => await ProcessSingle(x));
    });

    // return created id immediately
    return id;
}

public static async Task ProcessSingle(MyInputData inputData)
{
    var dbData = await GetDataFromDb(); // get data from DB async using Dapper
    // some lasting processing (sync)
    await SaveDataToDb(); // async save processed data to DB using Dapper
}

如果我理解正确，问题出在 Parallel.ForEach 内的async x => await ProcessSingle(x)，不是吗？

有人可以描述一下，它应该如何以正确的方式实施？

更新

由于我的问题有些模棱两可，有必要关注主要方面：

1. ProcessSingle方法分三部分：

   • 从数据库异步获取数据

   • 进行长时间高 CPU 负载的数学计算

   • 将结果保存到数据库异步

2. 这个问题包括两个独立的：

   • 如何降低 CPU 使用率（例如同时运行不超过三个数学计算）？

   • 如何保持 ProcessSingle 方法的结构 - 因为异步 DB 调用而使它们保持异步。

希望现在会更清楚。

附：已经给出了合适的答案，它可以工作（特别感谢@MatrixTai）。编写此更新是为了进行一般说明。

Answer 1

更新

正如我刚刚注意到你在评论中提到的，问题是由数学计算引起的。

计算和更新DB的部分最好分开。

计算部分使用Parallel.ForEach()，这样可以优化你的工作，你可以控制线程数。

只有在所有这些任务完成之后。使用async-await 将您的数据更新到数据库，而不需要我提到的SemaphoreSlim。

public static async Task<int> Work()
{
    var id = await CreateIdInDB() // async create record in DB

    // run background task, don't wait when it finishes
    Task.Run(async () => {

        //Calculation Part
        ConcurrentBag<int> data = new ConcurrentBag<int>();
        Parallel.ForEach(
            listOfData,
            new ParallelOptions { CancellationToken = token, MaxDegreeOfParallelism = 3 },
            x => {ConcurrentBag.Add(calculationPart(x))});

        //Update DB part
        int[] data_arr = data.ToArray();
        List<Task> worker = new List<Task>();
        foreach (var i in data_arr)
        {
            worker.Add(DBPart(x));
        }
        await Task.WhenAll(worker);
    });

    // return created id immediately
    return id;
}

---

当您在Parallel.forEach 中使用async-await 时，它们肯定是一起开始的。

首先，请阅读question 第一个和第二个答案。将这两者结合起来毫无意义。

其实async-await会最大化可用线程的使用率，所以直接用吧。

public static async Task<int> Work()
{
    var id = await CreateIdInDB() // async create record in DB

    // run background task, don't wait when it finishes
    Task.Run(async () => {
        List<Task> worker = new List<Task>();
        foreach (var i in listOfData)
        {
            worker.Add(ProcessSingle(x));
        }
        await Task.WhenAll(worker);
    });

    // return created id immediately
    return id;
}

但是这里有另一个问题，在这种情况下，这些任务仍然一起开始，消耗你的 CPU 使用率。

为了避免这种情况，请使用SemaphoreSlim

public static async Task<int> Work()
{
    var id = await CreateIdInDB() // async create record in DB

    // run background task, don't wait when it finishes
    Task.Run(async () => {
        List<Task> worker = new List<Task>();
        //To limit the number of Task started.
        var throttler = new SemaphoreSlim(initialCount: 20);
        foreach (var i in listOfData)
        {
            await throttler.WaitAsync();
            worker.Add(Task.Run(async () =>
            {
                await ProcessSingle(x);
                throttler.Release();
            }
            ));
        }
        await Task.WhenAll(worker);
    });

    // return created id immediately
    return id;
}

阅读更多How to limit the amount of concurrent async I/O operations?。

另外，当简单的Task.Run() 足以完成你想做的工作时，不要使用Task.Factory.StartNew()，请阅读Stephen Cleary 的这篇出色的article。

Answer 2

如果您更熟悉“传统”并行处理概念，请像这样重写您的 ProcessSingle() 方法：

public static void ProcessSingle(MyInputData inputData)
{
    var dbData = GetDataFromDb(); // get data from DB async using Dapper
    // some lasting processing (sync)
    SaveDataToDb(); // async save processed data to DB using Dapper
}

当然，您最好也以类似的方式更改 Work() 方法。