如何在 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 您想要进行分组客户端。在 GitHub和Microsoft 文档中对此进行了讨论。

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 操作，事情会容易得多。这个扩展命名空间正好提供了这一点。它由 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();