异步/等待竞争条件答案

【问题标题】：async/await race condition异步/等待竞争条件
【发布时间】：2016-03-17 12:49:45
【问题描述】：

我相信我在下面有一个竞争条件。我正在手动构建一个带有 JSON 输出的 HttpResponseMessage 以异步流式传输。问题似乎出在计数器 (i) 上。我想在列表中第一次写入之后的任何元素之前添加一个逗号。

在列表的开头，有时第一次写入之后的前几条记录（我见过最多 3 条）不会有前面的逗号。该数字不一致，有时按预期工作。我没有在我的本地机器上看到它，但在具有更强大硬件的部署环境中它存在。

var LastUpdate = JsonConvert.SerializeObject(dt);
var pre = $"{{ \"LastUpdate\": {LastUpdate}, \"List\":[";
var post = "]}";

HttpResponseMessage response = Request.CreateResponse();
response.Content = new PushStreamContent(
    async (stream, http, context) =>
    {
        try
        {
            int i = 0;
            var buffer = Encoding.UTF8.GetBytes(pre);
            await stream.WriteAsync(buffer, 0, buffer.Length, cancellationToken).ConfigureAwait(false);

            var query = getQuery(id);                               
            await query
                .ForEachAsync(async entity =>
                {
                    var student = MapRecord(entity);
                    if (student != null)
                    {
                        var json = JsonConvert.SerializeObject(student);
                        buffer = Encoding.UTF8.GetBytes(((i > 0) ? ", " : "") + json);
                        await stream.WriteAsync(buffer, 0, buffer.Length, cancellationToken).ConfigureAwait(false);
                        i++;
                    }
                }, cancellationToken).ConfigureAwait(false);

            buffer = Encoding.UTF8.GetBytes(post);
            await stream.WriteAsync(buffer, 0, buffer.Length, cancellationToken).ConfigureAwait(false);
        }

【问题讨论】：

ForEachAsync 是如何声明的？
@YuvalItzchakov 也许 OP 正在使用 QueryableExtensions.ForEachAsync

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

【解决方案1】：

如果您使用的是QueryableExtentions.ForEachAsync（感谢@juharr），那么是的，您有竞争条件。

方法的签名是：

public static Task ForEachAsync<T>(
    this IQueryable<T> source,
    Action<T> action
)

请注意，该方法接受Action<T>。在异步世界中，这相当于async void。这意味着每次您在异步委托中 await 时，ForEachAsync 迭代器实际上会继续下一个元素，而不是等待您的委托完成。

相反（如果不是在非常大的数据集上调用查询），请在其中使用常规的 foreach 语句和 await：

foreach (var entity in query)
{
    var student = MapRecord(entity);
    if (student != null)
    {
        var json = JsonConvert.SerializeObject(student);
        buffer = Encoding.UTF8.GetBytes(((i > 0) ? ", " : "") + json);
        await stream.WriteAsync(buffer, 0, buffer.Length, cancellationToken)
                    .ConfigureAwait(false);
        i++;
    }
}

【讨论】：

使用 foreachasync 防止数据库级别的阻塞（此查询返回约 100 万行），关于同步计数器还有其他建议吗？
如果是 100 万行，foreach 可能不是最好的主意。您是否有任何理由实际手动进行序列化？如果这是不可避免的，我建议只使用同步委托并使用stream.Write 而不是stream.WriteAsync。或者更好的是，为 EF 创建一个拉取请求，并在 ForEachAsync 中添加一个用于调用异步委托的选项：)
引入整个结果集，映射和序列化命中内存异常。手动构造并不理想，但确实允许一次传输少量比特。这种方法占用的资源非常少。
也许更聪明的想法是使用JsonTextWriter。请参阅this 问题。

【解决方案2】：

如果您主要关心的是 (i) 计数器，您可以使用 Interlocked.Increment 线程安全地递增 i。这将导致同步的少量性能争用，但您可以通过这种方式持续更新 i。

例子：

Interlocked.Increment(ref i);

Interlocked.Increment MSDN

【讨论】：