这种并发问题可以用 async/await 解决吗？

Question

我有这样的功能：

static void AddResultsToDb(IEnumerable<int> numbers)
{
    foreach (int number in numbers)
    {
        int result = ComputeResult(number); // This takes a long time, but is thread safe.
        AddResultToDb(number, result); // This is quick but not thread safe.
    }
}

我可以解决这个问题，例如，使用Parallel.ForEach 计算结果，然后使用常规foreach 将结果添加到数据库中。

但是，出于教育目的，我想要一个围绕等待/异步的解决方案。但是，无论我读了多少关于它的内容，我都无法将其全神贯注。如果 await/async 在这种情况下不适用，我想了解原因。

Answer 1

正如其他人所建议的那样，这不是使用async/await 的情况，因为那是异步的。你正在做的是并发。微软有一个专门为此而设计的框架，它很好地解决了这个问题。

因此，出于学习目的，您应该使用 Microsoft 的响应式框架（又名 Rx）- NuGet System.Reactive 并添加 using System.Reactive.Linq; - 然后您可以这样做：

static void AddResultsToDb(IEnumerable<int> numbers)
{
    numbers 
        .ToObservable()
        .SelectMany(n => Observable.Start(() => new { n, r = ComputeResult(n) }))
        .Do(x => AddResultToDb(x.n, x.r))
        .Wait();
}

SelectMany/Observable.Start 组合允许尽可能多的ComputeResult 调用同时发生。 Rx 的好处是它会序列化结果，这样一次只有一个调用会转到AddResultToDb。

---

要控制并行度，您可以将 SelectMany 更改为 Select/Merge，如下所示：

static void AddResultsToDb(IEnumerable<int> numbers)
{
    numbers 
        .ToObservable()
        .Select(n => Observable.Start(() => new { n, r = ComputeResult(n) }))
        .Merge(maxConcurrent: 2)
        .Do(x => AddResultToDb(x.n, x.r))
        .Wait();
}

Answer 2

async 和 await 模式并不真正适合您的第一种方法。它非常适合IO Bound 工作负载以实现可扩展性，或者适合具有 UI 响应能力的框架。它不太适合原始 CPU 工作负载。

但是您仍然可以从并行处理中获益，因为您的第一种方法昂贵并且线程安全。

在下面的示例中，我使用Parallel LINQ (PLINQ) 来流畅地表达结果，而无需担心预先确定大小的数组 / 并发集合 / 锁定，尽管您可以使用其他 TPL 功能，例如 Parallel.For/ForEach

// Potentially break up the workloads in parallel
// return the number and result in a ValueTuple
var results = numbers.AsParallel()
                     .Select(x => (number: x, result: ComputeResult(x)))
                     .ToList();

// iterate through the number and results and execute them serially 
foreach (var (number, result) in results)
   AddResultToDb(number, result);

_{注意：这里的假设是顺序不重要}

---

补充

您的方法AddResultToDb 看起来只是将结果插入到数据库 中，这是IO Bound 并且值得async，此外可能会获取所有结果一次并将它们插入 bulk/batch 节省 往返行程

---

来自评论信用@TheodorZoulias

到preserve the order，你可以使用AsOrdered的方法，在 一些性能损失的代价。可能的表现 改进是去掉ToList()，这样结果就加起来了 与计算同时发送到数据库。

为了尽快提供结果，这可能是一个不错的选择 禁用happens by default 的部分缓冲的想法，通过 链接方法 查询中的.WithMergeOptions(ParallelMergeOptions.NotBuffered)

var results = numbers.AsParallel()
                     .Select(x => (number: x, result: ComputeResult(x)))
                     .WithMergeOptions(ParallelMergeOptions.NotBuffered)
                     .AsOrdered();

示例

---

其他资源

ParallelEnumerable.AsOrdered Method

启用对数据源的处理，就好像它是有序的一样，覆盖 无序的默认值。 AsOrdered 只能在非泛型上调用 序列

ParallelEnumerable.WithMergeOptions

设置此查询的合并选项，指定查询的方式 将缓冲输出。

ParallelMergeOptions Enum

NotBuffered 使用没有输出缓冲区的合并。计算出结果元素后，立即将该元素提供给 查询的消费者。

Answer 3

async/await 的情况并非如此，因为听起来ComputeResult 的计算成本很高，而不是只需要很长的、不确定的时间。 aync/await 更适合您真正等待的任务。 Parallel.ForEach 实际上会处理您的工作负载。

如果有的话，AddResultToDb 是您想要异步/等待的 - 您将等待外部操作完成。

好深入的解释：https://stackoverflow.com/a/35485780/127257

Answer 4

老实说，使用Parallel.For 似乎是最简单的解决方案，因为您的计算可能会受 CPU 限制。 Async/await 更适合 I/O 绑定操作，因为它不需要另一个线程等待 I/O 操作完成（请参阅there is no thread）。

话虽如此，您仍然可以将 async/await 用于放置在线程池中的任务。所以你可以这样做。

static void AddResultToDb(int number)
{
    int result = ComputeResult(number); 
    AddResultToDb(number, result); 
}

static async Task AddResultsToDb(IEnumerable<int> numbers)
{
    var tasks = numbers.Select
    (
        number => Task.Run( () => AddResultToDb(number) )  
    )
    .ToList();

    await Task.WhenAll(tasks);
}