c# - 异步/等待竞争条件

Question

我相信我有下面的竞争条件。我正在手动构建一个带有 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);
        }

Answer 1

如果您使用 QueryableExtentions.ForEachAsync (感谢@juharr)，那么是的，你有一个竞争条件。

方法的签名是:

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

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

相反(如果未在非常大的数据集上调用查询)，请使用常规 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++;
    }
}