在 EF 的 DbContext.SaveChangesAsync() 中引发异步事件

Question

我在 ASP.NET Core 环境中使用 EF Core。我的上下文根据请求在我的 DI 容器中注册。

我需要在上下文的SaveChanges() 或SaveChangesAsync() 之前执行额外的工作，例如验证、审核、调度通知等。其中一些工作是同步的，一些是异步的。

所以我想引发一个同步或异步事件以允许侦听器做额外的工作，阻塞直到它们完成（！），然后调用DbContext 基类来实际保存。

public class MyContext : DbContext
{

  // sync: ------------------------------

  // define sync event handler
  public event EventHandler<EventArgs> SavingChanges;

  // sync save
  public override int SaveChanges(bool acceptAllChangesOnSuccess)
  {
    // raise event for sync handlers to do work BEFORE the save
    var handler = SavingChanges;
    if (handler != null)
      handler(this, EventArgs.Empty);
    // all work done, now save
    return base.SaveChanges(acceptAllChangesOnSuccess);
  }

  // async: ------------------------------

  // define async event handler
  //public event /* ??? */ SavingChangesAsync;

  // async save
  public override async Task<int> SaveChangesAsync(bool acceptAllChangesOnSuccess, CancellationToken cancellationToken = default(CancellationToken))
  {
    // raise event for async handlers to do work BEFORE the save (block until they are done!)
    //await ???
    // all work done, now save
    return await base.SaveChangesAsync(acceptAllChangesOnSuccess,  cancellationToken);
  }

}

如您所见，SaveChanges() 很容易，但SaveChangesAsync() 该怎么做呢？

Answer 1

所以我想引发一个同步或异步事件以允许侦听器做额外的工作，阻塞直到它们完成（！），然后调用 DbContext 基类来实际保存。

如你所见，SaveChanges() 很简单

不是真的...SaveChanges 不会等待任何异步处理程序完成。一般来说，不建议阻塞异步工作；即使在environments such as ASP.NET Core where you won't deadlock 中，它也会影响您的可扩展性。由于您的 MyContext 允许异步处理程序，因此您可能希望覆盖 SaveChanges 以抛出异常。或者，您可以选择只是阻塞，并希望用户不要过多地使用同步 SaveChanges 的异步处理程序。

关于实现本身，我在blog post on async events 中描述了一些方法。我个人最喜欢的是延迟方法，它看起来像这样（使用我的Nito.AsyncEx.Oop 库）：

public class MyEventArgs: EventArgs, IDeferralSource
{
  internal DeferralManager DeferralManager { get; } = new DeferralManager();
  public IDisposable GetDeferral() => DeferralManager.DeferralSource.GetDeferral();
}

public class MyContext : DbContext
{
  public event EventHandler<MyEventArgs> SavingChanges;

  public override int SaveChanges(bool acceptAllChangesOnSuccess)
  {
    // You must decide to either throw or block here (see above).

    // Example code for blocking.
    var args = new MyEventArgs();
    SavingChanges?.Invoke(this, args);
    args.DeferralManager.WaitForDeferralsAsync().GetAwaiter().GetResult();

    return base.SaveChanges(acceptAllChangesOnSuccess);
  }

  public override async Task<int> SaveChangesAsync(bool acceptAllChangesOnSuccess, CancellationToken cancellationToken = default(CancellationToken))
  {
    var args = new MyEventArgs();
    SavingChanges?.Invoke(this, args);
    await args.DeferralManager.WaitForDeferralsAsync();

    return await base.SaveChangesAsync(acceptAllChangesOnSuccess,  cancellationToken);
  }
}

// Usage (synchronous handler):
myContext.SavingChanges += (sender, e) =>
{
  Thread.Sleep(1000); // Synchronous code
};

// Usage (asynchronous handler):
myContext.SavingChanges += async (sender, e) =>
{
  using (e.GetDeferral())
  {
    await Task.Delay(1000); // Asynchronous code
  }
};

