Microsoft.EntityFrameworkCore.EntityFrameworkQueryableExtensions.ForEachAsync<T>() 的意外行为

Question

这里是重现的步骤。下面的程序使用 .Net Core 控制台应用程序和 EF Core 将 10,000 行从一个 SQL table 复制到另一个。该程序分 100 批插入记录，并且（这很重要！）它为每次插入创建一个新的 DbContext 实例。

1) 创建SQL服务器数据库，"Froms"和"Tos" tables:

create table Froms (
    Id int identity(1, 1) not null,
    Guid [uniqueidentifier] not null,

    constraint [PK_Froms] primary key clustered (Id asc)
)
go

create table Tos (
    Id int not null,
    Guid [uniqueidentifier] not null,

    constraint [PK_Tos] primary key clustered (Id asc)
)
go

2) 填充 "Froms" table:

set nocount on
declare @i int = 0

while @i < 10000
begin
    insert Froms (Guid)
    values (newid())

    set @i += 1
end
go

3) 创建名为 TestForEachAsync 的 .Net Core 控制台应用程序项目。将 C# 版本更改为 7.1 或更高版本（async Main 需要）。添加 Microsoft.EntityFrameworkCore.SqlServer nuget 包。

4) 创建类:

数据库实体

using System;

namespace TestForEachAsync
{
    public class From
    {
        public int Id { get; set; }
        public Guid Guid { get; set; }
    }
}

using System;

namespace TestForEachAsync
{
    public class To
    {
        public int Id { get; set; }
        public Guid Guid { get; set; }
    }
}

DbContext

using Microsoft.EntityFrameworkCore;

namespace TestForEachAsync
{
    public class Context : DbContext
    {
        public DbSet<From> Froms { get; set; }
        public DbSet<To> Tos { get; set; }

        protected override void OnConfiguring(DbContextOptionsBuilder optionsBuilder)
        {
            optionsBuilder.UseSqlServer("YOUR_CONNECTION_STRING");
        }
    }
}

主要

using System;
using System.Linq;
using System.Threading.Tasks;
using Microsoft.EntityFrameworkCore;

namespace TestForEachAsync
{
    internal class Program
    {
        private static async Task Main(string[] args)
        {
            //Get the "froms"
            var selectContext = new Context();
            var froms = selectContext.Froms.Select(f => new { f.Id, f.Guid });

            int count = 0;
            Task<int> saveChangesTask = null;
            Context insertContext = new Context();
            Context prevInsertContext = null;

            //Iterate through "froms"
            await froms.ForEachAsync(
                async f =>
                {
                    //Add instace of "to" to the context
                    var to = new To { Id = f.Id, Guid = f.Guid };
                    await insertContext.Tos.AddAsync(to);
                    count++;

                    //If another 100 of "to"s has been added to the context...
                    if (count % 100 == 0)
                    {
                        //Wait for the previous 100 "to"s to finish saving to the database
                        if (saveChangesTask != null)
                        {
                            await saveChangesTask;
                        }

                        //Start saving the next 100 "to"s
                        saveChangesTask = insertContext.SaveChangesAsync();

                        //Dispose of the context that was used to save previous 100 "to"s
                        prevInsertContext?.Dispose();

                        //Reassign the context used to save the current 100 "to"s to a "prev" variable,
                        //and set context variable to the new Context instance.
                        prevInsertContext = insertContext;
                        insertContext = new Context();
                    }
                }
            );

            //Wait for second last 100 "to"s to finish saving to the database
            if (saveChangesTask != null)
            {
                await saveChangesTask;
            }

            //Save the last 100 "to"s to the database
            await insertContext.SaveChangesAsync();
            insertContext.Dispose();

            Console.WriteLine("Done");
            Console.ReadKey();
        }
    }
}

5) 运行应用程序 - 您遇到异常 The connection does not support MultipleActiveResultSets。看起来 insertContext 上正在启动多个操作，但我不明白为什么。

6) 我找到了两种解决问题的方法：

将await froms.ForEachAsync(...)循环替换为"normal"循环foreach (var f in froms) {...}，或
在异步循环中，将 await saveChangesTask; 替换为 saveChangesTask.Wait();

但是有人能解释一下为什么原始代码不能像我预期的那样工作吗？

注意：如果您多次运行应用程序，请不要忘记在每个运行之前添加运行 "Tos" table。

Answer 1

您陷入了将异步 lambda 传递给期望委托 returns void（在此特定情况下为 Action<T>）的方法的典型陷阱，如 Stephen Toub in Potential pitfalls to avoid when passing around async lambdas 所述。它实际上等同于使用 async void 和它的陷阱，因为你的异步代码根本不是 await-ed，因此破坏了它的内部逻辑。

解决方案是像往常一样接受 Func<T, Task> 而不是 Action<T> 的特殊重载。可能它应该由 EF Core 提供（您可以考虑为此发布一个请求），但现在您可以自己用这样的东西实现它：

using System;
using System.Collections.Generic;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
using Microsoft.EntityFrameworkCore.Extensions.Internal;

namespace Microsoft.EntityFrameworkCore
{
    public static class AsyncExtensions
    {
        public static Task ForEachAsync<T>(this IQueryable<T> source, Func<T, Task> action, CancellationToken cancellationToken = default) =>
            source.AsAsyncEnumerable().ForEachAsync(action, cancellationToken);

        public static async Task ForEachAsync<T>(this IAsyncEnumerable<T> source, Func<T, Task> action, CancellationToken cancellationToken = default)
        {
            using (var asyncEnumerator = source.GetEnumerator())
                while (await asyncEnumerator.MoveNext(cancellationToken))
                    await action(asyncEnumerator.Current);
        }
    }
}

基本上是 EF Core implementation 添加了 action 的 await。

执行此操作后，您的代码将解析为此方法，一切都应按预期工作。

Microsoft.EntityFrameworkCore.EntityFrameworkQueryableExtensions.ForEachAsync<T>() 的意外行为

Unexpected behaviour with Microsoft.EntityFrameworkCore.EntityFrameworkQueryableExtensions.ForEachAsync<T>()

c#

entity-framework-core

ef-core-2.0

ef-core-2.1