如何在 EF Core 3.1 中以异步方式使用 GroupBy？

Question

当我使用 GroupBy 作为对 EFCore 的 LINQ 查询的一部分时，我收到错误 System.InvalidOperationException: Client-side GroupBy is not supported。

这是因为 EF Core 3.1 尝试尽可能多地在服务器端评估查询，而不是在客户端评估它们，并且调用无法转换为 SQL。

所以下面的语句不起作用，会产生上面提到的错误：

var blogs = await context.Blogs
    .Where(blog => blog.Url.Contains("dotnet"))
    .GroupBy(t => t.BlobNumber)
    .Select(b => b)
    .ToListAsync();

现在显然解决方案是在调用 GroupBy() 之前使用 .AsEnumerable() 或 .ToList()，因为这明确告诉 EF Core 您想要进行分组客户端。有一个关于这个on GitHub和in the Microsoft docs的讨论。

var blogs = context.Blogs
    .Where(blog => blog.Url.Contains("dotnet"))
    .AsEnumerable()
    .GroupBy(t => t.BlobNumber)
    .Select(b => b)
    .ToList();

但是，这不是异步的。我怎样才能使它异步？

如果我将 AsEnumerable() 更改为 AsAsyncEnumerable()，则会收到错误消息。如果我尝试将 AsEnumerable() 更改为 ToListAsync() 则 GroupBy() 命令失败。

我正在考虑将它包装在 Task.FromResult 中，但这实际上是异步的吗？还是数据库查询还是同步的，只有后面的分组是异步的？

var blogs = await Task.FromResult(context.Blogs
    .Where(blog => blog.Url.Contains("dotnet"))
    .AsEnumerable()
    .GroupBy(t => t.BlobNumber)
    .Select(b => b)
    .ToList());

或者如果这不起作用，还有其他方法吗？

Answer 1

我认为你唯一的方法就是这样做

var blogs = await context.Blogs
    .Where(blog => blog.Url.Contains("dotnet"))
    .ToListAsync();

var groupedBlogs = blogs.GroupBy(t => t.BlobNumber).Select(b => b).ToList();

因为 GroupBy 无论如何都会在客户端进行评估

Answer 2

此查询并未尝试对 SQL/EF Core 意义上的数据进行分组。不涉及聚合。

它正在加载所有详细信息行，然后将它们批处理到客户端上的不同存储桶中。 EF Core 不参与其中，这是纯粹的客户端操作。相当于：

var blogs = await context.Blogs
    .Where(blog => blog.Url.Contains("dotnet"))
    .ToListAsync();

var blogsByNum = blogs.ToLookup(t => t.BlobNumber);

加速分组

批处理/分组/查找操作完全受 CPU 限制，因此加速它的唯一方法是并行化，即使用所有 CPU 对数据进行分组，例如：

var blogsByNum = blogs.AsParallel()
                      .ToLookup(t => t.BlobNumber);

ToLookup 或多或少与GroupBy().ToList() 相比 - 它根据键将行分组到存储桶中

加载时分组

另一种方法是异步加载结果，并在它们到达时将它们放入存储桶中。为此，我们需要AsAsyncEnumerable()。 ToListAsync() 一次返回所有结果，所以不能使用。

这种方法与ToLookup 的做法非常相似。

var blogs = await context.Blogs
    .Where(blog => blog.Url.Contains("dotnet"));

var blogsByNum=new Dictionary<string,List<Blog>>();

await foreach(var blog in blogs.AsAsyncEnumerable())
{
    if(blogsByNum.TryGetValue(blog.BlobNumber,out var blogList))
    {
        blogList.Add(blog);
    }
    else
    {
        blogsByNum[blog.BlobNumber=new List<Blog>(100){blog};
    }
}

查询是通过调用AsAsyncEnumerable() 来执行的。结果是异步到达的，所以现在我们可以在迭代时将它们添加到存储桶中。

capacity 参数用于列表构造函数中，以避免重新分配列表的内部缓冲区。

使用 System.LINQ.Async

如果我们对 IAsyncEnumerable 本身进行 LINQ 操作，事情会容易得多。 This extension 命名空间提供了这一点。它由 ReactiveX 团队开发。可通过NuGet 获取，当前主要版本为 4.0。

有了这个，我们可以写：

var blogs = await context.Blogs
    .Where(blog => blog.Url.Contains("dotnet"));

var blogsByNum=await blogs.AsAsyncEnumerable()   individual rows asynchronously
                          .ToLookupAsync(blog=>blog.BlobNumber);

或者

var blogsByNum=await blogs.AsAsyncEnumerable()   
                          .GroupBy(blog=>blog.BlobNumber)
                          .Select(b=>b)
                          .ToListAsync();