Answer 2

有一种更简单的方法（基于on this）。

声明一个返回Task的多播委托：

namespace MyProject
{
  public delegate Task AsyncEventHandler<TEventArgs>(object sender, TEventArgs e);
}

更新上下文（我只显示异步的东西，因为同步的东西没有改变）：

public class MyContext : DbContext
{

  public event AsyncEventHandler<EventArgs> SavingChangesAsync;

  public override async Task<int> SaveChangesAsync(bool acceptAllChangesOnSuccess, CancellationToken cancellationToken = default(CancellationToken))
  {
    var delegates = SavingChangesAsync;
    if (delegates != null)
    {
      var tasks = delegates
        .GetInvocationList()
        .Select(d => ((AsyncEventHandler<EventArgs>)d)(this, EventArgs.Empty))
        .ToList();
      await Task.WhenAll(tasks);
    }
    return await base.SaveChangesAsync(acceptAllChangesOnSuccess, cancellationToken);
  }

}

调用代码如下：

context.SavingChanges += OnContextSavingChanges;
context.SavingChangesAsync += OnContextSavingChangesAsync;

public void OnContextSavingChanges(object sender, EventArgs e)
{
  someSyncMethod();
}

public async Task OnContextSavingChangesAsync(object sender, EventArgs e)
{
  await someAsyncMethod();
}

我不确定这是否是 100% 安全的方法。异步事件很棘手。我对多个订阅者进行了测试，并且成功了。我的环境是ASP.NET Core，所以不知道其他地方能不能用。

我不知道它与其他解决方案相比如何，或者哪个更好，但这个更简单，对我来说更有意义。

编辑：如果您的处理程序不更改共享状态，这会很好。如果是这样，请参阅上面@stephencleary 提供的更强大的方法

Answer 3

我建议修改这个async event handler

public AsyncEvent SavingChangesAsync;

用法

  // async save
  public override async Task<int> SaveChangesAsync(bool acceptAllChangesOnSuccess, CancellationToken cancellationToken = default(CancellationToken))
  {
    await SavingChangesAsync?.InvokeAsync(cancellationToken);
    return await base.SaveChangesAsync(acceptAllChangesOnSuccess,  cancellationToken);
  }

在哪里

public class AsyncEvent
{
    private readonly List<Func<CancellationToken, Task>> invocationList;
    private readonly object locker;

    private AsyncEvent()
    {
        invocationList = new List<Func<CancellationToken, Task>>();
        locker = new object();
    }

    public static AsyncEvent operator +(
        AsyncEvent e, Func<CancellationToken, Task> callback)
    {
        if (callback == null) throw new NullReferenceException("callback is null");

        //Note: Thread safety issue- if two threads register to the same event (on the first time, i.e when it is null)
        //they could get a different instance, so whoever was first will be overridden.
        //A solution for that would be to switch to a public constructor and use it, but then we'll 'lose' the similar syntax to c# events             
        if (e == null) e = new AsyncEvent();

        lock (e.locker)
        {
            e.invocationList.Add(callback);
        }
        return e;
    }

    public static AsyncEvent operator -(
        AsyncEvent e, Func<CancellationToken, Task> callback)
    {
        if (callback == null) throw new NullReferenceException("callback is null");
        if (e == null) return null;

        lock (e.locker)
        {
            e.invocationList.Remove(callback);
        }
        return e;
    }

    public async Task InvokeAsync(CancellationToken cancellation)
    {
        List<Func<CancellationToken, Task>> tmpInvocationList;
        lock (locker)
        {
            tmpInvocationList = new List<Func<CancellationToken, Task>>(invocationList);
        }

        foreach (var callback in tmpInvocationList)
        {
            //Assuming we want a serial invocation, for a parallel invocation we can use Task.WhenAll instead
            await callback(cancellation);
        }
    }
}